fix: Removes config check to insert SPQ Processor (confluentinc#8062)

* fix: Adds a new config check to insert SPQ Processor
wicknicks · Sep 1, 2021 · a0be1d5 · a0be1d5
1 parent ccb8936
commit a0be1d5
Show file tree

Hide file tree

Showing 6 changed files with 20 additions and 2 deletions.
diff --git a/ksqldb-common/src/main/java/io/confluent/ksql/util/KsqlConfig.java b/ksqldb-common/src/main/java/io/confluent/ksql/util/KsqlConfig.java
@@ -285,6 +285,13 @@ public class KsqlConfig extends AbstractConfig {
           + "functions, aggregations, or joins, but may include projections and filters.";
   public static final boolean KSQL_QUERY_PUSH_SCALABLE_ENABLED_DEFAULT = false;
 
+  public static final String KSQL_QUERY_PUSH_SCALABLE_REGISTRY_INSTALLED
+      = "ksql.query.push.scalable.registry.installed";
+  public static final String KSQL_QUERY_PUSH_SCALABLE_REGISTRY_INSTALLED_DOC =
+      "Enables whether scalable push registry should be installed. This is a requirement of "
+          + "enabling scalable push queries using ksql.query.push.scalable.enabled.";
+  public static final boolean KSQL_QUERY_PUSH_SCALABLE_REGISTRY_INSTALLED_DEFAULT = false;
+
   public static final String KSQL_QUERY_PUSH_SCALABLE_NEW_NODE_CONTINUITY
       = "ksql.query.push.scalable.new.node.continuity";
   public static final String KSQL_QUERY_PUSH_SCALABLE_NEW_NODE_CONTINUITY_DOC =
@@ -923,6 +930,13 @@ private static ConfigDef buildConfigDef(final ConfigGeneration generation) {
             Importance.LOW,
             KSQL_QUERY_PUSH_SCALABLE_ENABLED_DOC
         )
+        .define(
+            KSQL_QUERY_PUSH_SCALABLE_REGISTRY_INSTALLED,
+            Type.BOOLEAN,
+            KSQL_QUERY_PUSH_SCALABLE_REGISTRY_INSTALLED_DEFAULT,
+            Importance.LOW,
+            KSQL_QUERY_PUSH_SCALABLE_REGISTRY_INSTALLED_DOC
+        )
         .define(
             KSQL_QUERY_PUSH_SCALABLE_NEW_NODE_CONTINUITY,
             Type.BOOLEAN,

diff --git a/ksqldb-engine/src/main/java/io/confluent/ksql/physical/scalablepush/ProcessingQueue.java b/ksqldb-engine/src/main/java/io/confluent/ksql/physical/scalablepush/ProcessingQueue.java
@@ -30,7 +30,7 @@
  */
 public class ProcessingQueue {
 
-  static final int BLOCKING_QUEUE_CAPACITY = 100;
+  static final int BLOCKING_QUEUE_CAPACITY = 1000;
 
   private final Deque<TableRow> rowQueue;
   private final QueryId queryId;

diff --git a/ksqldb-engine/src/main/java/io/confluent/ksql/query/QueryExecutor.java b/ksqldb-engine/src/main/java/io/confluent/ksql/query/QueryExecutor.java
@@ -225,7 +225,7 @@ private static Optional<ScalablePushRegistry> applyScalablePushProcessor(
       final Map<String, Object> streamsProperties,
       final KsqlConfig ksqlConfig
   ) {
-    if (!ksqlConfig.getBoolean(KsqlConfig.KSQL_QUERY_PUSH_SCALABLE_ENABLED)) {
+    if (!ksqlConfig.getBoolean(KsqlConfig.KSQL_QUERY_PUSH_SCALABLE_REGISTRY_INSTALLED)) {
       return Optional.empty();
     }
     final KStream<?, GenericRow> stream;

diff --git a/...-functional-tests/src/test/java/io/confluent/ksql/test/rest/RestQueryTranslationTest.java b/...-functional-tests/src/test/java/io/confluent/ksql/test/rest/RestQueryTranslationTest.java
@@ -102,6 +102,7 @@ public class RestQueryTranslationTest {
       .withProperty(KsqlConfig.SCHEMA_REGISTRY_URL_PROPERTY, "set")
       .withProperty(KsqlConfig.KSQL_QUERY_PULL_TABLE_SCAN_ENABLED, true)
       .withProperty(KsqlConfig.KSQL_QUERY_PULL_INTERPRETER_ENABLED, true)
+      .withProperty(KsqlConfig.KSQL_QUERY_PUSH_SCALABLE_REGISTRY_INSTALLED, true)
       .withProperty(KsqlConfig.KSQL_QUERY_PUSH_SCALABLE_ENABLED, true)
       .withStaticServiceContext(TEST_HARNESS::getServiceContext)
       .build();

diff --git a/ksqldb-rest-app/src/test/java/io/confluent/ksql/rest/integration/RestApiTest.java b/ksqldb-rest-app/src/test/java/io/confluent/ksql/rest/integration/RestApiTest.java
@@ -215,6 +215,7 @@ public class RestApiTest {
       .withProperty("sasl.mechanism", "PLAIN")
       .withProperty("sasl.jaas.config", SecureKafkaHelper.buildJaasConfig(NORMAL_USER))
       .withProperties(ClientTrustStore.trustStoreProps())
+      .withProperty(KsqlConfig.KSQL_QUERY_PUSH_SCALABLE_REGISTRY_INSTALLED, true)
       .withProperty(KsqlConfig.KSQL_QUERY_PUSH_SCALABLE_ENABLED, true)
       .build();
 

diff --git a/...app/src/test/java/io/confluent/ksql/rest/integration/ScalablePushQueryFunctionalTest.java b/...app/src/test/java/io/confluent/ksql/rest/integration/ScalablePushQueryFunctionalTest.java
@@ -98,6 +98,7 @@ public class ScalablePushQueryFunctionalTest {
       .withProperty(KsqlRestConfig.LISTENERS_CONFIG, "http://localhost:8088")
       .withProperty(KsqlRestConfig.ADVERTISED_LISTENER_CONFIG, "http://localhost:8088")
       .withProperty(KsqlConfig.KSQL_QUERY_PULL_ENABLE_STANDBY_READS, true)
+      .withProperty(KsqlConfig.KSQL_QUERY_PUSH_SCALABLE_REGISTRY_INSTALLED, true)
       .withProperty(KsqlConfig.KSQL_QUERY_PUSH_SCALABLE_ENABLED, true)
       // Make rebalances happen quicker for the sake of the test
       .withProperty(KSQL_STREAMS_PREFIX + "max.poll.interval.ms", 5000)
@@ -112,6 +113,7 @@ public class ScalablePushQueryFunctionalTest {
       .withProperty(KsqlRestConfig.LISTENERS_CONFIG, "http://localhost:8089")
       .withProperty(KsqlRestConfig.ADVERTISED_LISTENER_CONFIG, "http://localhost:8089")
       .withProperty(KsqlConfig.KSQL_QUERY_PULL_ENABLE_STANDBY_READS, true)
+      .withProperty(KsqlConfig.KSQL_QUERY_PUSH_SCALABLE_REGISTRY_INSTALLED, true)
       .withProperty(KsqlConfig.KSQL_QUERY_PUSH_SCALABLE_ENABLED, true)
       // Make rebalances happen quicker for the sake of the test
       .withProperty(KSQL_STREAMS_PREFIX + "max.poll.interval.ms", 5000)