fixup! fix: Consider topics created by join operations internal

confluentinc · Dec 17, 2021 · 9e2652e · 9e2652e
1 parent 0c9865e
commit 9e2652e
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 2 deletions.
diff --git a/ksqldb-engine/src/main/java/io/confluent/ksql/services/KafkaTopicClientImpl.java b/ksqldb-engine/src/main/java/io/confluent/ksql/services/KafkaTopicClientImpl.java
@@ -375,9 +375,9 @@ private Config getConfig() {
   private static boolean isInternalTopic(final String topicName, final String applicationId) {
     return topicName.startsWith(applicationId + "-")
         && (topicName.endsWith(KsqlConstants.STREAMS_CHANGELOG_TOPIC_SUFFIX)
-        || topicName.endsWith(KsqlConstants.STREAMS_REPARTITION_TOPIC_SUFFIX))
+        || topicName.endsWith(KsqlConstants.STREAMS_REPARTITION_TOPIC_SUFFIX)
         || topicName.matches(KsqlConstants.STREAMS_JOIN_REGISTRATION_TOPIC_PATTERN)
-        || topicName.matches(KsqlConstants.STREAMS_JOIN_RESPONSE_TOPIC_PATTERN);
+        || topicName.matches(KsqlConstants.STREAMS_JOIN_RESPONSE_TOPIC_PATTERN));
   }
 
   private void validateTopicProperties(

diff --git a/ksqldb-engine/src/test/java/io/confluent/ksql/services/KafkaTopicClientImplTest.java b/ksqldb-engine/src/test/java/io/confluent/ksql/services/KafkaTopicClientImplTest.java
@@ -453,12 +453,21 @@ public void shouldDeleteInternalTopics() {
         + "-KTABLE-FK-JOIN-SUBSCRIPTION-RESPONSE-0000000012-topic";
     final String internalTopic3 = applicationId + "-something-changelog";
     final String internalTopic4 = applicationId + "-something-repartition";
+    // the next 2 topics four are not prefixed with the application id, so they are not internal
+    final String customTopic1 = "what-KTABLE-FK-JOIN-SUBSCRIPTION-REGISTRATION-0000000012-topic";
+    final String customTopic2 = "eva-KTABLE-FK-JOIN-SUBSCRIPTION-RESPONSE-0000000012-topic";
+    final String customTopic3 = "-something-changelog";
+    final String customTopic4 = "-something-repartition";
 
     givenTopicExists("topic1", 1, 1);
     givenTopicExists(internalTopic1, 1, 1);
     givenTopicExists(internalTopic2, 1, 1);
     givenTopicExists(internalTopic3, 1, 1);
     givenTopicExists(internalTopic4, 1, 1);
+    givenTopicExists(customTopic1, 1, 1);
+    givenTopicExists(customTopic2, 1, 1);
+    givenTopicExists(customTopic3, 1, 1);
+    givenTopicExists(customTopic4, 1, 1);
     givenTopicExists("topic2", 1, 1);
 
     // When: