Performed retries on apply/delete timeout (#208)

* Perform retry when ns4kafka timeout on schema registry/kconnect

* Perform retry when ns4kafka timeout on schema registry/kconnect

* Update condition

* Update config

Author: Loïc GREFFIER

cli/src/main/java/com/michelin/ns4kafka/cli/client/NamespacedResourceClient.java

@@ -1,10 +1,12 @@
 package com.michelin.ns4kafka.cli.client;
 
+import com.michelin.ns4kafka.cli.client.predicates.RetryTimeoutPredicate;
 import com.michelin.ns4kafka.cli.models.Resource;
 import com.michelin.ns4kafka.cli.models.SchemaCompatibility;
 import io.micronaut.http.HttpResponse;
 import io.micronaut.http.annotation.*;
 import io.micronaut.http.client.annotation.Client;
+import io.micronaut.retry.annotation.Retryable;
 
 import java.util.List;
 import java.util.Map;
@@ -21,6 +23,10 @@ public interface NamespacedResourceClient {
      * @return The delete response
      */
     @Delete("{namespace}/{kind}/{resourceName}{?dryrun}")
+    @Retryable(delay = "${kafkactl.retry.delete.delay}",
+            attempts = "${kafkactl.retry.delete.attempt}",
+            multiplier = "${kafkactl.retry.delete.multiplier}",
+            predicate = RetryTimeoutPredicate.class)
     HttpResponse<Void> delete(
             String namespace,
             String kind,
@@ -38,11 +44,15 @@ HttpResponse<Void> delete(
      * @return The resource
      */
     @Post("{namespace}/{kind}{?dryrun}")
+    @Retryable(delay = "${kafkactl.retry.apply.delay}",
+            attempts = "${kafkactl.retry.apply.attempt}",
+            multiplier = "${kafkactl.retry.apply.multiplier}",
+            predicate = RetryTimeoutPredicate.class)
     HttpResponse<Resource> apply(
             String namespace,
             String kind,
             @Header("Authorization") String token,
-            @Body Resource json,
+            @Body Resource resource,
             @QueryValue boolean dryrun);
 
     /**

cli/src/main/java/com/michelin/ns4kafka/cli/client/predicates/RetryTimeoutPredicate.java

@@ -0,0 +1,30 @@
+package com.michelin.ns4kafka.cli.client.predicates;
+
+import com.michelin.ns4kafka.cli.models.Status;
+import io.micronaut.http.client.exceptions.HttpClientResponseException;
+import io.micronaut.retry.annotation.RetryPredicate;
+import io.netty.handler.codec.http.HttpResponseStatus;
+
+import java.util.Optional;
+
+public class RetryTimeoutPredicate implements RetryPredicate {
+    /**
+     * Detect when Ns4Kafka return a timeout exception (e.g. when deploying schemas, connectors)
+     * @param throwable the input argument
+     * @return true if a retry is necessary, false otherwise
+     */
+    @Override
+    public boolean test(Throwable throwable) {
+        if (throwable instanceof HttpClientResponseException) {
+            Optional<Status> statusOptional = ((HttpClientResponseException) throwable).getResponse().getBody(Status.class);
+            if (statusOptional.isPresent() && statusOptional.get().getDetails() != null
+                    && !statusOptional.get().getDetails().getCauses().isEmpty()
+                    && HttpResponseStatus.INTERNAL_SERVER_ERROR.code() == statusOptional.get().getCode()
+                    && statusOptional.get().getDetails().getCauses().stream().anyMatch(cause -> cause.contains("Read Timeout"))) {
+                System.out.println("Read timeout... retrying...");
+                return true;
+            }
+        }
+        return false;
+    }
+}

cli/src/main/java/com/michelin/ns4kafka/cli/services/ResourceService.java

@@ -90,7 +90,7 @@ public HttpResponse<Resource> apply(ApiResource apiResource, String namespace, R
     public boolean delete(ApiResource apiResource, String namespace, String resource, boolean dryRun) {
         try {
             if (apiResource.isNamespaced()) {
-                HttpResponse response = namespacedClient.delete(namespace, apiResource.getPath(), resource, loginService.getAuthorization(), dryRun);
+                HttpResponse<Void> response = namespacedClient.delete(namespace, apiResource.getPath(), resource, loginService.getAuthorization(), dryRun);
                 if(response.getStatus() != HttpStatus.NO_CONTENT){
                     throw new HttpClientResponseException("Resource not Found", response);
                 }

cli/src/main/resources/application.yml

@@ -84,6 +84,15 @@ kafkactl:
       - "NAMESPACE:/spec/namespace"
       - "API:/spec/api"
       - "TOKEN:/spec/token"
+  retry:
+    apply:
+      delay: "2s"
+      attempt: "10"
+      multiplier: "1.0"
+    delete:
+      delay: "2s"
+      attempt: "10"
+      multiplier: "1.0"
   contexts:
     -
 

cli/src/main/resources/logback.xml

@@ -13,4 +13,4 @@
         <!-- Use application.yml to configure loggers-->
         <appender-ref ref="STDOUT" />
     </root>
-</configuration>
+</configuration>