implemented healthcheck and producer /consumer retries

Author: danrusei

IMPLEMENTATION_PLAN.md

@@ -45,6 +45,12 @@
       <artifactId>junit-jupiter</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.slf4j</groupId>
+      <artifactId>slf4j-simple</artifactId>
+      <version>${slf4j.version}</version>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>
@@ -55,4 +61,30 @@
       </plugin>
     </plugins>
   </build>
+
+  <profiles>
+    <profile>
+      <id>integration-tests</id>
+      <build>
+        <plugins>
+          <plugin>
+            <groupId>org.apache.maven.plugins</groupId>
+            <artifactId>maven-failsafe-plugin</artifactId>
+            <version>${maven.surefire.plugin.version}</version>
+            <configuration>
+              <useModulePath>false</useModulePath>
+            </configuration>
+            <executions>
+              <execution>
+                <goals>
+                  <goal>integration-test</goal>
+                  <goal>verify</goal>
+                </goals>
+              </execution>
+            </executions>
+          </plugin>
+        </plugins>
+      </build>
+    </profile>
+  </profiles>
 </project>

danube-client/pom.xml

@@ -2,6 +2,7 @@
 
 import com.danubemessaging.client.errors.DanubeClientException;
 import com.danubemessaging.client.internal.consumer.TopicConsumer;
+import com.danubemessaging.client.internal.retry.RetryManager;
 import com.danubemessaging.client.model.StreamMessage;
 import java.util.ArrayList;
 import java.util.List;
@@ -59,6 +60,10 @@ public synchronized void subscribe() {
         List<String> partitions = client.lookupService().topicPartitions(client.serviceUri(), options.topic());
         List<String> targets = partitions.isEmpty() ? List.of(options.topic()) : partitions;
 
+        RetryManager retryManager = hasCustomRetryOptions()
+                ? new RetryManager(options.maxRetries(), options.baseBackoffMs(), options.maxBackoffMs())
+                : client.retryManager();
+
         List<TopicConsumer> createdConsumers = new ArrayList<>(targets.size());
         try {
             for (int i = 0; i < targets.size(); i++) {
@@ -73,7 +78,7 @@ public synchronized void subscribe() {
                         client.lookupService(),
                         client.authService(),
                         client.healthCheckService(),
-                        client.retryManager(),
+                        retryManager,
                         options,
                         partitionTopic,
                         partitionConsumerName);
@@ -194,6 +199,10 @@ private void notifyFatalReceiveError(TopicConsumer topicConsumer, RuntimeExcepti
         }
     }
 
+    private boolean hasCustomRetryOptions() {
+        return options.maxRetries() > 0 || options.baseBackoffMs() > 0 || options.maxBackoffMs() > 0;
+    }
+
     private void ensureOpen() {
         if (lifecycleState.get() == LifecycleState.CLOSED) {
             throw new DanubeClientException("Consumer is closed");