You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by tzulitai <gi...@git.apache.org> on 2017/03/10 06:23:42 UTC
[GitHub] flink pull request #3505: [backport-1.2] [FLINK-6006] [kafka] Always use com...
GitHub user tzulitai opened a pull request:
https://github.com/apache/flink/pull/3505
[backport-1.2] [FLINK-6006] [kafka] Always use complete restored state in FlinkKafkaConsumer
(This PR is the fix of FLINK-6006 for Flink 1.2)
Previously, the Kafka Consumer performs partition list querying on
restore, and then uses it to filter out restored state of partitions
that doesn't exist in the list.
If in any case the returned partitions list is incomplete (i.e. missing
partitions that existed before perhaps due to temporary ZK / broker
downtimes), then the state of the missing partitions is dropped and
cannot be recovered anymore.
This PR fixes this by always restoring the complete state, without
any sort of filtering. We simply let the consumer fail if assigned
partitions to the consuming threads / Kafka clients are unreachable when
the consumer starts running.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/tzulitai/flink FLINK-6006
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/3505.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 #3505
----
commit d38e3fd4b123cacf4fd6ee3c1baab77e9b8593a0
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
Date: 2017-03-10T06:10:56Z
[FLINK-6006] [kafka] Always use complete restored state in FlinkKafkaConsumer
Previously, the Kafka Consumer performs partition list querying on
restore, and then uses it to filter out restored state of partitions
that doesn't exist in the list.
If in any case the returned partitions list is incomplete (i.e. missing
partitions that existed before perhaps due to temporary ZK / broker
downtimes), then the state of the missing partitions is dropped and
cannot be recovered anymore.
This commit fixes this by always restoring the complete state, without
any sort of filtering. We simply let the consumer fail if assigned
partitions to the consuming threads / Kafka clients are unreachable when
the consumer starts running.
----
---
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 issue #3505: [backport-1.2] [FLINK-6006] [kafka] Always use complete r...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on the issue:
https://github.com/apache/flink/pull/3505
+1 to merge
---
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 issue #3505: [backport-1.2] [FLINK-6006] [kafka] Always use complete r...
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on the issue:
https://github.com/apache/flink/pull/3505
Merging ..
---
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 #3505: [backport-1.2] [FLINK-6006] [kafka] Always use com...
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on a diff in the pull request:
https://github.com/apache/flink/pull/3505#discussion_r105860279
--- Diff: flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/AbstractFetcher.java ---
@@ -194,14 +194,29 @@ protected AbstractFetcher(
/**
* Restores the partition offsets.
+ * The partitions in the provided map of restored partitions to offsets must completely match
+ * the fetcher's subscribed partitions.
*
- * @param snapshotState The offsets for the partitions
+ * @param restoredOffsets The restored offsets for the partitions
+ *
+ * @throws IllegalStateException if the partitions in the provided restored offsets map
+ * cannot completely match the fetcher's subscribed partitions.
*/
- public void restoreOffsets(Map<KafkaTopicPartition, Long> snapshotState) {
- for (KafkaTopicPartitionState<?> partition : allPartitions) {
- Long offset = snapshotState.get(partition.getKafkaTopicPartition());
- if (offset != null) {
- partition.setOffset(offset);
+ public void restoreOffsets(Map<KafkaTopicPartition, Long> restoredOffsets) {
+ if (restoredOffsets.size() != allPartitions.length) {
+ throw new IllegalStateException(
+ "The fetcher was restored with partition offsets that do not " +
+ "match with the subscribed partitions: " + restoredOffsets);
--- End diff --
This would not happen with the changes of this PR.
In `open()`, I've set the `subscribedPartitions` to be exactly the same as the restored partition states. There is no filtering anymore. The `allPartitions` here is basically just the same list, but in their state holder form.
The condition checks exists simply because "setting the fetcher's subscribed partitions" and "restoring start offsets" is 2 separate calls (the former is passed in through the fetcher's constructor, while the latter is provided through the `restoreOffsets` method). I added these checks just to make the fetcher code more self-contained. These exceptions should actually never occur.
I agree this might be a bit confusing for the code reader. In the recent refactorings in `master`, the fetcher's subscribed partitions and start offsets (regardless of if it's a restore or fresh start) setup procedure is more atomic and less confusing in this aspect.
---
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 #3505: [backport-1.2] [FLINK-6006] [kafka] Always use com...
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on a diff in the pull request:
https://github.com/apache/flink/pull/3505#discussion_r105860440
--- Diff: flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java ---
@@ -489,16 +486,15 @@ public void notifyCheckpointComplete(long checkpointId) throws Exception {
// Utilities
// ------------------------------------------------------------------------
- private void assignTopicPartitions(List<KafkaTopicPartition> kafkaTopicPartitions) {
- subscribedPartitions = new ArrayList<>();
-
+ private void assignTopicPartitions() {
if (restoreToOffset != null) {
- for (KafkaTopicPartition kafkaTopicPartition : kafkaTopicPartitions) {
- if (restoreToOffset.containsKey(kafkaTopicPartition)) {
- subscribedPartitions.add(kafkaTopicPartition);
- }
+ subscribedPartitions = new ArrayList<>(restoreToOffset.size());
+ for (Map.Entry<KafkaTopicPartition, Long> restoredPartitionState : restoreToOffset.entrySet()) {
+ subscribedPartitions.add(restoredPartitionState.getKey());
--- End diff --
(note about your comment --->) subscribed partitions will always be completely identical to the restored state, if there is any.
I should actually just change this to `subscribedPartitions = new ArrayList<>(restoreToOffset.keySet);`, to be more clearer.
---
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 #3505: [backport-1.2] [FLINK-6006] [kafka] Always use com...
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai closed the pull request at:
https://github.com/apache/flink/pull/3505
---
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 #3505: [backport-1.2] [FLINK-6006] [kafka] Always use com...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/3505#discussion_r105851319
--- Diff: flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/AbstractFetcher.java ---
@@ -194,14 +194,29 @@ protected AbstractFetcher(
/**
* Restores the partition offsets.
+ * The partitions in the provided map of restored partitions to offsets must completely match
+ * the fetcher's subscribed partitions.
*
- * @param snapshotState The offsets for the partitions
+ * @param restoredOffsets The restored offsets for the partitions
+ *
+ * @throws IllegalStateException if the partitions in the provided restored offsets map
+ * cannot completely match the fetcher's subscribed partitions.
*/
- public void restoreOffsets(Map<KafkaTopicPartition, Long> snapshotState) {
- for (KafkaTopicPartitionState<?> partition : allPartitions) {
- Long offset = snapshotState.get(partition.getKafkaTopicPartition());
- if (offset != null) {
- partition.setOffset(offset);
+ public void restoreOffsets(Map<KafkaTopicPartition, Long> restoredOffsets) {
+ if (restoredOffsets.size() != allPartitions.length) {
+ throw new IllegalStateException(
+ "The fetcher was restored with partition offsets that do not " +
+ "match with the subscribed partitions: " + restoredOffsets);
--- End diff --
I'm not sure if this check is too strict. I think its easily possible in Kafka to add new partitions to a topic.
If a user does this, they can never restore their Kafka consumers.
---
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 #3505: [backport-1.2] [FLINK-6006] [kafka] Always use com...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/3505#discussion_r105869364
--- Diff: flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/AbstractFetcher.java ---
@@ -194,14 +194,29 @@ protected AbstractFetcher(
/**
* Restores the partition offsets.
+ * The partitions in the provided map of restored partitions to offsets must completely match
+ * the fetcher's subscribed partitions.
*
- * @param snapshotState The offsets for the partitions
+ * @param restoredOffsets The restored offsets for the partitions
+ *
+ * @throws IllegalStateException if the partitions in the provided restored offsets map
+ * cannot completely match the fetcher's subscribed partitions.
*/
- public void restoreOffsets(Map<KafkaTopicPartition, Long> snapshotState) {
- for (KafkaTopicPartitionState<?> partition : allPartitions) {
- Long offset = snapshotState.get(partition.getKafkaTopicPartition());
- if (offset != null) {
- partition.setOffset(offset);
+ public void restoreOffsets(Map<KafkaTopicPartition, Long> restoredOffsets) {
+ if (restoredOffsets.size() != allPartitions.length) {
+ throw new IllegalStateException(
+ "The fetcher was restored with partition offsets that do not " +
+ "match with the subscribed partitions: " + restoredOffsets);
--- End diff --
I see. Thank you.
---
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.
---