You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by rm...@apache.org on 2016/11/02 14:21:42 UTC
flink git commit: [FLINK-4945] FlinkKafkaConsumer logs wrong warning
about confirmation for unknown checkpoint
Repository: flink
Updated Branches:
refs/heads/master b4bf99dfc -> 223b0aa0e
[FLINK-4945] FlinkKafkaConsumer logs wrong warning about confirmation for unknown checkpoint
This closes #2706
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/223b0aa0
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/223b0aa0
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/223b0aa0
Branch: refs/heads/master
Commit: 223b0aa0ec9fa0fc53ca823d70e91010643b1ef2
Parents: b4bf99d
Author: Stefan Richter <s....@data-artisans.com>
Authored: Thu Oct 27 18:48:11 2016 +0200
Committer: Robert Metzger <rm...@apache.org>
Committed: Wed Nov 2 12:03:38 2016 +0100
----------------------------------------------------------------------
.../streaming/connectors/kafka/FlinkKafkaConsumerBase.java | 7 ++++---
1 file changed, 4 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/223b0aa0/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java b/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java
index db092f0..5161b35 100644
--- a/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java
+++ b/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java
@@ -346,9 +346,6 @@ public abstract class FlinkKafkaConsumerBase<T> extends RichParallelSourceFuncti
// originally restored offsets or the assigned partitions
if (restoreToOffset != null) {
- // the map cannot be asynchronously updated, because only one checkpoint call can happen
- // on this function at a time: either snapshotState() or notifyCheckpointComplete()
- pendingOffsetsToCommit.put(context.getCheckpointId(), restoreToOffset);
for (Map.Entry<KafkaTopicPartition, Long> kafkaTopicPartitionLongEntry : restoreToOffset.entrySet()) {
offsetsStateForCheckpoint.add(
@@ -360,6 +357,10 @@ public abstract class FlinkKafkaConsumerBase<T> extends RichParallelSourceFuncti
Tuple2.of(subscribedPartition, KafkaTopicPartitionState.OFFSET_NOT_SET));
}
}
+
+ // the map cannot be asynchronously updated, because only one checkpoint call can happen
+ // on this function at a time: either snapshotState() or notifyCheckpointComplete()
+ pendingOffsetsToCommit.put(context.getCheckpointId(), restoreToOffset);
} else {
HashMap<KafkaTopicPartition, Long> currentOffsets = fetcher.snapshotCurrentState();