You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by rmetzger <gi...@git.apache.org> on 2015/12/12 19:57:32 UTC
[GitHub] flink pull request: [FLINK-3067] Use curator for committing offset...
GitHub user rmetzger opened a pull request:
https://github.com/apache/flink/pull/1451
[FLINK-3067] Use curator for committing offsets to ZK from Kafka
@aljoscha reported issues with our latest FlinkKafkaConnector: It was still throwing NPE.
Therefore, I decided to abandon the [zkclient](https://github.com/sgroschupf/zkclient) and use Apache Curator instead.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/rmetzger/flink flink3067-curator
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/1451.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #1451
----
commit 648ce763fa5537b118e02b47b5c0c1547340a6c7
Author: Robert Metzger <rm...@apache.org>
Date: 2015-12-12T18:44:17Z
[FLINK-3067] Use curator for committing offsets to ZK from Kafka
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3067] Use curator for committing offset...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/1451#discussion_r47536514
--- Diff: flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/ZookeeperOffsetHandler.java ---
@@ -61,30 +61,29 @@ public ZookeeperOffsetHandler(Properties props) {
if (zkConnect == null) {
throw new IllegalArgumentException("Required property 'zookeeper.connect' has not been set");
}
-
- zkClient = new ZkClient(zkConnect,
- Integer.valueOf(props.getProperty("zookeeper.session.timeout.ms", "6000")),
- Integer.valueOf(props.getProperty("zookeeper.connection.timeout.ms", "6000")),
- new ZooKeeperStringSerializer());
+
+ RetryPolicy retryPolicy = new ExponentialBackoffRetry(100, 10);
+ curatorClient = CuratorFrameworkFactory.newClient(zkConnect, retryPolicy);
--- End diff --
Good point. I'm now allowing users to change the default values. (I chose to use the curator defaults rather than the kafka defaults). The retry policy is also configurable.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3067] Use curator for committing offset...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/flink/pull/1451#discussion_r47480028
--- Diff: flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer.java ---
@@ -632,6 +632,7 @@ public void run() {
// ------------ commit current offsets ----------------
// create copy of current offsets
+ //noinspection unchecked
--- End diff --
Is this this is only suppressing IntellIJ warnings or is this meant to be a replacement for the `@SuppressWarnings` annotation?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3067] Use curator for committing offset...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/1451#discussion_r47531635
--- Diff: flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer.java ---
@@ -632,6 +632,7 @@ public void run() {
// ------------ commit current offsets ----------------
// create copy of current offsets
+ //noinspection unchecked
--- End diff --
Its only supressing IntelliJ warnings I think. I don't know if we have any policy regarding this. The annotation will also suppress compiler warnings.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3067] Use curator for committing offset...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/flink/pull/1451#discussion_r47480476
--- Diff: flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/ZookeeperOffsetHandler.java ---
@@ -98,30 +97,42 @@ public void seekFetcherToInitialOffsets(List<KafkaTopicPartitionLeader> partitio
@Override
public void close() throws IOException {
- zkClient.close();
+ curatorClient.close();
}
// ------------------------------------------------------------------------
// Communication with Zookeeper
// ------------------------------------------------------------------------
- public static void setOffsetInZooKeeper(ZkClient zkClient, String groupId, String topic, int partition, long offset) {
- TopicAndPartition tap = new TopicAndPartition(topic, partition);
- ZKGroupTopicDirs topicDirs = new ZKGroupTopicDirs(groupId, tap.topic());
- ZkUtils.updatePersistentPath(zkClient, topicDirs.consumerOffsetDir() + "/" + tap.partition(), Long.toString(offset));
+ public static void setOffsetInZooKeeper(CuratorFramework curatorClient, String groupId, String topic, int partition, long offset) throws Exception {
+ ZKGroupTopicDirs topicDirs = new ZKGroupTopicDirs(groupId, topic);
+ String path = topicDirs.consumerOffsetDir() + "/" + partition;
+ ensureExists(curatorClient, path);
+ byte[] data = Long.toString(offset).getBytes();
+ curatorClient.setData().forPath(path, data);
}
- public static long getOffsetFromZooKeeper(ZkClient zkClient, String groupId, String topic, int partition) {
- TopicAndPartition tap = new TopicAndPartition(topic, partition);
- ZKGroupTopicDirs topicDirs = new ZKGroupTopicDirs(groupId, tap.topic());
-
- scala.Tuple2<Option<String>, Stat> data = ZkUtils.readDataMaybeNull(zkClient,
- topicDirs.consumerOffsetDir() + "/" + tap.partition());
-
- if (data._1().isEmpty()) {
+ public static long getOffsetFromZooKeeper(CuratorFramework curatorClient, String groupId, String topic, int partition) throws Exception {
+ ZKGroupTopicDirs topicDirs = new ZKGroupTopicDirs(groupId, topic);
+ String path = topicDirs.consumerOffsetDir() + "/" + partition;
+ ensureExists(curatorClient, path);
+ byte[] data = curatorClient.getData().forPath(path);
+ if(data == null) {
return OFFSET_NOT_SET;
} else {
- return Long.valueOf(data._1().get());
+ String asString = new String(data);
+ if(asString.length() == 0) {
+ return OFFSET_NOT_SET;
+ } else {
+ return Long.valueOf(asString);
+ }
+ }
+ }
+
+ private static void ensureExists(CuratorFramework curatorClient, String path) throws Exception {
--- End diff --
There is a ensure method in `CuratorFramework`, which does something similar in a retry loop.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3067] Use curator for committing offset...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/1451
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3067] Use curator for committing offset...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/flink/pull/1451#discussion_r47482606
--- Diff: flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/ZookeeperOffsetHandler.java ---
@@ -98,30 +97,42 @@ public void seekFetcherToInitialOffsets(List<KafkaTopicPartitionLeader> partitio
@Override
public void close() throws IOException {
- zkClient.close();
+ curatorClient.close();
}
// ------------------------------------------------------------------------
// Communication with Zookeeper
// ------------------------------------------------------------------------
- public static void setOffsetInZooKeeper(ZkClient zkClient, String groupId, String topic, int partition, long offset) {
- TopicAndPartition tap = new TopicAndPartition(topic, partition);
- ZKGroupTopicDirs topicDirs = new ZKGroupTopicDirs(groupId, tap.topic());
- ZkUtils.updatePersistentPath(zkClient, topicDirs.consumerOffsetDir() + "/" + tap.partition(), Long.toString(offset));
+ public static void setOffsetInZooKeeper(CuratorFramework curatorClient, String groupId, String topic, int partition, long offset) throws Exception {
+ ZKGroupTopicDirs topicDirs = new ZKGroupTopicDirs(groupId, topic);
+ String path = topicDirs.consumerOffsetDir() + "/" + partition;
+ ensureExists(curatorClient, path);
+ byte[] data = Long.toString(offset).getBytes();
--- End diff --
Depending on how often this is called, we could replace it with another way of converting it to the original bytes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3067] Use curator for committing offset...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the pull request:
https://github.com/apache/flink/pull/1451#issuecomment-164477144
I tested the changes on the cluster with a job where I previously saw failures in zkClient. Now it seems to work. So I would give my :+1. Others would probably look at the implementation but that seems to be going on already.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3067] Use curator for committing offset...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on the pull request:
https://github.com/apache/flink/pull/1451#issuecomment-165047710
With Aljoscha's test +1 to merge (minus two trivial comments).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3067] Use curator for committing offset...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/1451#discussion_r47536647
--- Diff: flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/ZookeeperOffsetHandler.java ---
@@ -98,30 +97,42 @@ public void seekFetcherToInitialOffsets(List<KafkaTopicPartitionLeader> partitio
@Override
public void close() throws IOException {
- zkClient.close();
+ curatorClient.close();
}
// ------------------------------------------------------------------------
// Communication with Zookeeper
// ------------------------------------------------------------------------
- public static void setOffsetInZooKeeper(ZkClient zkClient, String groupId, String topic, int partition, long offset) {
- TopicAndPartition tap = new TopicAndPartition(topic, partition);
- ZKGroupTopicDirs topicDirs = new ZKGroupTopicDirs(groupId, tap.topic());
- ZkUtils.updatePersistentPath(zkClient, topicDirs.consumerOffsetDir() + "/" + tap.partition(), Long.toString(offset));
+ public static void setOffsetInZooKeeper(CuratorFramework curatorClient, String groupId, String topic, int partition, long offset) throws Exception {
+ ZKGroupTopicDirs topicDirs = new ZKGroupTopicDirs(groupId, topic);
+ String path = topicDirs.consumerOffsetDir() + "/" + partition;
+ ensureExists(curatorClient, path);
+ byte[] data = Long.toString(offset).getBytes();
--- End diff --
Compared to all the other stuff going on with the Kafka connector, this is probably not an issue (it only happens at the rate of checkpoints).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3067] Use curator for committing offset...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/flink/pull/1451#discussion_r47753669
--- Diff: flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/ZookeeperOffsetHandler.java ---
@@ -42,13 +43,12 @@
private static final Logger LOG = LoggerFactory.getLogger(ZookeeperOffsetHandler.class);
private static final long OFFSET_NOT_SET = FlinkKafkaConsumer.OFFSET_NOT_SET;
-
-
- private final ZkClient zkClient;
-
+
private final String groupId;
-
+ private final CuratorFramework curatorClient;
+
+
--- End diff --
empty line
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3067] Use curator for committing offset...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/flink/pull/1451#discussion_r47480218
--- Diff: flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/ZookeeperOffsetHandler.java ---
@@ -61,30 +61,29 @@ public ZookeeperOffsetHandler(Properties props) {
if (zkConnect == null) {
throw new IllegalArgumentException("Required property 'zookeeper.connect' has not been set");
}
-
- zkClient = new ZkClient(zkConnect,
- Integer.valueOf(props.getProperty("zookeeper.session.timeout.ms", "6000")),
- Integer.valueOf(props.getProperty("zookeeper.connection.timeout.ms", "6000")),
- new ZooKeeperStringSerializer());
+
+ RetryPolicy retryPolicy = new ExponentialBackoffRetry(100, 10);
+ curatorClient = CuratorFrameworkFactory.newClient(zkConnect, retryPolicy);
--- End diff --
Have you checked that the default timeouts of Curator (for session and connection) match the previously configured timeouts?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3067] Use curator for committing offset...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/flink/pull/1451#discussion_r47480714
--- Diff: flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/ZookeeperOffsetHandler.java ---
@@ -98,30 +97,42 @@ public void seekFetcherToInitialOffsets(List<KafkaTopicPartitionLeader> partitio
@Override
public void close() throws IOException {
- zkClient.close();
+ curatorClient.close();
}
// ------------------------------------------------------------------------
// Communication with Zookeeper
// ------------------------------------------------------------------------
- public static void setOffsetInZooKeeper(ZkClient zkClient, String groupId, String topic, int partition, long offset) {
- TopicAndPartition tap = new TopicAndPartition(topic, partition);
- ZKGroupTopicDirs topicDirs = new ZKGroupTopicDirs(groupId, tap.topic());
- ZkUtils.updatePersistentPath(zkClient, topicDirs.consumerOffsetDir() + "/" + tap.partition(), Long.toString(offset));
+ public static void setOffsetInZooKeeper(CuratorFramework curatorClient, String groupId, String topic, int partition, long offset) throws Exception {
+ ZKGroupTopicDirs topicDirs = new ZKGroupTopicDirs(groupId, topic);
+ String path = topicDirs.consumerOffsetDir() + "/" + partition;
+ ensureExists(curatorClient, path);
+ byte[] data = Long.toString(offset).getBytes();
+ curatorClient.setData().forPath(path, data);
}
- public static long getOffsetFromZooKeeper(ZkClient zkClient, String groupId, String topic, int partition) {
- TopicAndPartition tap = new TopicAndPartition(topic, partition);
- ZKGroupTopicDirs topicDirs = new ZKGroupTopicDirs(groupId, tap.topic());
-
- scala.Tuple2<Option<String>, Stat> data = ZkUtils.readDataMaybeNull(zkClient,
- topicDirs.consumerOffsetDir() + "/" + tap.partition());
-
- if (data._1().isEmpty()) {
+ public static long getOffsetFromZooKeeper(CuratorFramework curatorClient, String groupId, String topic, int partition) throws Exception {
+ ZKGroupTopicDirs topicDirs = new ZKGroupTopicDirs(groupId, topic);
+ String path = topicDirs.consumerOffsetDir() + "/" + partition;
+ ensureExists(curatorClient, path);
+ byte[] data = curatorClient.getData().forPath(path);
+ if(data == null) {
--- End diff --
code style missing space (and line 124 as well)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3067] Use curator for committing offset...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/flink/pull/1451#discussion_r47753777
--- Diff: flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer.java ---
@@ -632,6 +632,7 @@ public void run() {
// ------------ commit current offsets ----------------
// create copy of current offsets
+ //noinspection unchecked
--- End diff --
I think you can remove the line after the annotation. I don't think we have a policy about this, but I've seen it in many places.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3067] Use curator for committing offset...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the pull request:
https://github.com/apache/flink/pull/1451#issuecomment-167411820
Good fix, merging this!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---