You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ij...@apache.org on 2017/06/01 20:54:26 UTC
kafka git commit: KAFKA-5345;
Close KafkaClient when streams client is closed
Repository: kafka
Updated Branches:
refs/heads/trunk be392f95d -> 6b9bc3fd2
KAFKA-5345; Close KafkaClient when streams client is closed
Author: Rajini Sivaram <ra...@googlemail.com>
Reviewers: Guozhang Wang <wa...@gmail.com>, Ismael Juma <is...@juma.me.uk>
Closes #3195 from rajinisivaram/KAFKA-5345
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/6b9bc3fd
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/6b9bc3fd
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/6b9bc3fd
Branch: refs/heads/trunk
Commit: 6b9bc3fd26d0e565c1fe86ce46859f9c1a024987
Parents: be392f9
Author: Rajini Sivaram <ra...@googlemail.com>
Authored: Thu Jun 1 21:54:28 2017 +0100
Committer: Ismael Juma <is...@juma.me.uk>
Committed: Thu Jun 1 21:54:28 2017 +0100
----------------------------------------------------------------------
.../streams/processor/internals/StreamsKafkaClient.java | 8 ++++++--
1 file changed, 6 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/6b9bc3fd/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsKafkaClient.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsKafkaClient.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsKafkaClient.java
index b1c3f2b..2d47876 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsKafkaClient.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsKafkaClient.java
@@ -141,8 +141,12 @@ public class StreamsKafkaClient {
}
public void close() throws IOException {
- for (MetricsReporter metricsReporter: this.reporters) {
- metricsReporter.close();
+ try {
+ kafkaClient.close();
+ } finally {
+ for (MetricsReporter metricsReporter: this.reporters) {
+ metricsReporter.close();
+ }
}
}