modified to use shutdownhook instead of finalize()

Telesia · Nov 18, 2020 · f20be45 · f20be45
1 parent 276ed04
commit f20be45
Show file tree

Hide file tree

Showing 2 changed files with 15 additions and 8 deletions.
diff --git a/...core/src/main/java/org/apache/samza/coordinator/metadatastore/CoordinatorStreamStore.java b/...core/src/main/java/org/apache/samza/coordinator/metadatastore/CoordinatorStreamStore.java
@@ -111,6 +111,12 @@ public void init() {
       systemConsumer.start();
       systemProducer.register(SOURCE);
       systemProducer.start();
+      Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+        LOG.info("CoordinatorStreamStore Shut Down Hook thread is closing kafka clients");
+        this.systemProducer.stop();
+        this.systemConsumer.stop();
+        this.systemAdmin.stop();
+      }));
       iterator = new SystemStreamPartitionIterator(systemConsumer, coordinatorSystemStreamPartition);
       readMessagesFromCoordinatorStream();
     } else {
@@ -188,14 +194,6 @@ public void flush() {
     }
   }
 
-  @Override
-  protected void finalize() throws Throwable {
-    super.finalize();
-    systemAdmin.stop();
-    systemProducer.stop();
-    systemConsumer.stop();
-  }
-
   /**
    * <p>
    *   Fetches the metadata of the topic partition of coordinator stream. Registers the oldest offset

diff --git a/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManager.scala b/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManager.scala
@@ -88,6 +88,9 @@ class KafkaCheckpointManager(checkpointSpec: KafkaStreamSpec,
     Preconditions.checkNotNull(systemAdmin)
 
     systemAdmin.start()
+    Runtime.getRuntime.addShutdownHook(new Thread("KafkaCheckPointManagerShutdownHook") {
+      override def run = systemAdmin.stop()
+    })
 
     info(s"Creating checkpoint stream: ${checkpointSpec.getPhysicalName} with " +
       s"partition count: ${checkpointSpec.getPartitionCount}")
@@ -112,6 +115,12 @@ class KafkaCheckpointManager(checkpointSpec: KafkaStreamSpec,
     info(s"Starting the checkpoint SystemConsumer from oldest offset $oldestOffset")
     systemConsumer.register(checkpointSsp, oldestOffset)
     systemConsumer.start()
+    Runtime.getRuntime.addShutdownHook(new Thread("KafkaCheckPointManagerShutdownHook") {
+      override def run(): Unit = {
+        producerRef.get().stop()
+        systemConsumer.stop()
+      }
+    })
   }
 
   /**