Add reconnect backoff config for kafka stream to avoid repeatedly con…

…necting to host in a tight loop.
Commonjava · Jul 5, 2024 · 2f427b6 · 2f427b6
1 parent 3ed2fd2
commit 2f427b6
Showing 1 changed file with 6 additions and 0 deletions.
diff --git a/subsys/kafka/src/main/java/org/commonjava/indy/subsys/kafka/boot/KafkaStreamBooter.java b/subsys/kafka/src/main/java/org/commonjava/indy/subsys/kafka/boot/KafkaStreamBooter.java
@@ -59,6 +59,10 @@ public class KafkaStreamBooter
 
     private static final long DEFAULT_KAFKA_STREAM_CLOSE_TIMEOUT = 5 * 60 * 3000L;
 
+    private static final long RECONNECT_BACKOFF_MS = 60 * 1000L;
+
+    private static final long RECONNECT_BACKOFF_MAX_MS = 30 * 60 * 1000L;
+
     private KafkaStreams streams;
 
     @Override
@@ -133,6 +137,8 @@ private Properties setKafkaProps()
         props.putIfAbsent( StreamsConfig.APPLICATION_ID_CONFIG, config.getGroup() );
         props.putIfAbsent( StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, config.getBootstrapServers() );
         props.putIfAbsent( StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_CONFIG, config.getRecordsPerPartition() );
+        props.putIfAbsent( StreamsConfig.RECONNECT_BACKOFF_MS_CONFIG, RECONNECT_BACKOFF_MS );
+        props.putIfAbsent( StreamsConfig.RECONNECT_BACKOFF_MAX_MS_CONFIG, RECONNECT_BACKOFF_MAX_MS );
 
         logger.info( "Kafka props: {}", props );
         return props;