fix: fix_application_id_to_work_with_acls (#8277)

* fix: fix_application_id_to_work_with_acls * checkstyle
confluentinc · Oct 22, 2021 · 64f58e8 · 64f58e8
1 parent 0848946
commit 64f58e8
Showing 1 changed file with 18 additions and 2 deletions.
diff --git a/ksqldb-engine/src/main/java/io/confluent/ksql/query/QueryBuilder.java b/ksqldb-engine/src/main/java/io/confluent/ksql/query/QueryBuilder.java
@@ -63,6 +63,7 @@
 import io.confluent.ksql.util.PushQueryMetadata.ResultType;
 import io.confluent.ksql.util.QueryApplicationId;
 import io.confluent.ksql.util.QueryMetadata;
+import io.confluent.ksql.util.ReservedInternalTopics;
 import io.confluent.ksql.util.SharedKafkaStreamsRuntime;
 import io.confluent.ksql.util.SharedKafkaStreamsRuntimeImpl;
 import io.confluent.ksql.util.TransientQueryMetadata;
@@ -484,20 +485,35 @@ private SharedKafkaStreamsRuntime getKafkaStreamsInstance(
       }
     }
     final SharedKafkaStreamsRuntime stream;
+    final KsqlConfig ksqlConfig = config.getConfig(true);
+    final String queryPrefix = ksqlConfig
+        .getString(KsqlConfig.KSQL_PERSISTENT_QUERY_NAME_PREFIX_CONFIG);
+    final String serviceId = ksqlConfig.getString(KsqlConfig.KSQL_SERVICE_ID_CONFIG);
     if (real) {
       stream = new SharedKafkaStreamsRuntimeImpl(
           kafkaStreamsBuilder,
           config.getConfig(true).getInt(KsqlConfig.KSQL_QUERY_ERROR_MAX_QUEUE_SIZE),
           buildStreamsProperties(
-              "_confluent-ksql-" + streams.size() + "-" + UUID.randomUUID(),
+              ReservedInternalTopics.KSQL_INTERNAL_TOPIC_PREFIX
+                  + serviceId
+                  + queryPrefix
+                  + streams.size()
+                  + "-"
+                  + UUID.randomUUID(),
               queryID)
       );
     } else {
       stream = new ValidationSharedKafkaStreamsRuntimeImpl(
           kafkaStreamsBuilder,
           config.getConfig(true).getInt(KsqlConfig.KSQL_QUERY_ERROR_MAX_QUEUE_SIZE),
           buildStreamsProperties(
-              "_confluent-ksql-" + streams.size() + "-" + UUID.randomUUID() + "-validation",
+              ReservedInternalTopics.KSQL_INTERNAL_TOPIC_PREFIX
+                  + serviceId
+                  + queryPrefix
+                  + streams.size()
+                  + "-"
+                  + UUID.randomUUID()
+                  + "-validation",
               queryID)
       );
     }