You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tz...@apache.org on 2019/01/30 14:54:20 UTC
[flink] 15/18: [FLINK-11322] [kafka] Use try-with-resource for
short-living connections in Kafka connectors
This is an automated email from the ASF dual-hosted git repository.
tzulitai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 64cbf2a20178cf8b4f643ba510653881a3e5ce2f
Author: Fokko Driesprong <fo...@godatadriven.com>
AuthorDate: Tue Jan 15 00:41:13 2019 +0100
[FLINK-11322] [kafka] Use try-with-resource for short-living connections in Kafka connectors
This closes #7488.
---
.../connectors/kafka/FlinkKafkaConsumer010.java | 16 ++++++------
.../streaming/connectors/kafka/Kafka08ITCase.java | 29 +++++++++++-----------
.../connectors/kafka/FlinkKafkaConsumer.java | 17 ++++++-------
3 files changed, 31 insertions(+), 31 deletions(-)
diff --git a/flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer010.java b/flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer010.java
index 3508d6d..0831516 100644
--- a/flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer010.java
+++ b/flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer010.java
@@ -243,20 +243,20 @@ public class FlinkKafkaConsumer010<T> extends FlinkKafkaConsumer09<T> {
timestamp);
}
+ final Map<KafkaTopicPartition, Long> result = new HashMap<>(partitions.size());
+
// use a short-lived consumer to fetch the offsets;
// this is ok because this is a one-time operation that happens only on startup
- KafkaConsumer<?, ?> consumer = new KafkaConsumer(properties);
-
- Map<KafkaTopicPartition, Long> result = new HashMap<>(partitions.size());
- for (Map.Entry<TopicPartition, OffsetAndTimestamp> partitionToOffset :
+ try (KafkaConsumer<?, ?> consumer = new KafkaConsumer(properties)) {
+ for (Map.Entry<TopicPartition, OffsetAndTimestamp> partitionToOffset :
consumer.offsetsForTimes(partitionOffsetsRequest).entrySet()) {
- result.put(
- new KafkaTopicPartition(partitionToOffset.getKey().topic(), partitionToOffset.getKey().partition()),
- (partitionToOffset.getValue() == null) ? null : partitionToOffset.getValue().offset());
+ result.put(
+ new KafkaTopicPartition(partitionToOffset.getKey().topic(), partitionToOffset.getKey().partition()),
+ (partitionToOffset.getValue() == null) ? null : partitionToOffset.getValue().offset());
+ }
}
- consumer.close();
return result;
}
}
diff --git a/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08ITCase.java b/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08ITCase.java
index 5af219e..a250d86 100644
--- a/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08ITCase.java
+++ b/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08ITCase.java
@@ -83,9 +83,9 @@ public class Kafka08ITCase extends KafkaConsumerTestBase {
final String topic = writeSequence("invalidOffsetTopic", 20, parallelism, 1);
// set invalid offset:
- CuratorFramework curatorClient = ((KafkaTestEnvironmentImpl) kafkaServer).createCuratorClient();
- ZookeeperOffsetHandler.setOffsetInZooKeeper(curatorClient, standardProps.getProperty("group.id"), topic, 0, 1234);
- curatorClient.close();
+ try (CuratorFramework curatorClient = ((KafkaTestEnvironmentImpl) kafkaServer).createCuratorClient()) {
+ ZookeeperOffsetHandler.setOffsetInZooKeeper(curatorClient, standardProps.getProperty("group.id"), topic, 0, 1234);
+ }
// read from topic
final int valuesCount = 20;
@@ -165,14 +165,14 @@ public class Kafka08ITCase extends KafkaConsumerTestBase {
final Long offset = (long) (Math.random() * Long.MAX_VALUE);
- CuratorFramework curatorFramework = ((KafkaTestEnvironmentImpl) kafkaServer).createCuratorClient();
- kafkaServer.createTestTopic(topicName, 3, 2);
-
- ZookeeperOffsetHandler.setOffsetInZooKeeper(curatorFramework, groupId, topicName, 0, offset);
+ final Long fetchedOffset;
+ try (CuratorFramework curatorFramework = ((KafkaTestEnvironmentImpl) kafkaServer).createCuratorClient()) {
+ kafkaServer.createTestTopic(topicName, 3, 2);
- Long fetchedOffset = ZookeeperOffsetHandler.getOffsetFromZooKeeper(curatorFramework, groupId, topicName, 0);
+ ZookeeperOffsetHandler.setOffsetInZooKeeper(curatorFramework, groupId, topicName, 0, offset);
- curatorFramework.close();
+ fetchedOffset = ZookeeperOffsetHandler.getOffsetFromZooKeeper(curatorFramework, groupId, topicName, 0);
+ }
assertEquals(offset, fetchedOffset);
}
@@ -209,13 +209,14 @@ public class Kafka08ITCase extends KafkaConsumerTestBase {
// read so that the offset can be committed to ZK
readSequence(env, StartupMode.GROUP_OFFSETS, null, null, readProps, parallelism, topicName, 100, 0);
+ final Long o1, o2, o3;
// get the offset
- CuratorFramework curatorFramework = ((KafkaTestEnvironmentImpl) kafkaServer).createCuratorClient();
+ try (CuratorFramework curatorFramework = ((KafkaTestEnvironmentImpl) kafkaServer).createCuratorClient()) {
+ o1 = ZookeeperOffsetHandler.getOffsetFromZooKeeper(curatorFramework, standardProps.getProperty("group.id"), topicName, 0);
+ o2 = ZookeeperOffsetHandler.getOffsetFromZooKeeper(curatorFramework, standardProps.getProperty("group.id"), topicName, 1);
+ o3 = ZookeeperOffsetHandler.getOffsetFromZooKeeper(curatorFramework, standardProps.getProperty("group.id"), topicName, 2);
+ }
- Long o1 = ZookeeperOffsetHandler.getOffsetFromZooKeeper(curatorFramework, standardProps.getProperty("group.id"), topicName, 0);
- Long o2 = ZookeeperOffsetHandler.getOffsetFromZooKeeper(curatorFramework, standardProps.getProperty("group.id"), topicName, 1);
- Long o3 = ZookeeperOffsetHandler.getOffsetFromZooKeeper(curatorFramework, standardProps.getProperty("group.id"), topicName, 2);
- curatorFramework.close();
LOG.info("Got final offsets from zookeeper o1={}, o2={}, o3={}", o1, o2, o3);
// ensure that the offset has been committed
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer.java
index ddbe421..e6f1638 100644
--- a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer.java
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer.java
@@ -274,20 +274,19 @@ public class FlinkKafkaConsumer<T> extends FlinkKafkaConsumerBase<T> {
timestamp);
}
+ final Map<KafkaTopicPartition, Long> result = new HashMap<>(partitions.size());
// use a short-lived consumer to fetch the offsets;
// this is ok because this is a one-time operation that happens only on startup
- KafkaConsumer<?, ?> consumer = new KafkaConsumer(properties);
+ try (KafkaConsumer<?, ?> consumer = new KafkaConsumer(properties)) {
+ for (Map.Entry<TopicPartition, OffsetAndTimestamp> partitionToOffset :
+ consumer.offsetsForTimes(partitionOffsetsRequest).entrySet()) {
- Map<KafkaTopicPartition, Long> result = new HashMap<>(partitions.size());
- for (Map.Entry<TopicPartition, OffsetAndTimestamp> partitionToOffset :
- consumer.offsetsForTimes(partitionOffsetsRequest).entrySet()) {
+ result.put(
+ new KafkaTopicPartition(partitionToOffset.getKey().topic(), partitionToOffset.getKey().partition()),
+ (partitionToOffset.getValue() == null) ? null : partitionToOffset.getValue().offset());
+ }
- result.put(
- new KafkaTopicPartition(partitionToOffset.getKey().topic(), partitionToOffset.getKey().partition()),
- (partitionToOffset.getValue() == null) ? null : partitionToOffset.getValue().offset());
}
-
- consumer.close();
return result;
}