fix(cloud): use RetryUtils to retry (#204)

Author: fdelbrayelle

src/main/java/io/kestra/plugin/dbt/cloud/AbstractDbtCloud.java

@@ -12,7 +12,9 @@
 import io.kestra.core.http.client.configurations.HttpConfiguration;
 import io.kestra.core.models.property.Property;
 import io.kestra.core.models.tasks.Task;
+import io.kestra.core.models.tasks.retrys.Exponential;
 import io.kestra.core.runners.RunContext;
+import io.kestra.core.utils.RetryUtils;
 import io.swagger.v3.oas.annotations.media.Schema;
 import jakarta.validation.constraints.NotNull;
 import lombok.*;
@@ -21,8 +23,6 @@
 import java.io.IOException;
 import java.time.Duration;
 
-import static org.awaitility.Awaitility.await;
-
 @SuperBuilder
 @ToString
 @EqualsAndHashCode
@@ -63,15 +63,6 @@ public abstract class AbstractDbtCloud extends Task {
     @Builder.Default
     Property<Long> initialDelayMs = Property.ofValue(1000L);
 
-    /**
-     * Perform an HTTP request using Kestra HttpClient with retry logic.
-     *
-     * @param runContext     The Kestra execution context.
-     * @param requestBuilder The prepared HTTP request builder.
-     * @param responseType   The expected response type.
-     * @param <RES>          The response class.
-     * @return HttpResponse of type RES.
-     */
     protected <RES> HttpResponse<RES> request(
         RunContext runContext,
         HttpRequest.HttpRequestBuilder requestBuilder,
@@ -83,55 +74,33 @@ protected <RES> HttpResponse<RES> request(
             .addHeader("Content-Type", "application/json")
             .build();
 
-        int rMaxRetries = runContext.render(this.maxRetries).as(Integer.class).orElse(3);
-        long rInitialDelay = runContext.render(this.initialDelayMs).as(Long.class).orElse(1000L);
-
-        int attempt = 0;
+        var rMaxRetries = runContext.render(this.maxRetries).as(Integer.class).orElse(3);
+        var rInitialDelay = runContext.render(this.initialDelayMs).as(Long.class).orElse(1000L);
 
         try (var client = new HttpClient(runContext, options)) {
-            while (true) {
-                try {
-                    HttpResponse<String> response = client.request(request, String.class);
-
-                    RES parsedResponse = MAPPER.readValue(response.getBody(), responseType);
+            return new RetryUtils().<HttpResponse<RES>, HttpClientException>of(
+                Exponential.builder()
+                    .delayFactor(2.0)
+                    .interval(Duration.ofMillis(rInitialDelay))
+                    .maxInterval(Duration.ofSeconds(30))
+                    .maxAttempts(rMaxRetries)
+                    .build()
+            ).run(
+                (res, throwable) -> throwable instanceof HttpClientResponseException ex &&
+                    (ex.getResponse().getStatus().getCode() == 502 ||
+                        ex.getResponse().getStatus().getCode() == 503 ||
+                        ex.getResponse().getStatus().getCode() == 504),
+                () -> {
+                    var response = client.request(request, String.class);
+                    var parsedResponse = MAPPER.readValue(response.getBody(), responseType);
                     return HttpResponse.<RES>builder()
                         .request(request)
                         .body(parsedResponse)
                         .headers(response.getHeaders())
                         .status(response.getStatus())
                         .build();
-
-                } catch (HttpClientException e) {
-                    int statusCode = extractStatusCode(e);
-
-                    if ((statusCode == 502 || statusCode == 503 || statusCode == 504) && attempt < rMaxRetries) {
-                        long backoff = (long) (rInitialDelay * Math.pow(2, attempt));
-                        runContext.logger().warn(
-                            "Request failed with status {}. Retrying in {} ms (attempt {}/{})",
-                            statusCode, backoff, attempt + 1, rMaxRetries
-                        );
-
-                        await()
-                            .pollDelay(Duration.ofMillis(backoff))
-                            .atMost(Duration.ofMillis(backoff + 50))
-                            .until(() -> true);
-
-                        attempt++;
-                        continue;
-                    }
-
-                    throw e;
-                } catch (IOException e) {
-                    throw new RuntimeException("Error executing HTTP request", e);
                 }
-            }
-        }
-    }
-
-    private int extractStatusCode(HttpClientException e) {
-        if (e instanceof HttpClientResponseException ex) {
-            return ex.getResponse().getStatus().getCode();
+            );
         }
-        return -1;
     }
 }

src/test/java/io/kestra/plugin/dbt/cloud/CheckStatusRetryTest.java

@@ -8,15 +8,15 @@
 import io.kestra.core.models.property.Property;
 import io.kestra.core.runners.RunContextFactory;
 import io.kestra.core.utils.IdUtils;
+import io.kestra.core.utils.RetryUtils;
 import jakarta.inject.Inject;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
 
 import java.net.URI;
 import java.util.Map;
 
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.*;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.*;
@@ -86,12 +86,6 @@ void shouldFailAfterMaxRetries() throws Exception {
                         .status(HttpResponse.Status.builder().code(502).build())
                         .build()
                 ))
-                .thenThrow(new HttpClientResponseException(
-                    "Bad Gateway",
-                    HttpResponse.<String>builder()
-                        .status(HttpResponse.Status.builder().code(502).build())
-                        .build()
-                ))
                 .thenThrow(new HttpClientResponseException(
                     "Bad Gateway",
                     HttpResponse.<String>builder()
@@ -109,12 +103,16 @@ void shouldFailAfterMaxRetries() throws Exception {
                 .initialDelayMs(Property.ofValue(100L))
                 .build();
 
-            assertThrows(HttpClientResponseException.class,
+            var ex = assertThrows(RetryUtils.RetryFailed.class,
                 () -> task.request(runContext, requestBuilder, Map.class)
             );
 
+            assertInstanceOf(HttpClientResponseException.class, ex.getCause());
+            var cause = (HttpClientResponseException) ex.getCause();
+            assertEquals(502, cause.getResponse().getStatus().getCode());
+
             var mockClient = mocked.constructed().getFirst();
-            verify(mockClient, times(3)).request(any(HttpRequest.class), eq(String.class));
+            verify(mockClient, times(2)).request(any(HttpRequest.class), eq(String.class));
         }
     }
 }