Retry pattern for a specific topic support (#274)

* retry pattern for specific topic and docs

* removed commented cod

* review comments and updated docs

* removed else part code improve

* Update guide-kafka.asciidoc

* retry pattern for specific topic and docs

* removed commented cod

* review comments and updated docs

* removed else part code improve

* Update guide-kafka.asciidoc

* modified value to key for loggers

Co-authored-by: Simon Spielmann <[email protected]>

Author: ravicm2

documentation/guide-kafka.asciidoc

@@ -114,7 +114,7 @@ Example:
   @Inject
   private MessageSender messageSender;
   private ProducerRecord<K,V> producerRecord;
-  
+
   public void sendMessageToKafka(){
   producerRecord=new ProducerRecord<>("topic-name","message");
   messageSender.sendMessage(this.producerRecord);
@@ -225,24 +225,63 @@ It works as follows:
 === Retry configuration and naming convention of redelivery topics.
 The following properties should be added in the `application.properties` or `application.yml` file.
 
+The retry pattern in devon4j-kafka will perform for specific topic of a message. So its mandatory to specify the properties for each topic. Below properties are example,
+
 [source,properties]
 -----
-# Retry back off policy properties
-messaging.retry.default.back-off-policy.retryReEnqueueDelay=1000
-messaging.retry.default.back-off-policy.retryDelay=600000
-messaging.retry.default.back-off-policy.retryDelayMultiplier=1.0
-messaging.retry.default.back-off-policy.retryMaxDelay=600000
-messaging.retry.default.back-off-policy.retryCount=2
-
-# default retry policy properties
-messaging.retry.default.retry-policy.retryPeriod=1800
-messaging.retry.default.retry-policy.retryableExceptions=<Class names of exceptions for which a retry should be performed>
-messaging.retry.default.retry-policy.retryableExceptionsTraverseCauses=true
+# Back off policy properties for employeeapp-employee-v1-delete
+messaging.retry.back-off-policy.retryReEnqueueDelay.employeeapp-employee-v1-delete=1000
+messaging.retry.back-off-policy.retryDelay.employeeapp-employee-v1-delete=600000
+messaging.retry.back-off-policy.retryDelayMultiplier.employeeapp-employee-v1-delete=1.0
+messaging.retry.back-off-policy.retryMaxDelay.employeeapp-employee-v1-delete=600000
+messaging.retry.back-off-policy.retryCount.employeeapp-employee-v1-delete=2
+
+# Retry policy properties for employeeapp-employee-v1-delete
+messaging.retry.retry-policy.retryPeriod.employeeapp-employee-v1-delete=1800
+messaging.retry.retry-policy.retryableExceptions.employeeapp-employee-v1-delete=<Class names of exceptions for which a retry should be performed>
+messaging.retry.retry-policy.retryableExceptionsTraverseCauses.employeeapp-employee-v1-delete=true
+
+# Back off policy properties for employeeapp-employee-v1-add
+messaging.retry.back-off-policy.retryReEnqueueDelay.employeeapp-employee-v1-add=1000
+messaging.retry.back-off-policy.retryDelay.employeeapp-employee-v1-add=600000
+messaging.retry.back-off-policy.retryDelayMultiplier.employeeapp-employee-v1-add=2.0
+messaging.retry.back-off-policy.retryMaxDelay.employeeapp-employee-v1-add=600000
+messaging.retry.back-off-policy.retryCount.employeeapp-employee-v1-add=4
+
+# Retry policy properties for employeeapp-employee-v1-add
+messaging.retry.retry-policy.retryPeriod.employeeapp-employee-v1-add=3000
+messaging.retry.retry-policy.retryableExceptions.employeeapp-employee-v1-add=<Class names of exceptions for which a retry should be performed>
+messaging.retry.retry-policy.retryableExceptionsTraverseCauses.employeeapp-employee-v1-add=true
 -----
 
+If you notice the above properties, the `retry-policy` and `back-off policy` properties are repeated twice as i have 2 topics for the retry to be performed with different level of values. The topic name should be added at the last of attribute.
+
+So, the retry will be performed for each topic according to their configuration values.
+
+If you want to provide same/default values for all the topics, then its required to add `default` in the place of topic on the above properties example.
+
+For example,
+
+[source,properties]
+-----
+# Default back off policy properties
+messaging.retry.back-off-policy.retryReEnqueueDelay.default=1000
+messaging.retry.back-off-policy.retryDelay.default=600000
+messaging.retry.back-off-policy.retryDelayMultiplier.default=1.0
+messaging.retry.back-off-policy.retryMaxDelay.default=600000
+messaging.retry.back-off-policy.retryCount.default=2
+
+# Default retry policy properties
+messaging.retry.retry-policy.retryPeriod.default=1800
+messaging.retry.retry-policy.retryableExceptions.default=<Class names of exceptions for which a retry should be performed>
+messaging.retry.retry-policy.retryableExceptionsTraverseCauses.default=true
+-----
+
+By giving properties like above , the same values will be passed for all the topics and the way of processing retry for all the topics are same.
+
 All these above property values are mapped to the classes `DefaultBackOffPolicyProperties.java` and `DefaultRetryPolicyProperties.java` and configured by the class `MessageDefaultRetryConfig.java`.
 
-The MessageRetryContext in devon kafka is used to perform the retry pattern with the properties from DefaultBackOffPolicyProperties and DefaultRetryPolicyProperties. 
+The MessageRetryContext in devon kafka is used to perform the retry pattern with the properties from DefaultBackOffPolicyProperties and DefaultRetryPolicyProperties.
 
 The 2 main properties of MessageRetryContext is nextRetry and retryUntil which is a `Instant` date format and it is calculated internally using the properties given in DefaultBackOffPolicyProperties and DefaultRetryPolicyProperties.
 
@@ -259,7 +298,7 @@ Devon4-kafka enqueues a new message for each retry attempt. It is very important
 
 === Handling retry finally failed
 
-Per default when the retry fails with final attempt  we just log the message and delete the payload of ProducerRecord which comes to proceed the retry pattern. 
+Per default when the retry fails with final attempt  we just log the message and delete the payload of ProducerRecord which comes to proceed the retry pattern.
 
 You can change this behavior by providing the implementation class for the interface `MessageRetryHandler.java`
 which has two method `retryTimeout` and `retryFailedFinal`.

modules/kafka/src/main/java/com/devonfw/module/kafka/common/messaging/logging/impl/ProducerLoggingListener.java

@@ -1,5 +1,7 @@
 package com.devonfw.module.kafka.common.messaging.logging.impl;
 
+import java.util.Optional;
+
 import org.apache.kafka.clients.producer.RecordMetadata;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -32,10 +34,8 @@ public ProducerLoggingListener(MessageLoggingSupport loggingSupport) {
   @Override
   public void onSuccess(String topic, Integer partition, Object key, Object value, RecordMetadata recordMetadata) {
 
-    String messageKey = "<no message key>";
-    if (key != null) {
-      key = key.toString();
-    }
+    String messageKey = (String) Optional.ofNullable(key).orElse("<no message key>");
+
     if (recordMetadata != null) {
       this.loggingSupport.logMessageSent(LOG, messageKey, recordMetadata.topic(), recordMetadata.partition(),
           recordMetadata.offset());

modules/kafka/src/main/java/com/devonfw/module/kafka/common/messaging/retry/api/client/MessageBackOffPolicy.java

@@ -16,14 +16,17 @@ public interface MessageBackOffPolicy {
    * @param currentRetryCount the count of attempted retry count.
    *
    * @param retryUntilTimestamp the retry until time stamp which requires a Instant in String format.
+   * @param topic the topic
    * @return the Instant
    */
-  Instant getNextRetryTimestamp(long currentRetryCount, String retryUntilTimestamp);
+  Instant getNextRetryTimestamp(long currentRetryCount, String retryUntilTimestamp, String topic);
 
   /**
    * This method is used to make the thread to sleep for {@link DefaultBackOffPolicyProperties#getRetryReEnqueueDelay()}
    * seconds.
+   * 
+   * @param topic the topic
    */
-  void sleepBeforeReEnqueue();
+  void sleepBeforeReEnqueue(String topic);
 
 }

modules/kafka/src/main/java/com/devonfw/module/kafka/common/messaging/retry/api/client/MessageRetryPolicy.java

@@ -38,8 +38,10 @@ public interface MessageRetryPolicy<K, V> {
   /**
    * This method is used to return the number of retry count.
    *
+   * @param topic the topic
+   *
    * @return the retry count as long.
    */
-  long getRetryCount();
+  long getRetryCount(String topic);
 
 }

modules/kafka/src/main/java/com/devonfw/module/kafka/common/messaging/retry/api/config/DefaultBackOffPolicyProperties.java

@@ -1,5 +1,8 @@
 package com.devonfw.module.kafka.common.messaging.retry.api.config;
 
+import java.util.HashMap;
+import java.util.Map;
+
 import com.devonfw.module.kafka.common.messaging.retry.api.client.MessageBackOffPolicy;
 
 /**
@@ -8,92 +11,124 @@
  */
 public class DefaultBackOffPolicyProperties {
 
-  private long retryDelay = 60000;
+  private Map<String, Long> retryDelay = new HashMap<>();
+
+  private Map<String, Double> retryDelayMultiplier = new HashMap<>();
+
+  private Map<String, Long> retryMaxDelay = new HashMap<>();
 
-  private double retryDelayMultiplier = 1.0;
+  private Map<String, Long> retryReEnqueueDelay = new HashMap<>();
 
-  private long retryMaxDelay = 60000;
+  private long retryDelayDefault = 60000;
 
-  private long retryReEnqueueDelay = 1000;
+  private double retryDelayMultiplierDefault = 1.0;
+
+  private long retryMaxDelayDefault = 60000;
+
+  private long retryReEnqueueDelayDefault = 1000;
 
   /**
    * The retry enque delay to send again the message to kafka. By default 1000.
    *
    * @return long
    */
-  public long getRetryReEnqueueDelay() {
+  public long getRetryReEnqueueDelayDefault() {
 
-    return this.retryReEnqueueDelay;
+    return this.retryReEnqueueDelayDefault;
   }
 
   /**
-   * Set the retry enque delay for {@link #getRetryReEnqueueDelay()}
+   * The retry delay.
    *
-   * @param retryReEnqueueDelay the enque delay.
+   * @return long.
    */
-  public void setRetryReEnqueueDelay(long retryReEnqueueDelay) {
+  public long getRetryDelayDefault() {
 
-    this.retryReEnqueueDelay = retryReEnqueueDelay;
+    return this.retryDelayDefault;
   }
 
   /**
-   * The retry delay.
+   * The retry delay multiplier
    *
-   * @return long.
+   * @return double.
    */
-  public long getRetryDelay() {
+  public double getRetryDelayMultiplierDefault() {
 
-    return this.retryDelay;
+    return this.retryDelayMultiplierDefault;
   }
 
   /**
-   * Set the retry delay for {@link #getRetryDelay()}. By default 1000
+   * The retry max delay.
    *
-   * @param retryDelay the retry delay.
+   * @return the long.
+   */
+  public long getRetryMaxDelayDefault() {
+
+    return this.retryMaxDelayDefault;
+  }
+
+  /**
+   * @return retryDelay
    */
-  public void setRetryDelay(long retryDelay) {
+  public Map<String, Long> getRetryDelay() {
+
+    return this.retryDelay;
+  }
+
+  /**
+   * @param retryDelay new value of {@link #getRetryDelay}.
+   */
+  public void setRetryDelay(Map<String, Long> retryDelay) {
 
     this.retryDelay = retryDelay;
   }
 
   /**
-   * The retry delay multiplier
-   *
-   * @return double.
+   * @return retryDelayMultiplier
    */
-  public double getRetryDelayMultiplier() {
+  public Map<String, Double> getRetryDelayMultiplier() {
 
     return this.retryDelayMultiplier;
   }
 
   /**
-   * Set the retry delay multiplier for {@link #getRetryDelayMultiplier()}. By default 1.0
-   *
-   * @param retryDelayMultiplier the retry delay multiplier.
+   * @param retryDelayMultiplier new value of {@link #getRetryDelayMultiplier}.
    */
-  public void setRetryDelayMultiplier(double retryDelayMultiplier) {
+  public void setRetryDelayMultiplier(Map<String, Double> retryDelayMultiplier) {
 
     this.retryDelayMultiplier = retryDelayMultiplier;
   }
 
   /**
-   * The retry max delay.
-   *
-   * @return the long.
+   * @return retryMaxDelay
    */
-  public long getRetryMaxDelay() {
+  public Map<String, Long> getRetryMaxDelay() {
 
     return this.retryMaxDelay;
   }
 
   /**
-   * Set the retry max delay for {@link #getRetryMaxDelay()}. by default 60000.
-   *
-   * @param retryMaxDelay the retry max delay.
+   * @param retryMaxDelay new value of {@link #getRetryMaxDelay}.
    */
-  public void setRetryMaxDelay(long retryMaxDelay) {
+  public void setRetryMaxDelay(Map<String, Long> retryMaxDelay) {
 
     this.retryMaxDelay = retryMaxDelay;
   }
 
+  /**
+   * @return retryReEnqueueDelay
+   */
+  public Map<String, Long> getRetryReEnqueueDelay() {
+
+    return this.retryReEnqueueDelay;
+  }
+
+  /**
+   * @param retryReEnqueueDelay new value of {@link #getRetryReEnqueueDelay}.
+   */
+  public void setRetryReEnqueueDelay(Map<String, Long> retryReEnqueueDelay) {
+
+    this.retryReEnqueueDelay = retryReEnqueueDelay;
+  }
+
 }