Repository: incubator-gobblin Updated Branches: refs/heads/master 1fbe7006d -> 9a05b70da
[GOBBLIN-347] Ensure KafkaPusher is registered with the closer Closes #2206 from beeramsunitha/master Project: http://git-wip-us.apache.org/repos/asf/incubator-gobblin/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-gobblin/commit/9a05b70d Tree: http://git-wip-us.apache.org/repos/asf/incubator-gobblin/tree/9a05b70d Diff: http://git-wip-us.apache.org/repos/asf/incubator-gobblin/diff/9a05b70d Branch: refs/heads/master Commit: 9a05b70dabcef075e8137bce94ae05f934bac1e4 Parents: 1fbe700 Author: Sunitha Beeram <[email protected]> Authored: Thu Dec 14 15:30:25 2017 -0800 Committer: Hung Tran <[email protected]> Committed: Thu Dec 14 15:30:33 2017 -0800 ---------------------------------------------------------------------- .../java/org/apache/gobblin/metrics/kafka/KafkaEventReporter.java | 1 + .../main/java/org/apache/gobblin/metrics/kafka/KafkaReporter.java | 1 + 2 files changed, 2 insertions(+) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-gobblin/blob/9a05b70d/gobblin-modules/gobblin-kafka-common/src/main/java/org/apache/gobblin/metrics/kafka/KafkaEventReporter.java ---------------------------------------------------------------------- diff --git a/gobblin-modules/gobblin-kafka-common/src/main/java/org/apache/gobblin/metrics/kafka/KafkaEventReporter.java b/gobblin-modules/gobblin-kafka-common/src/main/java/org/apache/gobblin/metrics/kafka/KafkaEventReporter.java index b15e96e..804d909 100644 --- a/gobblin-modules/gobblin-kafka-common/src/main/java/org/apache/gobblin/metrics/kafka/KafkaEventReporter.java +++ b/gobblin-modules/gobblin-kafka-common/src/main/java/org/apache/gobblin/metrics/kafka/KafkaEventReporter.java @@ -54,6 +54,7 @@ public class KafkaEventReporter extends EventReporter { 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 http://git-wip-us.apache.org/repos/asf/incubator-gobblin/blob/9a05b70d/gobblin-modules/gobblin-kafka-common/src/main/java/org/apache/gobblin/metrics/kafka/KafkaReporter.java ---------------------------------------------------------------------- diff --git a/gobblin-modules/gobblin-kafka-common/src/main/java/org/apache/gobblin/metrics/kafka/KafkaReporter.java b/gobblin-modules/gobblin-kafka-common/src/main/java/org/apache/gobblin/metrics/kafka/KafkaReporter.java index 40a9fed..35281b5 100644 --- a/gobblin-modules/gobblin-kafka-common/src/main/java/org/apache/gobblin/metrics/kafka/KafkaReporter.java +++ b/gobblin-modules/gobblin-kafka-common/src/main/java/org/apache/gobblin/metrics/kafka/KafkaReporter.java @@ -80,6 +80,7 @@ public class KafkaReporter extends MetricReportReporter { 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 {
