Handle 429 error Too Many Requests

Issue-ID: DCAEGEN2-2827
Signed-off-by: Pawel <pawel.kasperkiewicz@nokia.com>
Change-Id: Iedfb6572f008876b52a102948aeb640d27e51314

3 files changed, 12 insertions, 2 deletions

diff --git a/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/error/ClientErrorReasons.java b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/error/ClientErrorReasons.java
index 431843ae..f6e5c2bb 100644
--- a/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/error/ClientErrorReasons.java
+++ b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/error/ClientErrorReasons.java
@@ -30,7 +30,8 @@ public class ClientErrorReasons {
             .header("408 Request Timeout")
             .text("Client timeout exception occurred, Error code is %1")
             .messageId("SVC0001")
-            .variables(Collections.singletonList("408")).build();
+            .variables(Collections.singletonList("408"))
+            .build();
 
     public static final ClientErrorReason SERVICE_UNAVAILABLE = ImmutableClientErrorReason.builder()
             .header("503 Service unavailable")
@@ -38,4 +39,10 @@ public class ClientErrorReasons {
             .messageId("SVC2001")
             .build();
 
+    public static final ClientErrorReason CONNECTION_POLL_LIMIT = ImmutableClientErrorReason.builder()
+            .header("429 Too Many Requests")
+            .text("Pending acquire queue has reached its maximum size")
+            .messageId("SVC2000")
+            .variables(Collections.singletonList("429"))
+            .build();
 }
diff --git a/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterPublisherImpl.java b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterPublisherImpl.java
index 6e4679c3..534fca6b 100644
--- a/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterPublisherImpl.java
+++ b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterPublisherImpl.java
@@ -50,6 +50,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.netty.internal.shaded.reactor.pool.PoolAcquirePendingLimitException;
 
 import java.net.ConnectException;
 import java.time.Duration;
@@ -95,6 +96,7 @@ public class MessageRouterPublisherImpl implements MessageRouterPublisher {
                         e -> LOGGER.error("Timeout exception occurred when sending items to DMaaP MR", e))
                 .onErrorResume(ReadTimeoutException.class, e -> buildErrorResponse(ClientErrorReasons.TIMEOUT))
                 .doOnError(ConnectException.class, e -> LOGGER.error("DMaaP MR is unavailable, {}", e.getMessage()))
+                .onErrorResume(PoolAcquirePendingLimitException.class, e -> buildErrorResponse(ClientErrorReasons.CONNECTION_POLL_LIMIT))
                 .onErrorResume(ConnectException.class, e -> buildErrorResponse(ClientErrorReasons.SERVICE_UNAVAILABLE))
                 .onErrorResume(RetryableException.class, e -> Mono.just(buildResponse(e.getResponse(), batch)));
     }
diff --git a/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/model/config/DmaapConnectionPoolConfig.java b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/model/config/DmaapConnectionPoolConfig.java
index 38166e4e..4a4add0e 100644
--- a/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/model/config/DmaapConnectionPoolConfig.java
+++ b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/model/config/DmaapConnectionPoolConfig.java
@@ -20,13 +20,14 @@
 package org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.config;
 
 import org.immutables.value.Value;
+import reactor.netty.resources.ConnectionProvider;
 
 @Value.Immutable
 public interface DmaapConnectionPoolConfig {
 
     @Value.Default
     default int connectionPool(){
-        return 16;
+        return ConnectionProvider.DEFAULT_POOL_MAX_CONNECTIONS;
     }
     @Value.Default
     default int maxLifeTime(){