Add consumer shutdown hook.
@@ -76,6 +76,8 @@ public class FastenKafkaPlugin implements FastenServerPlugin {
this.connection = new KafkaConsumer<>(consumerProperties);
this.producer = new KafkaProducer<>(producerProperties);
+ handleConsumerShutdown();
+
this.skipOffsets = skipOffsets;
if (writeDirectory != null) {
this.writeDirectory = writeDirectory.endsWith(File.separator)
@@ -369,4 +371,12 @@ public class FastenKafkaPlugin implements FastenServerPlugin {
return statusRecords;
}
+
+ /**
+ * Add hook for graceful consumer shutdown.
+ */
+ private void handleConsumerShutdown() {
+ Thread closeConsumer = new Thread(() -> connection.close());
+ Runtime.getRuntime().addShutdownHook(closeConsumer);
+ }
}
