[GOBBLIN-347] Ensure KafkaPusher is registered with the closer

Closes #2206 from beeramsunitha/master
apache · Dec 14, 2017 · 9a05b70 · 9a05b70
1 parent 1fbe700
commit 9a05b70
Show file tree

Hide file tree

Showing 2 changed files with 2 additions and 0 deletions.
diff --git a/...bblin-kafka-common/src/main/java/org/apache/gobblin/metrics/kafka/KafkaEventReporter.java b/...bblin-kafka-common/src/main/java/org/apache/gobblin/metrics/kafka/KafkaEventReporter.java
@@ -54,6 +54,7 @@ public KafkaEventReporter(Builder<?> builder) throws IOException {
         String pusherClassName = builder.pusherClassName.or(PusherUtils.DEFAULT_KAFKA_PUSHER_CLASS_NAME);
         this.kafkaPusher = PusherUtils.getPusher(pusherClassName, builder.brokers, builder.topic, builder.config);
     }
+    this.closer.register(this.kafkaPusher);
   }
 
   @Override

diff --git a/...es/gobblin-kafka-common/src/main/java/org/apache/gobblin/metrics/kafka/KafkaReporter.java b/...es/gobblin-kafka-common/src/main/java/org/apache/gobblin/metrics/kafka/KafkaReporter.java
@@ -80,6 +80,7 @@ protected KafkaReporter(Builder<?> builder, Config config) throws IOException {
 
       this.kafkaPusher = PusherUtils.getPusher(pusherClassName, builder.brokers, builder.topic, Optional.of(kafkaConfig));
     }
+    this.closer.register(this.kafkaPusher);
   }
 
   protected AvroSerializer<MetricReport> createSerializer(SchemaVersionWriter schemaVersionWriter) throws IOException {