add kafka config topic as defult internal topic if using kafka backend

kafka-ops · Nov 12, 2021 · 2f84410 · 2f84410
1 parent 35acbe4
commit 2f84410
Showing 1 changed file with 6 additions and 1 deletion.
diff --git a/src/main/java/com/purbon/kafka/topology/Configuration.java b/src/main/java/com/purbon/kafka/topology/Configuration.java
@@ -190,9 +190,14 @@ public String getProperty(String key) {
   }
 
   public List<String> getKafkaInternalTopicPrefixes() {
-    return config.getStringList(KAFKA_INTERNAL_TOPIC_PREFIXES).stream()
+    var prefixes = config.getStringList(KAFKA_INTERNAL_TOPIC_PREFIXES).stream()
         .map(String::trim)
         .collect(Collectors.toList());
+    if (getStateProcessorImplementationClassName().equalsIgnoreCase(KAFKA_STATE_PROCESSOR_CLASS)) {
+      LOGGER.info("Adding "+getJulieKafkaConfigTopic()+" as internal topic because we are using the Kafka Backend");
+      prefixes.add(getJulieKafkaConfigTopic());
+    }
+    return prefixes;
   }
 
   public List<String> getServiceAccountManagedPrefixes() {