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 2017/11/08 10:50:57 UTC
flink git commit: [FLINK-8001] [kafka] Mark consumer subtask as IDLE
if it doesn't have partitions
Repository: flink
Updated Branches:
refs/heads/release-1.3 6bf24b968 -> b307081a3
[FLINK-8001] [kafka] Mark consumer subtask as IDLE if it doesn't have partitions
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/b307081a
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/b307081a
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/b307081a
Branch: refs/heads/release-1.3
Commit: b307081a3fbb9f245697f4248bc8b37dad961762
Parents: 6bf24b9
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
Authored: Tue Nov 7 20:12:23 2017 +0800
Committer: Tzu-Li (Gordon) Tai <tz...@apache.org>
Committed: Wed Nov 8 18:50:30 2017 +0800
----------------------------------------------------------------------
.../kafka/FlinkKafkaConsumerBase.java | 7 +++----
.../kafka/internals/AbstractFetcher.java | 10 +++++----
.../kafka/internals/AbstractFetcherTest.java | 22 ++++++++++++++++++++
3 files changed, 31 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/b307081a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java
index 6f2e538..eb26b4d 100644
--- a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java
+++ b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java
@@ -37,7 +37,6 @@ import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
-import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.connectors.kafka.config.OffsetCommitMode;
import org.apache.flink.streaming.connectors.kafka.config.OffsetCommitModes;
import org.apache.flink.streaming.connectors.kafka.config.StartupMode;
@@ -483,9 +482,9 @@ public abstract class FlinkKafkaConsumerBase<T> extends RichParallelSourceFuncti
fetcher.runFetchLoop();
}
else {
- // this source never completes, so emit a Long.MAX_VALUE watermark
- // to not block watermark forwarding
- sourceContext.emitWatermark(new Watermark(Long.MAX_VALUE));
+ // this source never emits any records (and therefore also no watermarks),
+ // so we mark this subtask as idle to not block watermark forwarding
+ sourceContext.markAsTemporarilyIdle();
// wait until this is canceled
final Object waitLock = new Object();
http://git-wip-us.apache.org/repos/asf/flink/blob/b307081a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/AbstractFetcher.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/AbstractFetcher.java b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/AbstractFetcher.java
index 10b7530..9337759 100644
--- a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/AbstractFetcher.java
+++ b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/AbstractFetcher.java
@@ -547,6 +547,7 @@ public abstract class AbstractFetcher<T, KPH> {
public void onProcessingTime(long timestamp) throws Exception {
long minAcrossAll = Long.MAX_VALUE;
+ boolean isEffectiveMinAggregation = false;
for (KafkaTopicPartitionStateWithPeriodicWatermarks<?, ?> state : allPartitions) {
// we access the current watermark for the periodic assigners under the state
@@ -556,16 +557,17 @@ public abstract class AbstractFetcher<T, KPH> {
synchronized (state) {
curr = state.getCurrentWatermarkTimestamp();
}
-
+
minAcrossAll = Math.min(minAcrossAll, curr);
+ isEffectiveMinAggregation = true;
}
-
+
// emit next watermark, if there is one
- if (minAcrossAll > lastWatermarkTimestamp) {
+ if (isEffectiveMinAggregation && minAcrossAll > lastWatermarkTimestamp) {
lastWatermarkTimestamp = minAcrossAll;
emitter.emitWatermark(new Watermark(minAcrossAll));
}
-
+
// schedule the next watermark
timerService.registerTimer(timerService.getCurrentProcessingTime() + interval, this);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/b307081a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/internals/AbstractFetcherTest.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/internals/AbstractFetcherTest.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/internals/AbstractFetcherTest.java
index 3a5f83c..81a9ac3 100644
--- a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/internals/AbstractFetcherTest.java
+++ b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/internals/AbstractFetcherTest.java
@@ -318,6 +318,28 @@ public class AbstractFetcherTest {
assertTrue(watermarkTs >= 13L && watermarkTs <= 15L);
}
+ @Test
+ public void testPeriodicWatermarksWithNoSubscribedPartitionsShouldYieldNoWatermarks() throws Exception {
+ Map<KafkaTopicPartition, Long> originalPartitions = new HashMap<>();
+
+ TestSourceContext<Long> sourceContext = new TestSourceContext<>();
+
+ TestProcessingTimeService processingTimeProvider = new TestProcessingTimeService();
+
+ TestFetcher<Long> fetcher = new TestFetcher<>(
+ sourceContext,
+ originalPartitions,
+ new SerializedValue<AssignerWithPeriodicWatermarks<Long>>(new PeriodicTestExtractor()),
+ null, /* punctuated watermarks assigner*/
+ processingTimeProvider,
+ 10);
+
+ processingTimeProvider.setCurrentTime(10);
+ // no partitions; when the periodic watermark emitter fires, no watermark should be emitted
+ assertFalse(sourceContext.hasWatermark());
+ }
+
+
// ------------------------------------------------------------------------
// Test mocks
// ------------------------------------------------------------------------