Merge branch 'master' of github.com:snowflakedb/snowflake-kafka-conne…

…ctor
snowflakedb · Aug 18, 2020 · 39532f3 · 39532f3
2 parents c32827c + 3886aa6
commit 39532f3
Show file tree

Hide file tree

Showing 3 changed files with 15 additions and 1 deletion.
diff --git a/src/main/java/com/snowflake/kafka/connector/SnowflakeSinkTask.java b/src/main/java/com/snowflake/kafka/connector/SnowflakeSinkTask.java
@@ -156,7 +156,10 @@ public void start(final Map<String, String> parsedConfig)
   public void stop()
   {
     LOGGER.info(Logging.logMessage("SnowflakeSinkTask[ID:{}]:stop", this.id));
-    // do nothing
+    if (this.sink != null)
+    {
+      this.sink.setIsStoppedToTrue(); // close cleaner thread
+    }
   }
 
   /**

diff --git a/src/main/java/com/snowflake/kafka/connector/internal/SnowflakeSinkService.java b/src/main/java/com/snowflake/kafka/connector/internal/SnowflakeSinkService.java
@@ -66,6 +66,11 @@ public interface SnowflakeSinkService
    */
   void close(Collection<TopicPartition> partitions);
 
+  /**
+   * close all cleaner thread but have no effect on sink service context
+   */
+  void setIsStoppedToTrue();
+
   /**
    * retrieve sink service status
    * @return true is closed

diff --git a/src/main/java/com/snowflake/kafka/connector/internal/SnowflakeSinkServiceV1.java b/src/main/java/com/snowflake/kafka/connector/internal/SnowflakeSinkServiceV1.java
@@ -174,6 +174,12 @@ public void closeAll()
     pipes.clear();
   }
 
+  @Override
+  public void setIsStoppedToTrue()
+  {
+    this.isStopped = true; // release all cleaner and flusher threads
+  }
+
   @Override
   public boolean isClosed()
   {