Improve retry mechanism in dmaap-client.

Return last exception instead of timeout exception when retry exhausted.
Handle no connection exception when sending requests to dmaap-mr.

Issue-ID: DCAEGEN2-1483
Signed-off-by: tkogut <tomasz.kogut@nokia.com>
Change-Id: Ibe318fa349b79999a5c8054e04e72e444a42ea78

4 files changed, 55 insertions, 67 deletions

diff --git a/rest-services/http-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/RxHttpClient.java b/rest-services/http-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/RxHttpClient.java
index 76bde27e..d25d7469 100644
--- a/rest-services/http-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/RxHttpClient.java
+++ b/rest-services/http-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/RxHttpClient.java
@@ -25,6 +25,7 @@ import io.vavr.collection.HashSet;
 import io.vavr.collection.Stream;
 import io.vavr.control.Option;
 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.config.RetryConfig;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.exceptions.RetryableException;
 import org.onap.dcaegen2.services.sdk.rest.services.model.logging.RequestDiagnosticContext;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -83,12 +84,17 @@ public class RxHttpClient {
     }
 
     private Mono<HttpResponse> mapResponse(String url, HttpResponseStatus status, reactor.netty.ByteBufMono content) {
-        if (shouldRetry(status.code())) {
-            return Mono.error(new RetryConfig.RetryableException());
-        }
         return content.asByteArray()
                 .defaultIfEmpty(new byte[0])
-                .map(bytes -> new NettyHttpResponse(url, status, bytes));
+                .map(bytes -> new NettyHttpResponse(url, status, bytes))
+                .map(this::validatedResponse);
+    }
+
+    private HttpResponse validatedResponse(HttpResponse response) {
+        if (shouldRetry(response.statusCode())) {
+            throw new RetryableException(response);
+        }
+        return response;
     }
 
     private boolean shouldRetry(int code) {
@@ -149,15 +155,12 @@ public class RxHttpClient {
     }
 
     private RetryBackoffSpec retryConfig(RetryConfig retryConfig, RequestDiagnosticContext context) {
-        RetryBackoffSpec retry = Retry
+        return Retry
                 .fixedDelay(retryConfig.retryCount(), retryConfig.retryInterval())
                 .doBeforeRetry(retrySignal -> context.withSlf4jMdc(
                         LOGGER.isTraceEnabled(), () -> LOGGER.trace("Retry: {}", retrySignal)))
-                .filter(ex -> isRetryable(retryConfig, ex));
-
-        return Option.of(retryConfig.onRetryExhaustedException())
-                .map(ex -> retry.onRetryExhaustedThrow((retryBackoffSpec, retrySignal) -> ex))
-                .getOrElse(retry);
+                .filter(ex -> isRetryable(retryConfig, ex))
+                .onRetryExhaustedThrow((retryBackoffSpec, retrySignal) -> retrySignal.failure());
     }
 
     private boolean isRetryable(RetryConfig retryConfig, Throwable ex) {
diff --git a/rest-services/http-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/config/RetryConfig.java b/rest-services/http-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/config/RetryConfig.java
index a0ae1991..e4584905 100644
--- a/rest-services/http-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/config/RetryConfig.java
+++ b/rest-services/http-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/config/RetryConfig.java
@@ -23,7 +23,7 @@ package org.onap.dcaegen2.services.sdk.rest.services.adapters.http.config;
 import io.vavr.collection.HashSet;
 import io.vavr.collection.Set;
 import org.immutables.value.Value;
-import org.jetbrains.annotations.Nullable;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.exceptions.RetryableException;
 
 import java.time.Duration;
 
@@ -52,8 +52,4 @@ public interface RetryConfig {
         }
         return result;
     }
-
-    @Nullable RuntimeException onRetryExhaustedException();
-
-    class RetryableException extends RuntimeException {}
 }
diff --git a/rest-services/http-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/exceptions/RetryableException.java b/rest-services/http-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/exceptions/RetryableException.java
new file mode 100644
index 00000000..aa48497a
--- /dev/null
+++ b/rest-services/http-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/exceptions/RetryableException.java
@@ -0,0 +1,36 @@
+/*
+ * ============LICENSE_START====================================
+ * DCAEGEN2-SERVICES-SDK
+ * =========================================================
+ * Copyright (C) 2021 Nokia. All rights reserved.
+ * =========================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=====================================
+ */
+
+package org.onap.dcaegen2.services.sdk.rest.services.adapters.http.exceptions;
+
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpResponse;
+
+public class RetryableException extends RuntimeException {
+
+    private final HttpResponse response;
+
+    public RetryableException(HttpResponse response) {
+        this.response = response;
+    }
+
+    public HttpResponse getResponse() {
+        return response;
+    }
+}
diff --git a/rest-services/http-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/RxHttpClientIT.java b/rest-services/http-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/RxHttpClientIT.java
index daf04c6e..8d076e02 100644
--- a/rest-services/http-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/RxHttpClientIT.java
+++ b/rest-services/http-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/RxHttpClientIT.java
@@ -29,6 +29,7 @@ import org.junit.jupiter.api.Test;
 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.config.ImmutableRetryConfig;
 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.config.ImmutableRxHttpClientConfig;
 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.exceptions.HttpException;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.exceptions.RetryableException;
 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.test.DummyHttpServer;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
@@ -214,7 +215,7 @@ class RxHttpClientIT {
     }
 
     @Test
-    void getWithRetryExhaustedExceptionWhenClosedServer() throws Exception {
+    void getWithConnectExceptionWhenClosedServer() throws Exception {
         // given
         REQUEST_COUNTER = new AtomicInteger();
         final HttpRequest httpRequest = requestForClosedServer("/sample-get")
@@ -231,37 +232,13 @@ class RxHttpClientIT {
 
         // then
         StepVerifier.create(response)
-                .expectError(IllegalStateException.class)
+                .expectError(ConnectException.class)
                 .verify(TIMEOUT);
         assertNoServerResponse();
     }
 
     @Test
-    void getWithCustomRetryExhaustedExceptionWhenClosedServer() throws Exception {
-        // given
-        REQUEST_COUNTER = new AtomicInteger();
-        final HttpRequest httpRequest = requestForClosedServer("/sample-get")
-                .method(HttpMethod.GET)
-                .build();
-        final RxHttpClient cut = RxHttpClientFactory.create(ImmutableRxHttpClientConfig.builder()
-                .retryConfig(defaultRetryConfig()
-                        .customRetryableExceptions(HashSet.of(ConnectException.class))
-                        .onRetryExhaustedException(ReadTimeoutException.INSTANCE)
-                        .build())
-                .build());
-
-        // when
-        final Mono<HttpResponse> response = cut.call(httpRequest);
-
-        // then
-        StepVerifier.create(response)
-                .expectError(ReadTimeoutException.class)
-                .verify(TIMEOUT);
-        assertNoServerResponse();
-    }
-
-    @Test
-    void getWithRetryExhaustedExceptionWhen500() throws Exception {
+    void getWithRetryableExceptionWhen500() throws Exception {
         // given
         REQUEST_COUNTER = new AtomicInteger();
         final HttpRequest httpRequest = requestFor("/retry-get-500")
@@ -278,31 +255,7 @@ class RxHttpClientIT {
 
         // then
         StepVerifier.create(response)
-                .expectError(IllegalStateException.class)
-                .verify(TIMEOUT);
-        assertRetry();
-    }
-
-    @Test
-    void getWithCustomRetryExhaustedExceptionWhen500() throws Exception {
-        // given
-        REQUEST_COUNTER = new AtomicInteger();
-        final HttpRequest httpRequest = requestFor("/retry-get-500")
-                .method(HttpMethod.GET)
-                .build();
-        final RxHttpClient cut = RxHttpClientFactory.create(ImmutableRxHttpClientConfig.builder()
-                .retryConfig(defaultRetryConfig()
-                        .onRetryExhaustedException(ReadTimeoutException.INSTANCE)
-                        .retryableHttpResponseCodes(HashSet.of(500))
-                        .build())
-                .build());
-
-        // when
-        final Mono<HttpResponse> response = cut.call(httpRequest);
-
-        // then
-        StepVerifier.create(response)
-                .expectError(ReadTimeoutException.class)
+                .expectError(RetryableException.class)
                 .verify(TIMEOUT);
         assertRetry();
     }