Set default Kafka auto offset explicitly if it is not provided in con…

…figuration (#1992)
apache · Oct 3, 2023 · 504ae68 · 504ae68
1 parent 59a59a8
commit 504ae68
Showing 1 changed file with 9 additions and 2 deletions.
diff --git a/.../src/main/java/org/apache/streampipes/pe/shared/config/kafka/kafka/KafkaConnectUtils.java b/.../src/main/java/org/apache/streampipes/pe/shared/config/kafka/kafka/KafkaConnectUtils.java
@@ -123,10 +123,17 @@ public static KafkaConfig getConfig(IStaticPropertyExtractor extractor, boolean
           new KafkaSecurityUnauthenticatedPlainConfig();
     }
 
+    // Set default value if no value is provided.
+    if (alternatives == null) {
+      AutoOffsetResetConfig autoOffsetResetConfig = new AutoOffsetResetConfig(KafkaConnectUtils.LATEST);
 
+      return new KafkaConfig(brokerUrl, port, topic, securityConfig, autoOffsetResetConfig);
+    } else {
+      String auto = extractor.selectedAlternativeInternalId(AUTO_OFFSET_RESET_CONFIG);
+      AutoOffsetResetConfig autoOffsetResetConfig = new AutoOffsetResetConfig(auto);
 
-    String auto = extractor.selectedAlternativeInternalId(AUTO_OFFSET_RESET_CONFIG);
-    AutoOffsetResetConfig autoOffsetResetConfig = new AutoOffsetResetConfig(auto);
+      return new KafkaConfig(brokerUrl, port, topic, securityConfig, autoOffsetResetConfig);
+    }
 
     return new KafkaConfig(brokerUrl, port, topic, securityConfig, autoOffsetResetConfig);
   }