[BUG] Retry interval for kafka consumer

- added retry interval of 10 seconds in case of any auth exception when
  consuming kafka messages.
- This fix prevents us from restarting the whole app inorder to restart
  the consumer.

Issue-ID: CPS-2070
Change-Id: Ie02d3cb686741cd92dd006dab1d49c3480d58720
Signed-off-by: mpriyank <priyank.maheshwari@est.tech>

diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/config/kafka/KafkaConfig.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/config/kafka/KafkaConfig.java
index 5149675..4f2674a 100644 (file)
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/config/kafka/KafkaConfig.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/config/kafka/KafkaConfig.java
@@ -21,6 +21,7 @@
 package org.onap.cps.ncmp.api.impl.config.kafka;
 
 import io.cloudevents.CloudEvent;
+import java.time.Duration;
 import java.util.Map;
 import lombok.RequiredArgsConstructor;
 import org.apache.kafka.clients.producer.ProducerConfig;
@@ -99,6 +100,7 @@ public class KafkaConfig<T> {
         final ConcurrentKafkaListenerContainerFactory<String, T> containerFactory =
                 new ConcurrentKafkaListenerContainerFactory<>();
         containerFactory.setConsumerFactory(legacyEventConsumerFactory());
+        containerFactory.getContainerProperties().setAuthExceptionRetryInterval(Duration.ofSeconds(10));
         return containerFactory;
     }
 
@@ -150,6 +152,7 @@ public class KafkaConfig<T> {
         final ConcurrentKafkaListenerContainerFactory<String, CloudEvent> containerFactory =
                 new ConcurrentKafkaListenerContainerFactory<>();
         containerFactory.setConsumerFactory(cloudEventConsumerFactory());
+        containerFactory.getContainerProperties().setAuthExceptionRetryInterval(Duration.ofSeconds(10));
         return containerFactory;
     }