Merge branch 'master' of github.com:confluentinc/ksql

confluentinc · Dec 12, 2019 · 3eecd40 · 3eecd40
2 parents 54bc2be + da96259
commit 3eecd40
Show file tree

Hide file tree

Showing 8 changed files with 54 additions and 24 deletions.
diff --git a/ksql-common/src/main/java/io/confluent/ksql/util/KsqlConfig.java b/ksql-common/src/main/java/io/confluent/ksql/util/KsqlConfig.java
@@ -187,6 +187,13 @@ public class KsqlConfig extends AbstractConfig {
   public static final Collection<CompatibilityBreakingConfigDef> COMPATIBLY_BREAKING_CONFIG_DEFS
       = ImmutableList.of();
 
+  public static final String KSQL_SHUTDOWN_TIMEOUT_MS_CONFIG =
+      "ksql.streams.shutdown.timeout.ms";
+  public static final Long KSQL_SHUTDOWN_TIMEOUT_MS_DEFAULT = 300_000L;
+  public static final String KSQL_SHUTDOWN_TIMEOUT_MS_DOC = "Timeout in "
+      + "milliseconds to block waiting for the underlying streams instance to exit";
+
+
   private enum ConfigGeneration {
     LEGACY,
     CURRENT
@@ -499,6 +506,12 @@ private static ConfigDef buildConfigDef(final ConfigGeneration generation) {
             KSQL_ACTIVE_PERSISTENT_QUERY_LIMIT_DEFAULT,
             Importance.MEDIUM,
             KSQL_ACTIVE_PERSISTENT_QUERY_LIMIT_DOC
+        ).define(
+            KSQL_SHUTDOWN_TIMEOUT_MS_CONFIG,
+            Type.LONG,
+            KSQL_SHUTDOWN_TIMEOUT_MS_DEFAULT,
+            Importance.MEDIUM,
+            KSQL_SHUTDOWN_TIMEOUT_MS_DOC
         )
         .withClientSslSupport();
 

diff --git a/ksql-engine/src/main/java/io/confluent/ksql/query/QueryExecutor.java b/ksql-engine/src/main/java/io/confluent/ksql/query/QueryExecutor.java
@@ -15,6 +15,8 @@
 
 package io.confluent.ksql.query;
 
+import static io.confluent.ksql.util.KsqlConfig.KSQL_SHUTDOWN_TIMEOUT_MS_CONFIG;
+
 import io.confluent.ksql.GenericRow;
 import io.confluent.ksql.errors.ProductionExceptionHandlerUtil;
 import io.confluent.ksql.execution.builder.KsqlQueryBuilder;
@@ -176,8 +178,8 @@ public TransientQueryMetadata buildTransientQuery(
         built.topology,
         streamsProperties,
         overrides,
-        queryCloseCallback
-    );
+        queryCloseCallback,
+        ksqlConfig.getLong(KSQL_SHUTDOWN_TIMEOUT_MS_CONFIG));
   }
 
   private static Optional<MaterializationInfo> getMaterializationInfo(final Object result) {
@@ -237,8 +239,8 @@ public PersistentQueryMetadata buildQuery(
         ksqlQueryBuilder.getSchemas(),
         streamsProperties,
         overrides,
-        queryCloseCallback
-    );
+        queryCloseCallback,
+        ksqlConfig.getLong(KSQL_SHUTDOWN_TIMEOUT_MS_CONFIG));
   }
 
   private TransientQueryQueue buildTransientQueryQueue(

diff --git a/ksql-engine/src/main/java/io/confluent/ksql/util/PersistentQueryMetadata.java b/ksql-engine/src/main/java/io/confluent/ksql/util/PersistentQueryMetadata.java
@@ -63,8 +63,8 @@ public PersistentQueryMetadata(
       final QuerySchemas schemas,
       final Map<String, Object> streamsProperties,
       final Map<String, Object> overriddenProperties,
-      final Consumer<QueryMetadata> closeCallback
-  ) {
+      final Consumer<QueryMetadata> closeCallback,
+      final Long closeTimeout) {
     // CHECKSTYLE_RULES.ON: ParameterNumberCheck
     super(
         statementString,
@@ -76,7 +76,8 @@ public PersistentQueryMetadata(
         topology,
         streamsProperties,
         overriddenProperties,
-        closeCallback);
+        closeCallback,
+        closeTimeout);
 
     this.id = requireNonNull(id, "id");
     this.resultTopic = requireNonNull(resultTopic, "resultTopic");

diff --git a/ksql-engine/src/main/java/io/confluent/ksql/util/QueryMetadata.java b/ksql-engine/src/main/java/io/confluent/ksql/util/QueryMetadata.java
@@ -20,6 +20,7 @@
 import io.confluent.ksql.name.SourceName;
 import io.confluent.ksql.schema.ksql.LogicalSchema;
 import java.lang.Thread.UncaughtExceptionHandler;
+import java.time.Duration;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
@@ -44,6 +45,7 @@ public class QueryMetadata {
   private final Consumer<QueryMetadata> closeCallback;
   private final Set<SourceName> sourceNames;
   private final LogicalSchema logicalSchema;
+  private final Long closeTimeout;
 
   private Optional<QueryStateListener> queryStateListener = Optional.empty();
   private boolean everStarted = false;
@@ -59,8 +61,8 @@ protected QueryMetadata(
       final Topology topology,
       final Map<String, Object> streamsProperties,
       final Map<String, Object> overriddenProperties,
-      final Consumer<QueryMetadata> closeCallback
-  ) {
+      final Consumer<QueryMetadata> closeCallback,
+      final Long closeTimeout) {
     // CHECKSTYLE_RULES.ON: ParameterNumberCheck
     this.statementString = Objects.requireNonNull(statementString, "statementString");
     this.kafkaStreams = Objects.requireNonNull(kafkaStreams, "kafkaStreams");
@@ -76,6 +78,7 @@ protected QueryMetadata(
     this.closeCallback = Objects.requireNonNull(closeCallback, "closeCallback");
     this.sourceNames = Objects.requireNonNull(sourceNames, "sourceNames");
     this.logicalSchema = Objects.requireNonNull(logicalSchema, "logicalSchema");
+    this.closeTimeout = Objects.requireNonNull(closeTimeout, "closeTimeout");
   }
 
   protected QueryMetadata(final QueryMetadata other, final Consumer<QueryMetadata> closeCallback) {
@@ -89,6 +92,7 @@ protected QueryMetadata(final QueryMetadata other, final Consumer<QueryMetadata>
     this.sourceNames = other.sourceNames;
     this.logicalSchema = other.logicalSchema;
     this.closeCallback = Objects.requireNonNull(closeCallback, "closeCallback");
+    this.closeTimeout = other.closeTimeout;
   }
 
   public void registerQueryStateListener(final QueryStateListener queryStateListener) {
@@ -141,7 +145,7 @@ public boolean hasEverBeenStarted() {
   }
 
   public void close() {
-    kafkaStreams.close();
+    kafkaStreams.close(Duration.ofMillis(closeTimeout));
 
     kafkaStreams.cleanUp();
 

diff --git a/ksql-engine/src/main/java/io/confluent/ksql/util/TransientQueryMetadata.java b/ksql-engine/src/main/java/io/confluent/ksql/util/TransientQueryMetadata.java
@@ -51,7 +51,8 @@ public TransientQueryMetadata(
       final Topology topology,
       final Map<String, Object> streamsProperties,
       final Map<String, Object> overriddenProperties,
-      final Consumer<QueryMetadata> closeCallback) {
+      final Consumer<QueryMetadata> closeCallback,
+      final Long closeTimeout) {
     // CHECKSTYLE_RULES.ON: ParameterNumberCheck
     super(
         statementString,
@@ -63,8 +64,8 @@ public TransientQueryMetadata(
         topology,
         streamsProperties,
         overriddenProperties,
-        closeCallback
-    );
+        closeCallback,
+        closeTimeout);
     this.limitHandlerSetter = Objects.requireNonNull(limitHandlerSetter, "limitHandlerSetter");
     this.rowQueue = Objects.requireNonNull(rowQueue, "rowQueue");
 

diff --git a/ksql-engine/src/test/java/io/confluent/ksql/util/QueryMetadataTest.java b/ksql-engine/src/test/java/io/confluent/ksql/util/QueryMetadataTest.java
@@ -15,7 +15,6 @@
 
 package io.confluent.ksql.util;
 
-import static io.confluent.ksql.metastore.model.DataSource.DataSourceType;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.is;
 import static org.mockito.Mockito.inOrder;
@@ -28,6 +27,7 @@
 import io.confluent.ksql.name.SourceName;
 import io.confluent.ksql.schema.ksql.LogicalSchema;
 import io.confluent.ksql.schema.ksql.types.SqlTypes;
+import java.time.Duration;
 import java.util.Collections;
 import java.util.Set;
 import java.util.function.Consumer;
@@ -49,6 +49,7 @@ public class QueryMetadataTest {
       .valueColumn(ColumnName.of("f0"), SqlTypes.STRING)
       .build();
   private static final Set<SourceName> SOME_SOURCES = ImmutableSet.of(SourceName.of("s1"), SourceName.of("s2"));
+  private static final Long closeTimeout = KsqlConfig.KSQL_SHUTDOWN_TIMEOUT_MS_DEFAULT;
 
   @Mock
   private Topology topoplogy;
@@ -72,8 +73,8 @@ public void setup() {
         topoplogy,
         Collections.emptyMap(),
         Collections.emptyMap(),
-        closeCallback
-    );
+        closeCallback,
+        closeTimeout);
   }
 
   @Test
@@ -131,7 +132,7 @@ public void shouldCloseKStreamsAppOnCloseThenCloseCallback() {
 
     // Then:
     final InOrder inOrder = inOrder(kafkaStreams, closeCallback);
-    inOrder.verify(kafkaStreams).close();
+    inOrder.verify(kafkaStreams).close(Duration.ofMillis(closeTimeout));
     inOrder.verify(closeCallback).accept(query);
   }
 
@@ -142,7 +143,7 @@ public void shouldCleanUpKStreamsAppAfterCloseOnClose() {
 
     // Then:
     final InOrder inOrder = inOrder(kafkaStreams);
-    inOrder.verify(kafkaStreams).close();
+    inOrder.verify(kafkaStreams).close(Duration.ofMillis(closeTimeout));
     inOrder.verify(kafkaStreams).cleanUp();
   }
 

diff --git a/ksql-rest-app/src/test/java/io/confluent/ksql/rest/entity/QueryDescriptionFactoryTest.java b/ksql-rest-app/src/test/java/io/confluent/ksql/rest/entity/QueryDescriptionFactoryTest.java
@@ -33,6 +33,7 @@
 import io.confluent.ksql.schema.ksql.SqlBaseType;
 import io.confluent.ksql.schema.ksql.types.SqlTypes;
 import io.confluent.ksql.serde.SerdeOption;
+import io.confluent.ksql.util.KsqlConfig;
 import io.confluent.ksql.util.PersistentQueryMetadata;
 import io.confluent.ksql.util.QueryMetadata;
 import io.confluent.ksql.util.QuerySchemas;
@@ -74,6 +75,7 @@ public class QueryDescriptionFactoryTest {
   private static final ImmutableSet<SourceName> SOURCE_NAMES = ImmutableSet.of(SourceName.of("s1"), SourceName.of("s2"));
   private static final String SQL_TEXT = "test statement";
   private static final String TOPOLOGY_TEXT = "Topology Text";
+  private static final Long closeTimeout = KsqlConfig.KSQL_SHUTDOWN_TIMEOUT_MS_DEFAULT;
 
   @Mock
   private Consumer<QueryMetadata> queryCloseCallback;
@@ -108,7 +110,8 @@ public void setUp() {
         topology,
         STREAMS_PROPS,
         PROP_OVERRIDES,
-        queryCloseCallback);
+        queryCloseCallback,
+        closeTimeout);
 
     transientQueryDescription = QueryDescriptionFactory.forQueryMetadata(transientQuery);
 
@@ -128,7 +131,8 @@ public void setUp() {
         QuerySchemas.of(new LinkedHashMap<>()),
         STREAMS_PROPS,
         PROP_OVERRIDES,
-        queryCloseCallback);
+        queryCloseCallback,
+        closeTimeout);
 
     persistentQueryDescription = QueryDescriptionFactory.forQueryMetadata(persistentQuery);
   }
@@ -221,7 +225,8 @@ public void shouldHandleRowTimeInValueSchemaForTransientQuery() {
         topology,
         STREAMS_PROPS,
         PROP_OVERRIDES,
-        queryCloseCallback);
+        queryCloseCallback,
+        closeTimeout);
 
     // When:
     transientQueryDescription = QueryDescriptionFactory.forQueryMetadata(transientQuery);
@@ -255,7 +260,8 @@ public void shouldHandleRowKeyInValueSchemaForTransientQuery() {
         topology,
         STREAMS_PROPS,
         PROP_OVERRIDES,
-        queryCloseCallback);
+        queryCloseCallback,
+        closeTimeout);
 
     // When:
     transientQueryDescription = QueryDescriptionFactory.forQueryMetadata(transientQuery);

diff --git a/...est/java/io/confluent/ksql/rest/server/resources/streaming/StreamedQueryResourceTest.java b/...est/java/io/confluent/ksql/rest/server/resources/streaming/StreamedQueryResourceTest.java
@@ -110,6 +110,7 @@ public class StreamedQueryResourceTest {
   private static final KsqlConfig VALID_CONFIG = new KsqlConfig(ImmutableMap.of(
       StreamsConfig.APPLICATION_SERVER_CONFIG, "something:1"
   ));
+  private static final Long closeTimeout = KsqlConfig.KSQL_SHUTDOWN_TIMEOUT_MS_DEFAULT;
 
   private static final String TOPIC_NAME = "test_stream";
   private static final String PUSH_QUERY_STRING = "SELECT * FROM " + TOPIC_NAME + " EMIT CHANGES;";
@@ -378,7 +379,8 @@ public void shouldStreamRowsCorrectly() throws Throwable {
             mock(Topology.class),
             Collections.emptyMap(),
             Collections.emptyMap(),
-            queryCloseCallback);
+            queryCloseCallback,
+            closeTimeout);
 
     when(mockKsqlEngine.executeQuery(serviceContext,
         ConfiguredStatement.of(query, requestStreamsProperties, VALID_CONFIG)))
@@ -451,7 +453,7 @@ public void shouldStreamRowsCorrectly() throws Throwable {
     verify(mockKafkaStreams).start();
     verify(mockKafkaStreams).setUncaughtExceptionHandler(any());
     verify(mockKafkaStreams).cleanUp();
-    verify(mockKafkaStreams).close();
+    verify(mockKafkaStreams).close(Duration.ofMillis(closeTimeout));
 
     // If one of the other threads has somehow managed to throw an exception without breaking things up until this
     // point, we throw that exception now in the main thread and cause the test to fail