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/07/22 12:01:54 UTC
[1/6] flink git commit: [FLINK-7143] [kafka] Fix indeterminate
partition assignment in FlinkKafkaConsumer
Repository: flink
Updated Branches:
refs/heads/release-1.3 6856875e7 -> cf70255d4
[FLINK-7143] [kafka] Fix indeterminate partition assignment in FlinkKafkaConsumer
Apart from fixing the previous incorrect, indeterministic assignment
logic, this commit also adds an explicitly defined method that properly
states a strict contract for the assignment, instead of just relying on
some hashCode implementation that doesn't convey this contract as well
as the importance of the assignment's deterministic characteristic well.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/6e0d90cc
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/6e0d90cc
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/6e0d90cc
Branch: refs/heads/release-1.3
Commit: 6e0d90ccfd9c457fd99add5833cd7bf6c976d6b8
Parents: 6856875
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
Authored: Wed Jul 12 02:03:01 2017 +0900
Committer: Tzu-Li (Gordon) Tai <tz...@apache.org>
Committed: Sat Jul 22 19:35:53 2017 +0800
----------------------------------------------------------------------
.../kafka/FlinkKafkaConsumerBase.java | 16 +--
.../internals/KafkaTopicPartitionAssigner.java | 56 +++++++++
.../KafkaConsumerPartitionAssignmentTest.java | 125 ++++++++++++++++---
3 files changed, 173 insertions(+), 24 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/6e0d90cc/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 87bedce..8dbca72 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
@@ -42,6 +42,7 @@ import org.apache.flink.streaming.connectors.kafka.config.OffsetCommitModes;
import org.apache.flink.streaming.connectors.kafka.config.StartupMode;
import org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher;
import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
+import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartitionAssigner;
import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartitionStateSentinel;
import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema;
import org.apache.flink.util.Preconditions;
@@ -698,10 +699,11 @@ public abstract class FlinkKafkaConsumerBase<T> extends RichParallelSourceFuncti
StartupMode startupMode,
Map<KafkaTopicPartition, Long> specificStartupOffsets) {
- for (int i = 0; i < kafkaTopicPartitions.size(); i++) {
- if (i % numParallelSubtasks == indexOfThisSubtask) {
+ for (KafkaTopicPartition kafkaTopicPartition : kafkaTopicPartitions) {
+ // only handle partitions that this subtask should subscribe to
+ if (KafkaTopicPartitionAssigner.assign(kafkaTopicPartition, numParallelSubtasks) == indexOfThisSubtask) {
if (startupMode != StartupMode.SPECIFIC_OFFSETS) {
- subscribedPartitionsToStartOffsets.put(kafkaTopicPartitions.get(i), startupMode.getStateSentinel());
+ subscribedPartitionsToStartOffsets.put(kafkaTopicPartition, startupMode.getStateSentinel());
} else {
if (specificStartupOffsets == null) {
throw new IllegalArgumentException(
@@ -709,15 +711,13 @@ public abstract class FlinkKafkaConsumerBase<T> extends RichParallelSourceFuncti
", but no specific offsets were specified");
}
- KafkaTopicPartition partition = kafkaTopicPartitions.get(i);
-
- Long specificOffset = specificStartupOffsets.get(partition);
+ Long specificOffset = specificStartupOffsets.get(kafkaTopicPartition);
if (specificOffset != null) {
// since the specified offsets represent the next record to read, we subtract
// it by one so that the initial state of the consumer will be correct
- subscribedPartitionsToStartOffsets.put(partition, specificOffset - 1);
+ subscribedPartitionsToStartOffsets.put(kafkaTopicPartition, specificOffset - 1);
} else {
- subscribedPartitionsToStartOffsets.put(partition, KafkaTopicPartitionStateSentinel.GROUP_OFFSET);
+ subscribedPartitionsToStartOffsets.put(kafkaTopicPartition, KafkaTopicPartitionStateSentinel.GROUP_OFFSET);
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6e0d90cc/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/KafkaTopicPartitionAssigner.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/KafkaTopicPartitionAssigner.java b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/KafkaTopicPartitionAssigner.java
new file mode 100644
index 0000000..944630f
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/KafkaTopicPartitionAssigner.java
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.connectors.kafka.internals;
+
+/**
+ * Utility for assigning Kafka partitions to consumer subtasks.
+ */
+public class KafkaTopicPartitionAssigner {
+
+ /**
+ * Returns the index of the target subtask that a specific Kafka partition should be
+ * assigned to.
+ *
+ * <p>The resulting distribution of partitions of a single topic has the following contract:
+ * <ul>
+ * <li>1. Uniformly distributed across subtasks</li>
+ * <li>2. Partitions are round-robin distributed (strictly clockwise w.r.t. ascending
+ * subtask indices) by using the partition id as the offset from a starting index
+ * (i.e., the index of the subtask which partition 0 of the topic will be assigned to,
+ * determined using the topic name).</li>
+ * </ul>
+ *
+ * <p>The above contract is crucial and cannot be broken. Consumer subtasks rely on this
+ * contract to locally filter out partitions that it should not subscribe to, guaranteeing
+ * that all partitions of a single topic will always be assigned to some subtask in a
+ * uniformly distributed manner.
+ *
+ * @param partition the Kafka partition
+ * @param numParallelSubtasks total number of parallel subtasks
+ *
+ * @return index of the target subtask that the Kafka partition should be assigned to.
+ */
+ public static int assign(KafkaTopicPartition partition, int numParallelSubtasks) {
+ int startIndex = ((partition.getTopic().hashCode() * 31) & 0x7FFFFFFF) % numParallelSubtasks;
+
+ // here, the assumption is that the id of Kafka partitions are always ascending
+ // starting from 0, and therefore can be used directly as the offset clockwise from the start index
+ return (startIndex + partition.getPartition()) % numParallelSubtasks;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/6e0d90cc/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerPartitionAssignmentTest.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerPartitionAssignmentTest.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerPartitionAssignmentTest.java
index c24640d..dd19c21 100644
--- a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerPartitionAssignmentTest.java
+++ b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerPartitionAssignmentTest.java
@@ -21,6 +21,7 @@ package org.apache.flink.streaming.connectors.kafka;
import org.apache.flink.streaming.connectors.kafka.config.StartupMode;
import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
+import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartitionAssigner;
import org.junit.Test;
import java.util.ArrayList;
@@ -44,17 +45,22 @@ public class KafkaConsumerPartitionAssignmentTest {
public void testPartitionsEqualConsumers() {
try {
List<KafkaTopicPartition> inPartitions = Arrays.asList(
- new KafkaTopicPartition("test-topic", 4),
- new KafkaTopicPartition("test-topic", 52),
- new KafkaTopicPartition("test-topic", 17),
- new KafkaTopicPartition("test-topic", 1));
+ new KafkaTopicPartition("test-topic", 0),
+ new KafkaTopicPartition("test-topic", 1),
+ new KafkaTopicPartition("test-topic", 2),
+ new KafkaTopicPartition("test-topic", 3));
- for (int i = 0; i < inPartitions.size(); i++) {
+ int numSubtasks = inPartitions.size();
+
+ // get the start index; the assertions below will fail if the assignment logic does not meet correct contracts
+ int startIndex = KafkaTopicPartitionAssigner.assign(inPartitions.get(0), numSubtasks);
+
+ for (int subtaskIndex = 0; subtaskIndex < numSubtasks; subtaskIndex++) {
Map<KafkaTopicPartition, Long> subscribedPartitionsToStartOffsets = new HashMap<>();
FlinkKafkaConsumerBase.initializeSubscribedPartitionsToStartOffsets(
subscribedPartitionsToStartOffsets,
inPartitions,
- i,
+ subtaskIndex,
inPartitions.size(),
StartupMode.GROUP_OFFSETS,
null);
@@ -63,6 +69,9 @@ public class KafkaConsumerPartitionAssignmentTest {
assertEquals(1, subscribedPartitions.size());
assertTrue(contains(inPartitions, subscribedPartitions.get(0).getPartition()));
+
+ // assert that round-robin distribution is used starting clockwise from the start index
+ assertEquals(getExpectedSubtaskIndex(subscribedPartitions.get(0), startIndex, numSubtasks), subtaskIndex);
}
}
catch (Exception e) {
@@ -74,7 +83,7 @@ public class KafkaConsumerPartitionAssignmentTest {
@Test
public void testMultiplePartitionsPerConsumers() {
try {
- final int[] partitionIDs = {4, 52, 17, 1, 2, 3, 89, 42, 31, 127, 14};
+ final int[] partitionIDs = {0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10};
final List<KafkaTopicPartition> partitions = new ArrayList<>();
final Set<KafkaTopicPartition> allPartitions = new HashSet<>();
@@ -89,12 +98,15 @@ public class KafkaConsumerPartitionAssignmentTest {
final int minPartitionsPerConsumer = partitions.size() / numConsumers;
final int maxPartitionsPerConsumer = partitions.size() / numConsumers + 1;
- for (int i = 0; i < numConsumers; i++) {
+ // get the start index; the assertions below will fail if the assignment logic does not meet correct contracts
+ int startIndex = KafkaTopicPartitionAssigner.assign(partitions.get(0), numConsumers);
+
+ for (int subtaskIndex = 0; subtaskIndex < numConsumers; subtaskIndex++) {
Map<KafkaTopicPartition, Long> subscribedPartitionsToStartOffsets = new HashMap<>();
FlinkKafkaConsumerBase.initializeSubscribedPartitionsToStartOffsets(
subscribedPartitionsToStartOffsets,
partitions,
- i,
+ subtaskIndex,
numConsumers,
StartupMode.GROUP_OFFSETS,
null);
@@ -107,6 +119,9 @@ public class KafkaConsumerPartitionAssignmentTest {
for (KafkaTopicPartition p : subscribedPartitions) {
// check that the element was actually contained
assertTrue(allPartitions.remove(p));
+
+ // assert that round-robin distribution is used starting clockwise from the start index
+ assertEquals(getExpectedSubtaskIndex(p, startIndex, numConsumers), subtaskIndex);
}
}
@@ -123,22 +138,25 @@ public class KafkaConsumerPartitionAssignmentTest {
public void testPartitionsFewerThanConsumers() {
try {
List<KafkaTopicPartition> inPartitions = Arrays.asList(
- new KafkaTopicPartition("test-topic", 4),
- new KafkaTopicPartition("test-topic", 52),
- new KafkaTopicPartition("test-topic", 17),
- new KafkaTopicPartition("test-topic", 1));
+ new KafkaTopicPartition("test-topic", 0),
+ new KafkaTopicPartition("test-topic", 1),
+ new KafkaTopicPartition("test-topic", 2),
+ new KafkaTopicPartition("test-topic", 3));
final Set<KafkaTopicPartition> allPartitions = new HashSet<>();
allPartitions.addAll(inPartitions);
final int numConsumers = 2 * inPartitions.size() + 3;
- for (int i = 0; i < numConsumers; i++) {
+ // get the start index; the assertions below will fail if the assignment logic does not meet correct contracts
+ int startIndex = KafkaTopicPartitionAssigner.assign(inPartitions.get(0), numConsumers);
+
+ for (int subtaskIndex = 0; subtaskIndex < numConsumers; subtaskIndex++) {
Map<KafkaTopicPartition, Long> subscribedPartitionsToStartOffsets = new HashMap<>();
FlinkKafkaConsumerBase.initializeSubscribedPartitionsToStartOffsets(
subscribedPartitionsToStartOffsets,
inPartitions,
- i,
+ subtaskIndex,
numConsumers,
StartupMode.GROUP_OFFSETS,
null);
@@ -150,6 +168,9 @@ public class KafkaConsumerPartitionAssignmentTest {
for (KafkaTopicPartition p : subscribedPartitions) {
// check that the element was actually contained
assertTrue(allPartitions.remove(p));
+
+ // assert that round-robin distribution is used starting clockwise from the start index
+ assertEquals(getExpectedSubtaskIndex(p, startIndex, numConsumers), subtaskIndex);
}
}
@@ -195,7 +216,7 @@ public class KafkaConsumerPartitionAssignmentTest {
@Test
public void testGrowingPartitionsRemainsStable() {
try {
- final int[] newPartitionIDs = {4, 52, 17, 1, 2, 3, 89, 42, 31, 127, 14};
+ final int[] newPartitionIDs = {0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10};
List<KafkaTopicPartition> newPartitions = new ArrayList<>();
for (int p : newPartitionIDs) {
@@ -253,19 +274,25 @@ public class KafkaConsumerPartitionAssignmentTest {
assertTrue(subscribedPartitions3.size() >= minInitialPartitionsPerConsumer);
assertTrue(subscribedPartitions3.size() <= maxInitialPartitionsPerConsumer);
+ // get the start index; the assertions below will fail if the assignment logic does not meet correct contracts
+ int startIndex = KafkaTopicPartitionAssigner.assign(initialPartitions.get(0), numConsumers);
+
for (KafkaTopicPartition p : subscribedPartitions1) {
// check that the element was actually contained
assertTrue(allInitialPartitions.remove(p));
+ assertEquals(getExpectedSubtaskIndex(p, startIndex, numConsumers), 0);
}
for (KafkaTopicPartition p : subscribedPartitions2) {
// check that the element was actually contained
assertTrue(allInitialPartitions.remove(p));
+ assertEquals(getExpectedSubtaskIndex(p, startIndex, numConsumers), 1);
}
for (KafkaTopicPartition p : subscribedPartitions3) {
// check that the element was actually contained
assertTrue(allInitialPartitions.remove(p));
+ assertEquals(getExpectedSubtaskIndex(p, startIndex, numConsumers), 2);
}
// all partitions must have been assigned
@@ -325,14 +352,17 @@ public class KafkaConsumerPartitionAssignmentTest {
for (KafkaTopicPartition p : subscribedPartitions1New) {
// check that the element was actually contained
assertTrue(allNewPartitions.remove(p));
+ assertEquals(getExpectedSubtaskIndex(p, startIndex, numConsumers), 0);
}
for (KafkaTopicPartition p : subscribedPartitions2New) {
// check that the element was actually contained
assertTrue(allNewPartitions.remove(p));
+ assertEquals(getExpectedSubtaskIndex(p, startIndex, numConsumers), 1);
}
for (KafkaTopicPartition p : subscribedPartitions3New) {
// check that the element was actually contained
assertTrue(allNewPartitions.remove(p));
+ assertEquals(getExpectedSubtaskIndex(p, startIndex, numConsumers), 2);
}
// all partitions must have been assigned
@@ -344,6 +374,69 @@ public class KafkaConsumerPartitionAssignmentTest {
}
}
+ @Test
+ public void testDeterministicAssignmentWithDifferentFetchedPartitionOrdering() {
+ final int numSubtasks = 4;
+
+ try {
+ List<KafkaTopicPartition> inPartitions = Arrays.asList(
+ new KafkaTopicPartition("test-topic", 0),
+ new KafkaTopicPartition("test-topic", 1),
+ new KafkaTopicPartition("test-topic", 2),
+ new KafkaTopicPartition("test-topic", 3),
+ new KafkaTopicPartition("test-topic2", 0),
+ new KafkaTopicPartition("test-topic2", 1));
+
+ List<KafkaTopicPartition> inPartitionsOutOfOrder = Arrays.asList(
+ new KafkaTopicPartition("test-topic", 3),
+ new KafkaTopicPartition("test-topic", 1),
+ new KafkaTopicPartition("test-topic2", 1),
+ new KafkaTopicPartition("test-topic", 0),
+ new KafkaTopicPartition("test-topic2", 0),
+ new KafkaTopicPartition("test-topic", 2));
+
+ for (int subtaskIndex = 0; subtaskIndex < numSubtasks; subtaskIndex++) {
+ Map<KafkaTopicPartition, Long> subscribedPartitionsToStartOffsets = new HashMap<>();
+ Map<KafkaTopicPartition, Long> subscribedPartitionsToStartOffsetsOutOfOrder = new HashMap<>();
+
+ FlinkKafkaConsumerBase.initializeSubscribedPartitionsToStartOffsets(
+ subscribedPartitionsToStartOffsets,
+ inPartitions,
+ subtaskIndex,
+ inPartitions.size(),
+ StartupMode.GROUP_OFFSETS,
+ null);
+
+ FlinkKafkaConsumerBase.initializeSubscribedPartitionsToStartOffsets(
+ subscribedPartitionsToStartOffsetsOutOfOrder,
+ inPartitionsOutOfOrder,
+ subtaskIndex,
+ inPartitions.size(),
+ StartupMode.GROUP_OFFSETS,
+ null);
+
+ // the subscribed partitions should be identical, regardless of the input partition ordering
+ assertEquals(subscribedPartitionsToStartOffsets, subscribedPartitionsToStartOffsetsOutOfOrder);
+ }
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ /**
+ * Utility method that determines the expected subtask index a partition should be assigned to,
+ * depending on the start index and using the partition id as the offset from that start index
+ * in clockwise direction.
+ *
+ * <p>The expectation is based on the distribution contract of
+ * {@link KafkaTopicPartitionAssigner#assign(KafkaTopicPartition, int)}.
+ */
+ private static int getExpectedSubtaskIndex(KafkaTopicPartition partition, int startIndex, int numSubtasks) {
+ return (startIndex + partition.getPartition()) % numSubtasks;
+ }
+
private boolean contains(List<KafkaTopicPartition> inPartitions, int partition) {
for (KafkaTopicPartition ktp : inPartitions) {
if (ktp.getPartition() == partition) {
[4/6] flink git commit: [FLINK-7143] [kafka] Add test for Kafka
Consumer rescaling
Posted by tz...@apache.org.
[FLINK-7143] [kafka] Add test for Kafka Consumer rescaling
This verifies that the consumer always correctly knows whether it is
restored or not and is not affected by changes in the partitions as
reported by Kafka.
Previously, operator state reshuffling could lead to partitions being
subscribed to multiple times.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/3369cfe2
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/3369cfe2
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/3369cfe2
Branch: refs/heads/release-1.3
Commit: 3369cfe200bf2cb7ed04caf19d8599075e4cfe21
Parents: ead25aa
Author: Aljoscha Krettek <al...@gmail.com>
Authored: Tue Jul 18 11:57:46 2017 +0200
Committer: Tzu-Li (Gordon) Tai <tz...@apache.org>
Committed: Sat Jul 22 19:37:49 2017 +0800
----------------------------------------------------------------------
.../kafka/FlinkKafkaConsumerBaseTest.java | 164 ++++++++++++++++++-
1 file changed, 155 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/3369cfe2/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseTest.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseTest.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseTest.java
index ccf2ed2..31883ce 100644
--- a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseTest.java
+++ b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseTest.java
@@ -24,16 +24,23 @@ import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.OperatorStateStore;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.state.OperatorStateHandle;
import org.apache.flink.runtime.state.StateInitializationContext;
import org.apache.flink.runtime.state.StateSnapshotContextSynchronousImpl;
+import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.api.operators.StreamSource;
import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
import org.apache.flink.streaming.connectors.kafka.config.OffsetCommitMode;
import org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher;
import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.OperatorStateHandles;
+import org.apache.flink.streaming.util.AbstractStreamOperatorTestHarness;
import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema;
+import org.apache.flink.util.Preconditions;
import org.apache.flink.util.SerializedValue;
import org.junit.Assert;
import org.junit.Test;
@@ -51,8 +58,16 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
+import static org.hamcrest.Matchers.everyItem;
+import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.collection.IsIn.isIn;
+import static org.hamcrest.collection.IsMapContaining.hasEntry;
+import static org.hamcrest.collection.IsMapContaining.hasKey;
+import static org.hamcrest.core.IsNot.not;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.Matchers.anyMap;
@@ -532,6 +547,118 @@ public class FlinkKafkaConsumerBaseTest {
verify(fetcher, never()).commitInternalOffsetsToKafka(anyMap()); // not offsets should be committed
}
+ @Test
+ public void testScaleUp() throws Exception {
+ testRescaling(5, 2, 15, 1000);
+ }
+
+ @Test
+ public void testScaleDown() throws Exception {
+ testRescaling(5, 10, 2, 1);
+ }
+
+ /**
+ * Tests whether the Kafka consumer behaves correctly when scaling the parallelism up/down,
+ * which means that operator state is being reshuffled.
+ *
+ * <p>This also verifies that a restoring source is always impervious to changes in the list
+ * of topics fetched from Kafka.
+ */
+ void testRescaling(
+ final int initialParallelism,
+ final int numPartitions,
+ final int restoredParallelism,
+ final int restoredNumPartitions) throws Exception {
+
+ List<KafkaTopicPartition> mockFetchedPartitionsOnStartup = new ArrayList<>();
+ for (int i = 0; i < numPartitions; i++) {
+ mockFetchedPartitionsOnStartup.add(new KafkaTopicPartition("test-topic", i));
+ }
+
+ DummyFlinkKafkaConsumer<String>[] consumers =
+ new DummyFlinkKafkaConsumer[initialParallelism];
+
+ AbstractStreamOperatorTestHarness<String>[] testHarnesses=
+ new AbstractStreamOperatorTestHarness[initialParallelism];
+
+
+ for (int i = 0; i < initialParallelism; i++) {
+ consumers[i] = new DummyFlinkKafkaConsumer<>(
+ mockFetchedPartitionsOnStartup);
+ testHarnesses[i] = createTestHarness(consumers[i], initialParallelism, i);
+
+ // initializeState() is always called, null signals that we didn't restore
+ testHarnesses[i].initializeState(null);
+ testHarnesses[i].open();
+ }
+
+ Map<KafkaTopicPartition, Long> globalSubscribedPartitions = new HashMap<>();
+
+ for (int i = 0; i < initialParallelism; i++) {
+ Map<KafkaTopicPartition, Long> subscribedPartitions =
+ consumers[i].getSubscribedPartitionsToStartOffsets();
+
+ // make sure that no one else is subscribed to these partitions
+ for (KafkaTopicPartition partition : subscribedPartitions.keySet()) {
+ assertThat(globalSubscribedPartitions, not(hasKey(partition)));
+ }
+ globalSubscribedPartitions.putAll(subscribedPartitions);
+ }
+
+
+ assertThat(globalSubscribedPartitions.values(), hasSize(numPartitions));
+ assertThat(mockFetchedPartitionsOnStartup, everyItem(isIn(globalSubscribedPartitions.keySet())));
+
+ OperatorStateHandles[] state = new OperatorStateHandles[initialParallelism];
+
+ for (int i = 0; i < initialParallelism; i++) {
+ state[i] = testHarnesses[i].snapshot(0, 0);
+ }
+
+ OperatorStateHandles mergedState = AbstractStreamOperatorTestHarness.repackageState(state);
+
+ // -----------------------------------------------------------------------------------------
+ // restore
+
+ List<KafkaTopicPartition> mockFetchedPartitionsAfterRestore = new ArrayList<>();
+ for (int i = 0; i < restoredNumPartitions; i++) {
+ mockFetchedPartitionsAfterRestore.add(new KafkaTopicPartition("test-topic", i));
+ }
+
+ DummyFlinkKafkaConsumer<String>[] restoredConsumers =
+ new DummyFlinkKafkaConsumer[restoredParallelism];
+
+ AbstractStreamOperatorTestHarness<String>[] restoredTestHarnesses =
+ new AbstractStreamOperatorTestHarness[restoredParallelism];
+
+
+ for (int i = 0; i < restoredParallelism; i++) {
+ restoredConsumers[i] = new DummyFlinkKafkaConsumer<>(
+ mockFetchedPartitionsAfterRestore);
+ restoredTestHarnesses[i] = createTestHarness(restoredConsumers[i], restoredParallelism, i);
+
+ // initializeState() is always called, null signals that we didn't restore
+ restoredTestHarnesses[i].initializeState(mergedState);
+ restoredTestHarnesses[i].open();
+ }
+
+ Map<KafkaTopicPartition, Long> restoredGlobalSubscribedPartitions = new HashMap<>();
+
+ for (int i = 0; i < restoredParallelism; i++) {
+ Map<KafkaTopicPartition, Long> subscribedPartitions =
+ restoredConsumers[i].getSubscribedPartitionsToStartOffsets();
+
+ // make sure that no one else is subscribed to these partitions
+ for (KafkaTopicPartition partition : subscribedPartitions.keySet()) {
+ assertThat(restoredGlobalSubscribedPartitions, not(hasKey(partition)));
+ }
+ restoredGlobalSubscribedPartitions.putAll(subscribedPartitions);
+ }
+
+ assertThat(restoredGlobalSubscribedPartitions.values(), hasSize(numPartitions));
+ assertThat(mockFetchedPartitionsOnStartup, everyItem(isIn(restoredGlobalSubscribedPartitions.keySet())));
+ }
+
// ------------------------------------------------------------------------
private static <T> FlinkKafkaConsumerBase<T> getConsumer(
@@ -557,6 +684,19 @@ public class FlinkKafkaConsumerBaseTest {
return consumer;
}
+ private static <T> AbstractStreamOperatorTestHarness<T> createTestHarness(
+ SourceFunction<T> source, int numSubtasks, int subtaskIndex) throws Exception {
+
+ AbstractStreamOperatorTestHarness<T> testHarness =
+ new AbstractStreamOperatorTestHarness<>(
+ new StreamSource<>(source), Short.MAX_VALUE / 2, numSubtasks, subtaskIndex);
+
+ testHarness.setTimeCharacteristic(TimeCharacteristic.EventTime);
+
+ return testHarness;
+ }
+
+
// ------------------------------------------------------------------------
private static class DummyFlinkKafkaConsumer<T> extends FlinkKafkaConsumerBase<T> {
@@ -564,26 +704,33 @@ public class FlinkKafkaConsumerBaseTest {
boolean isAutoCommitEnabled = false;
- @SuppressWarnings("unchecked")
+ private List<KafkaTopicPartition> mockFetchedPartitions;
+
public DummyFlinkKafkaConsumer() {
+ this(Collections.<KafkaTopicPartition>emptyList());
+ }
+
+ @SuppressWarnings("unchecked")
+ public DummyFlinkKafkaConsumer(List<KafkaTopicPartition> mockFetchedPartitions) {
super(Arrays.asList("dummy-topic"), (KeyedDeserializationSchema < T >) mock(KeyedDeserializationSchema.class));
+ this.mockFetchedPartitions = Preconditions.checkNotNull(mockFetchedPartitions);
}
@Override
@SuppressWarnings("unchecked")
protected AbstractFetcher<T, ?> createFetcher(
- SourceContext<T> sourceContext,
- Map<KafkaTopicPartition, Long> thisSubtaskPartitionsWithStartOffsets,
- SerializedValue<AssignerWithPeriodicWatermarks<T>> watermarksPeriodic,
- SerializedValue<AssignerWithPunctuatedWatermarks<T>> watermarksPunctuated,
- StreamingRuntimeContext runtimeContext,
- OffsetCommitMode offsetCommitMode) throws Exception {
+ SourceContext<T> sourceContext,
+ Map<KafkaTopicPartition, Long> thisSubtaskPartitionsWithStartOffsets,
+ SerializedValue<AssignerWithPeriodicWatermarks<T>> watermarksPeriodic,
+ SerializedValue<AssignerWithPunctuatedWatermarks<T>> watermarksPunctuated,
+ StreamingRuntimeContext runtimeContext,
+ OffsetCommitMode offsetCommitMode) throws Exception {
return mock(AbstractFetcher.class);
}
@Override
protected List<KafkaTopicPartition> getKafkaPartitions(List<String> topics) {
- return Collections.emptyList();
+ return mockFetchedPartitions;
}
@Override
@@ -595,7 +742,6 @@ public class FlinkKafkaConsumerBaseTest {
this.isAutoCommitEnabled = isAutoCommitEnabled;
}
}
-
private static final class TestingListState<T> implements ListState<T> {
private final List<T> list = new ArrayList<>();
[5/6] flink git commit: [FLINK-7195] [kafka] Remove partition list
querying when restoring state in FlinkKafkaConsumer
Posted by tz...@apache.org.
[FLINK-7195] [kafka] Remove partition list querying when restoring state in FlinkKafkaConsumer
Previously, querying the partition list and using it to filter out
restored partition states is problematic since the queried partition
list may be missing partitions due to temporary downtime of Kafka
brokers. Effectively, this caused the potential dropping of state on
restores.
This commit fixes this by completely removing partition querying if
we're restoring state (as notified by
FunctionInitializationContext.isRestored()). The subscribed partitions
will always be exactly what the restored state contains.
This closes #4357.
This closes #4344.
This closes #4301.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/e9d493db
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/e9d493db
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/e9d493db
Branch: refs/heads/release-1.3
Commit: e9d493db4a29f3c781644cec99fbd37a2f51806c
Parents: 3369cfe
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
Authored: Fri Jul 14 19:51:03 2017 +0800
Committer: Tzu-Li (Gordon) Tai <tz...@apache.org>
Committed: Sat Jul 22 19:41:04 2017 +0800
----------------------------------------------------------------------
.../kafka/FlinkKafkaConsumerBase.java | 27 +--
.../kafka/FlinkKafkaConsumerBaseTest.java | 108 +++++++++-
.../KafkaConsumerPartitionAssignmentTest.java | 208 +++++++++----------
3 files changed, 213 insertions(+), 130 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/e9d493db/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 199755c..718f1f2 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
@@ -351,24 +351,17 @@ public abstract class FlinkKafkaConsumerBase<T> extends RichParallelSourceFuncti
getRuntimeContext().getIndexOfThisSubtask());
}
- // initialize subscribed partitions
- List<KafkaTopicPartition> kafkaTopicPartitions = getKafkaPartitions(topics);
- Preconditions.checkNotNull(kafkaTopicPartitions, "TopicPartitions must not be null.");
-
- subscribedPartitionsToStartOffsets = new HashMap<>(kafkaTopicPartitions.size());
-
if (restored) {
- for (KafkaTopicPartition kafkaTopicPartition : kafkaTopicPartitions) {
- if (restoredState.containsKey(kafkaTopicPartition)) {
- subscribedPartitionsToStartOffsets.put(kafkaTopicPartition, restoredState.get(kafkaTopicPartition));
- }
- }
+ subscribedPartitionsToStartOffsets = restoredState;
LOG.info("Consumer subtask {} will start reading {} partitions with offsets in restored state: {}",
getRuntimeContext().getIndexOfThisSubtask(), subscribedPartitionsToStartOffsets.size(), subscribedPartitionsToStartOffsets);
} else {
- initializeSubscribedPartitionsToStartOffsets(
- subscribedPartitionsToStartOffsets,
+ // initialize subscribed partitions
+ List<KafkaTopicPartition> kafkaTopicPartitions = getKafkaPartitions(topics);
+ Preconditions.checkNotNull(kafkaTopicPartitions, "TopicPartitions must not be null.");
+
+ subscribedPartitionsToStartOffsets = initializeSubscribedPartitionsToStartOffsets(
kafkaTopicPartitions,
getRuntimeContext().getIndexOfThisSubtask(),
getRuntimeContext().getNumberOfParallelSubtasks(),
@@ -688,7 +681,6 @@ public abstract class FlinkKafkaConsumerBase<T> extends RichParallelSourceFuncti
* values. The method decides which partitions this consumer instance should subscribe to, and also
* sets the initial offset each subscribed partition should be started from based on the configured startup mode.
*
- * @param subscribedPartitionsToStartOffsets to subscribedPartitionsToStartOffsets to initialize
* @param kafkaTopicPartitions the complete list of all Kafka partitions
* @param indexOfThisSubtask the index of this consumer instance
* @param numParallelSubtasks total number of parallel consumer instances
@@ -698,14 +690,15 @@ public abstract class FlinkKafkaConsumerBase<T> extends RichParallelSourceFuncti
*
* Note: This method is also exposed for testing.
*/
- protected static void initializeSubscribedPartitionsToStartOffsets(
- Map<KafkaTopicPartition, Long> subscribedPartitionsToStartOffsets,
+ protected static Map<KafkaTopicPartition, Long> initializeSubscribedPartitionsToStartOffsets(
List<KafkaTopicPartition> kafkaTopicPartitions,
int indexOfThisSubtask,
int numParallelSubtasks,
StartupMode startupMode,
Map<KafkaTopicPartition, Long> specificStartupOffsets) {
+ Map<KafkaTopicPartition, Long> subscribedPartitionsToStartOffsets = new HashMap<>(kafkaTopicPartitions.size());
+
for (KafkaTopicPartition kafkaTopicPartition : kafkaTopicPartitions) {
// only handle partitions that this subtask should subscribe to
if (KafkaTopicPartitionAssigner.assign(kafkaTopicPartition, numParallelSubtasks) == indexOfThisSubtask) {
@@ -729,6 +722,8 @@ public abstract class FlinkKafkaConsumerBase<T> extends RichParallelSourceFuncti
}
}
}
+
+ return subscribedPartitionsToStartOffsets;
}
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/e9d493db/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseTest.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseTest.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseTest.java
index 31883ce..de0df5f 100644
--- a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseTest.java
+++ b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseTest.java
@@ -564,7 +564,7 @@ public class FlinkKafkaConsumerBaseTest {
* <p>This also verifies that a restoring source is always impervious to changes in the list
* of topics fetched from Kafka.
*/
- void testRescaling(
+ private void testRescaling(
final int initialParallelism,
final int numPartitions,
final int restoredParallelism,
@@ -578,7 +578,7 @@ public class FlinkKafkaConsumerBaseTest {
DummyFlinkKafkaConsumer<String>[] consumers =
new DummyFlinkKafkaConsumer[initialParallelism];
- AbstractStreamOperatorTestHarness<String>[] testHarnesses=
+ AbstractStreamOperatorTestHarness<String>[] testHarnesses =
new AbstractStreamOperatorTestHarness[initialParallelism];
@@ -659,6 +659,110 @@ public class FlinkKafkaConsumerBaseTest {
assertThat(mockFetchedPartitionsOnStartup, everyItem(isIn(restoredGlobalSubscribedPartitions.keySet())));
}
+ @Test
+ public void testRestoredStateInsensitiveToMissingPartitions() throws Exception {
+ List<KafkaTopicPartition> mockFetchedPartitionsOnStartup = Arrays.asList(
+ new KafkaTopicPartition("test-topic", 0),
+ new KafkaTopicPartition("test-topic", 1),
+ new KafkaTopicPartition("test-topic", 2));
+
+ // missing fetched partitions on restore
+ List<KafkaTopicPartition> mockFetchedPartitionsOnRestore = mockFetchedPartitionsOnStartup.subList(0, 2);
+
+ testRestoredStateInsensitiveToFetchedPartitions(mockFetchedPartitionsOnStartup, mockFetchedPartitionsOnRestore);
+ }
+
+ @Test
+ public void testRestoredStateInsensitiveToNewPartitions() throws Exception {
+ List<KafkaTopicPartition> mockFetchedPartitionsOnStartup = Arrays.asList(
+ new KafkaTopicPartition("test-topic", 0),
+ new KafkaTopicPartition("test-topic", 1),
+ new KafkaTopicPartition("test-topic", 2));
+
+ // new partitions (partition id 3 and 4) on restore
+ List<KafkaTopicPartition> mockFetchedPartitionsOnRestore = Arrays.asList(
+ new KafkaTopicPartition("test-topic", 0),
+ new KafkaTopicPartition("test-topic", 1),
+ new KafkaTopicPartition("test-topic", 2),
+ new KafkaTopicPartition("test-topic", 3),
+ new KafkaTopicPartition("test-topic", 4));
+
+ testRestoredStateInsensitiveToFetchedPartitions(mockFetchedPartitionsOnStartup, mockFetchedPartitionsOnRestore);
+ }
+
+ @Test
+ public void testRestoredStateInsensitiveToDifferentPartitionOrdering() throws Exception {
+ List<KafkaTopicPartition> mockFetchedPartitionsOnStartup = Arrays.asList(
+ new KafkaTopicPartition("test-topic", 0),
+ new KafkaTopicPartition("test-topic", 1),
+ new KafkaTopicPartition("test-topic", 2));
+
+ // different partition ordering on restore
+ List<KafkaTopicPartition> mockFetchedPartitionsOnRestore = Arrays.asList(
+ new KafkaTopicPartition("test-topic", 0),
+ new KafkaTopicPartition("test-topic", 2),
+ new KafkaTopicPartition("test-topic", 1));
+
+ testRestoredStateInsensitiveToFetchedPartitions(mockFetchedPartitionsOnStartup, mockFetchedPartitionsOnRestore);
+ }
+
+ private void testRestoredStateInsensitiveToFetchedPartitions(
+ List<KafkaTopicPartition> mockFetchedPartitionsOnStartup,
+ List<KafkaTopicPartition> mockFetchedPartitionsOnRestore) throws Exception {
+
+ StreamingRuntimeContext mockRuntimeContext = mock(StreamingRuntimeContext.class);
+ when(mockRuntimeContext.isCheckpointingEnabled()).thenReturn(true); // enable checkpointing
+ when(mockRuntimeContext.getNumberOfParallelSubtasks()).thenReturn(1);
+ when(mockRuntimeContext.getIndexOfThisSubtask()).thenReturn(0);
+
+ // startup run for consumer
+ DummyFlinkKafkaConsumer consumer = new DummyFlinkKafkaConsumer(mockFetchedPartitionsOnStartup);
+ consumer.setRuntimeContext(mockRuntimeContext);
+
+ TestingListState<Serializable> listState = new TestingListState<>();
+
+ OperatorStateStore backend = mock(OperatorStateStore.class);
+ when(backend.getSerializableListState(Matchers.any(String.class))).thenReturn(listState);
+
+ StateInitializationContext initializationContext = mock(StateInitializationContext.class);
+ when(initializationContext.getOperatorStateStore()).thenReturn(backend);
+ when(initializationContext.isRestored()).thenReturn(false);
+
+ consumer.initializeState(initializationContext);
+ consumer.open(new Configuration());
+ consumer.snapshotState(new StateSnapshotContextSynchronousImpl(141, 141));
+
+ HashMap<KafkaTopicPartition, Long> startupSnapshot = new HashMap<>();
+
+ for (Serializable serializable : listState.get()) {
+ Tuple2<KafkaTopicPartition, Long> kafkaTopicPartitionLongTuple2 = (Tuple2<KafkaTopicPartition, Long>) serializable;
+ startupSnapshot.put(kafkaTopicPartitionLongTuple2.f0, kafkaTopicPartitionLongTuple2.f1);
+ }
+
+ // restore run for consumer; re-initialize consumer, this
+ // time mocking partition fetching to have missing partitions
+ consumer = new DummyFlinkKafkaConsumer(mockFetchedPartitionsOnRestore);
+ consumer.setRuntimeContext(mockRuntimeContext);
+
+ // re-initialize mock state init context to return true for isRestored
+ initializationContext = mock(StateInitializationContext.class);
+ when(initializationContext.getOperatorStateStore()).thenReturn(backend);
+ when(initializationContext.isRestored()).thenReturn(true);
+
+ consumer.initializeState(initializationContext);
+ consumer.open(new Configuration());
+ consumer.snapshotState(new StateSnapshotContextSynchronousImpl(141, 141));
+
+ HashMap<KafkaTopicPartition, Long> restoreSnapshot = new HashMap<>();
+ for (Serializable serializable : listState.get()) {
+ Tuple2<KafkaTopicPartition, Long> kafkaTopicPartitionLongTuple2 = (Tuple2<KafkaTopicPartition, Long>) serializable;
+ restoreSnapshot.put(kafkaTopicPartitionLongTuple2.f0, kafkaTopicPartitionLongTuple2.f1);
+ }
+
+ // no state should be missing regardless of what partitions were fetched on the restore run
+ Assert.assertEquals(startupSnapshot, restoreSnapshot);
+ }
+
// ------------------------------------------------------------------------
private static <T> FlinkKafkaConsumerBase<T> getConsumer(
http://git-wip-us.apache.org/repos/asf/flink/blob/e9d493db/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerPartitionAssignmentTest.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerPartitionAssignmentTest.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerPartitionAssignmentTest.java
index dd19c21..f88dafe 100644
--- a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerPartitionAssignmentTest.java
+++ b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerPartitionAssignmentTest.java
@@ -28,7 +28,6 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.Map;
-import java.util.HashMap;
import java.util.List;
import java.util.Set;
@@ -56,14 +55,13 @@ public class KafkaConsumerPartitionAssignmentTest {
int startIndex = KafkaTopicPartitionAssigner.assign(inPartitions.get(0), numSubtasks);
for (int subtaskIndex = 0; subtaskIndex < numSubtasks; subtaskIndex++) {
- Map<KafkaTopicPartition, Long> subscribedPartitionsToStartOffsets = new HashMap<>();
- FlinkKafkaConsumerBase.initializeSubscribedPartitionsToStartOffsets(
- subscribedPartitionsToStartOffsets,
- inPartitions,
- subtaskIndex,
- inPartitions.size(),
- StartupMode.GROUP_OFFSETS,
- null);
+ Map<KafkaTopicPartition, Long> subscribedPartitionsToStartOffsets =
+ FlinkKafkaConsumerBase.initializeSubscribedPartitionsToStartOffsets(
+ inPartitions,
+ subtaskIndex,
+ inPartitions.size(),
+ StartupMode.GROUP_OFFSETS,
+ null);
List<KafkaTopicPartition> subscribedPartitions = new ArrayList<>(subscribedPartitionsToStartOffsets.keySet());
@@ -102,14 +100,13 @@ public class KafkaConsumerPartitionAssignmentTest {
int startIndex = KafkaTopicPartitionAssigner.assign(partitions.get(0), numConsumers);
for (int subtaskIndex = 0; subtaskIndex < numConsumers; subtaskIndex++) {
- Map<KafkaTopicPartition, Long> subscribedPartitionsToStartOffsets = new HashMap<>();
- FlinkKafkaConsumerBase.initializeSubscribedPartitionsToStartOffsets(
- subscribedPartitionsToStartOffsets,
- partitions,
- subtaskIndex,
- numConsumers,
- StartupMode.GROUP_OFFSETS,
- null);
+ Map<KafkaTopicPartition, Long> subscribedPartitionsToStartOffsets =
+ FlinkKafkaConsumerBase.initializeSubscribedPartitionsToStartOffsets(
+ partitions,
+ subtaskIndex,
+ numConsumers,
+ StartupMode.GROUP_OFFSETS,
+ null);
List<KafkaTopicPartition> subscribedPartitions = new ArrayList<>(subscribedPartitionsToStartOffsets.keySet());
@@ -152,14 +149,13 @@ public class KafkaConsumerPartitionAssignmentTest {
int startIndex = KafkaTopicPartitionAssigner.assign(inPartitions.get(0), numConsumers);
for (int subtaskIndex = 0; subtaskIndex < numConsumers; subtaskIndex++) {
- Map<KafkaTopicPartition, Long> subscribedPartitionsToStartOffsets = new HashMap<>();
- FlinkKafkaConsumerBase.initializeSubscribedPartitionsToStartOffsets(
- subscribedPartitionsToStartOffsets,
- inPartitions,
- subtaskIndex,
- numConsumers,
- StartupMode.GROUP_OFFSETS,
- null);
+ Map<KafkaTopicPartition, Long> subscribedPartitionsToStartOffsets =
+ FlinkKafkaConsumerBase.initializeSubscribedPartitionsToStartOffsets(
+ inPartitions,
+ subtaskIndex,
+ numConsumers,
+ StartupMode.GROUP_OFFSETS,
+ null);
List<KafkaTopicPartition> subscribedPartitions = new ArrayList<>(subscribedPartitionsToStartOffsets.keySet());
@@ -187,24 +183,22 @@ public class KafkaConsumerPartitionAssignmentTest {
public void testAssignEmptyPartitions() {
try {
List<KafkaTopicPartition> ep = new ArrayList<>();
- Map<KafkaTopicPartition, Long> subscribedPartitionsToStartOffsets = new HashMap<>();
- FlinkKafkaConsumerBase.initializeSubscribedPartitionsToStartOffsets(
- subscribedPartitionsToStartOffsets,
- ep,
- 2,
- 4,
- StartupMode.GROUP_OFFSETS,
- null);
+ Map<KafkaTopicPartition, Long> subscribedPartitionsToStartOffsets =
+ FlinkKafkaConsumerBase.initializeSubscribedPartitionsToStartOffsets(
+ ep,
+ 2,
+ 4,
+ StartupMode.GROUP_OFFSETS,
+ null);
assertTrue(subscribedPartitionsToStartOffsets.entrySet().isEmpty());
- subscribedPartitionsToStartOffsets = new HashMap<>();
- FlinkKafkaConsumerBase.initializeSubscribedPartitionsToStartOffsets(
- subscribedPartitionsToStartOffsets,
- ep,
- 0,
- 1,
- StartupMode.GROUP_OFFSETS,
- null);
+ subscribedPartitionsToStartOffsets =
+ FlinkKafkaConsumerBase.initializeSubscribedPartitionsToStartOffsets(
+ ep,
+ 0,
+ 1,
+ StartupMode.GROUP_OFFSETS,
+ null);
assertTrue(subscribedPartitionsToStartOffsets.entrySet().isEmpty());
}
catch (Exception e) {
@@ -235,33 +229,29 @@ public class KafkaConsumerPartitionAssignmentTest {
final int minNewPartitionsPerConsumer = newPartitions.size() / numConsumers;
final int maxNewPartitionsPerConsumer = newPartitions.size() / numConsumers + 1;
- Map<KafkaTopicPartition, Long> subscribedPartitionsToStartOffsets1 = new HashMap<>();
- Map<KafkaTopicPartition, Long> subscribedPartitionsToStartOffsets2 = new HashMap<>();
- Map<KafkaTopicPartition, Long> subscribedPartitionsToStartOffsets3 = new HashMap<>();
-
- FlinkKafkaConsumerBase.initializeSubscribedPartitionsToStartOffsets(
- subscribedPartitionsToStartOffsets1,
- initialPartitions,
- 0,
- numConsumers,
- StartupMode.GROUP_OFFSETS,
- null);
-
- FlinkKafkaConsumerBase.initializeSubscribedPartitionsToStartOffsets(
- subscribedPartitionsToStartOffsets2,
- initialPartitions,
- 1,
- numConsumers,
- StartupMode.GROUP_OFFSETS,
- null);
-
- FlinkKafkaConsumerBase.initializeSubscribedPartitionsToStartOffsets(
- subscribedPartitionsToStartOffsets3,
- initialPartitions,
- 2,
- numConsumers,
- StartupMode.GROUP_OFFSETS,
- null);
+ Map<KafkaTopicPartition, Long> subscribedPartitionsToStartOffsets1 =
+ FlinkKafkaConsumerBase.initializeSubscribedPartitionsToStartOffsets(
+ initialPartitions,
+ 0,
+ numConsumers,
+ StartupMode.GROUP_OFFSETS,
+ null);
+
+ Map<KafkaTopicPartition, Long> subscribedPartitionsToStartOffsets2 =
+ FlinkKafkaConsumerBase.initializeSubscribedPartitionsToStartOffsets(
+ initialPartitions,
+ 1,
+ numConsumers,
+ StartupMode.GROUP_OFFSETS,
+ null);
+
+ Map<KafkaTopicPartition, Long> subscribedPartitionsToStartOffsets3 =
+ FlinkKafkaConsumerBase.initializeSubscribedPartitionsToStartOffsets(
+ initialPartitions,
+ 2,
+ numConsumers,
+ StartupMode.GROUP_OFFSETS,
+ null);
List<KafkaTopicPartition> subscribedPartitions1 = new ArrayList<>(subscribedPartitionsToStartOffsets1.keySet());
List<KafkaTopicPartition> subscribedPartitions2 = new ArrayList<>(subscribedPartitionsToStartOffsets2.keySet());
@@ -300,33 +290,29 @@ public class KafkaConsumerPartitionAssignmentTest {
// grow the set of partitions and distribute anew
- subscribedPartitionsToStartOffsets1 = new HashMap<>();
- subscribedPartitionsToStartOffsets2 = new HashMap<>();
- subscribedPartitionsToStartOffsets3 = new HashMap<>();
-
- FlinkKafkaConsumerBase.initializeSubscribedPartitionsToStartOffsets(
- subscribedPartitionsToStartOffsets1,
- newPartitions,
- 0,
- numConsumers,
- StartupMode.GROUP_OFFSETS,
- null);
-
- FlinkKafkaConsumerBase.initializeSubscribedPartitionsToStartOffsets(
- subscribedPartitionsToStartOffsets2,
- newPartitions,
- 1,
- numConsumers,
- StartupMode.GROUP_OFFSETS,
- null);
-
- FlinkKafkaConsumerBase.initializeSubscribedPartitionsToStartOffsets(
- subscribedPartitionsToStartOffsets3,
- newPartitions,
- 2,
- numConsumers,
- StartupMode.GROUP_OFFSETS,
- null);
+ subscribedPartitionsToStartOffsets1 =
+ FlinkKafkaConsumerBase.initializeSubscribedPartitionsToStartOffsets(
+ newPartitions,
+ 0,
+ numConsumers,
+ StartupMode.GROUP_OFFSETS,
+ null);
+
+ subscribedPartitionsToStartOffsets2 =
+ FlinkKafkaConsumerBase.initializeSubscribedPartitionsToStartOffsets(
+ newPartitions,
+ 1,
+ numConsumers,
+ StartupMode.GROUP_OFFSETS,
+ null);
+
+ subscribedPartitionsToStartOffsets3 =
+ FlinkKafkaConsumerBase.initializeSubscribedPartitionsToStartOffsets(
+ newPartitions,
+ 2,
+ numConsumers,
+ StartupMode.GROUP_OFFSETS,
+ null);
List<KafkaTopicPartition> subscribedPartitions1New = new ArrayList<>(subscribedPartitionsToStartOffsets1.keySet());
List<KafkaTopicPartition> subscribedPartitions2New = new ArrayList<>(subscribedPartitionsToStartOffsets2.keySet());
@@ -396,24 +382,22 @@ public class KafkaConsumerPartitionAssignmentTest {
new KafkaTopicPartition("test-topic", 2));
for (int subtaskIndex = 0; subtaskIndex < numSubtasks; subtaskIndex++) {
- Map<KafkaTopicPartition, Long> subscribedPartitionsToStartOffsets = new HashMap<>();
- Map<KafkaTopicPartition, Long> subscribedPartitionsToStartOffsetsOutOfOrder = new HashMap<>();
- FlinkKafkaConsumerBase.initializeSubscribedPartitionsToStartOffsets(
- subscribedPartitionsToStartOffsets,
- inPartitions,
- subtaskIndex,
- inPartitions.size(),
- StartupMode.GROUP_OFFSETS,
- null);
-
- FlinkKafkaConsumerBase.initializeSubscribedPartitionsToStartOffsets(
- subscribedPartitionsToStartOffsetsOutOfOrder,
- inPartitionsOutOfOrder,
- subtaskIndex,
- inPartitions.size(),
- StartupMode.GROUP_OFFSETS,
- null);
+ Map<KafkaTopicPartition, Long> subscribedPartitionsToStartOffsets =
+ FlinkKafkaConsumerBase.initializeSubscribedPartitionsToStartOffsets(
+ inPartitions,
+ subtaskIndex,
+ inPartitions.size(),
+ StartupMode.GROUP_OFFSETS,
+ null);
+
+ Map<KafkaTopicPartition, Long> subscribedPartitionsToStartOffsetsOutOfOrder =
+ FlinkKafkaConsumerBase.initializeSubscribedPartitionsToStartOffsets(
+ inPartitionsOutOfOrder,
+ subtaskIndex,
+ inPartitions.size(),
+ StartupMode.GROUP_OFFSETS,
+ null);
// the subscribed partitions should be identical, regardless of the input partition ordering
assertEquals(subscribedPartitionsToStartOffsets, subscribedPartitionsToStartOffsetsOutOfOrder);
[2/6] flink git commit: [FLINK-7143] [kafka] Fix detection of
restored bit in Kafka Consumer
Posted by tz...@apache.org.
[FLINK-7143] [kafka] Fix detection of restored bit in Kafka Consumer
Before, the problem was that empty state was associated with the source
not being restored. However, a source can have empty restored state in
one of two cases:
1. The source was not restored.
2. The overall job was restored but the source simply didn't get any
operator state assigned.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/b0564322
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/b0564322
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/b0564322
Branch: refs/heads/release-1.3
Commit: b0564322c61168c3a7bb23bdca3db0648454a691
Parents: 6e0d90c
Author: Aljoscha Krettek <al...@gmail.com>
Authored: Mon Jul 17 19:06:09 2017 +0200
Committer: Tzu-Li (Gordon) Tai <tz...@apache.org>
Committed: Sat Jul 22 19:36:37 2017 +0800
----------------------------------------------------------------------
.../connectors/kafka/FlinkKafkaConsumerBase.java | 17 ++++++++++++-----
.../kafka/FlinkKafkaConsumerBaseMigrationTest.java | 11 ++++-------
2 files changed, 16 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/b0564322/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 8dbca72..d9b75bb 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
@@ -142,6 +142,9 @@ public abstract class FlinkKafkaConsumerBase<T> extends RichParallelSourceFuncti
/** Flag indicating whether the consumer is still running **/
private volatile boolean running = true;
+ /** Whether this operator instance was restored from checkpointed state. */
+ private transient boolean restored = false;
+
// ------------------------------------------------------------------------
/**
@@ -354,7 +357,7 @@ public abstract class FlinkKafkaConsumerBase<T> extends RichParallelSourceFuncti
subscribedPartitionsToStartOffsets = new HashMap<>(kafkaTopicPartitions.size());
- if (restoredState != null) {
+ if (restored) {
for (KafkaTopicPartition kafkaTopicPartition : kafkaTopicPartitions) {
if (restoredState.containsKey(kafkaTopicPartition)) {
subscribedPartitionsToStartOffsets.put(kafkaTopicPartition, restoredState.get(kafkaTopicPartition));
@@ -503,6 +506,11 @@ public abstract class FlinkKafkaConsumerBase<T> extends RichParallelSourceFuncti
@Override
public void initializeState(FunctionInitializationContext context) throws Exception {
+ // we might have been restored via restoreState() which restores from legacy operator state
+ if (!restored) {
+ restored = context.isRestored();
+ }
+
OperatorStateStore stateStore = context.getOperatorStateStore();
offsetsStateForCheckpoint = stateStore.getSerializableListState(DefaultOperatorStateBackend.DEFAULT_OPERATOR_STATE_NAME);
@@ -518,9 +526,6 @@ public abstract class FlinkKafkaConsumerBase<T> extends RichParallelSourceFuncti
LOG.debug("Using the following offsets: {}", restoredState);
}
}
- if (restoredState != null && restoredState.isEmpty()) {
- restoredState = null;
- }
} else {
LOG.info("No restore state for FlinkKafkaConsumer.");
}
@@ -576,7 +581,9 @@ public abstract class FlinkKafkaConsumerBase<T> extends RichParallelSourceFuncti
LOG.info("{} (taskIdx={}) restoring offsets from an older version.",
getClass().getSimpleName(), getRuntimeContext().getIndexOfThisSubtask());
- restoredState = restoredOffsets.isEmpty() ? null : restoredOffsets;
+ restoredState = restoredOffsets.isEmpty()
+ ? new HashMap<KafkaTopicPartition, Long>() : restoredOffsets;
+ restored = true;
if (LOG.isDebugEnabled()) {
LOG.debug("{} (taskIdx={}) restored offsets from an older Flink version: {}",
http://git-wip-us.apache.org/repos/asf/flink/blob/b0564322/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseMigrationTest.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseMigrationTest.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseMigrationTest.java
index 70e60f3..e7822a7 100644
--- a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseMigrationTest.java
+++ b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseMigrationTest.java
@@ -202,7 +202,7 @@ public class FlinkKafkaConsumerBaseMigrationTest {
assertTrue(consumerFunction.getSubscribedPartitionsToStartOffsets().isEmpty());
// assert that no state was restored
- assertTrue(consumerFunction.getRestoredState() == null);
+ assertTrue(consumerFunction.getRestoredState().isEmpty());
consumerOperator.close();
consumerOperator.cancel();
@@ -244,12 +244,9 @@ public class FlinkKafkaConsumerBaseMigrationTest {
expectedSubscribedPartitionsWithStartOffsets.put(partition, KafkaTopicPartitionStateSentinel.GROUP_OFFSET);
}
- // assert that there are partitions and is identical to expected list
- assertTrue(consumerFunction.getSubscribedPartitionsToStartOffsets() != null);
- assertTrue(!consumerFunction.getSubscribedPartitionsToStartOffsets().isEmpty());
- Assert.assertEquals(expectedSubscribedPartitionsWithStartOffsets, consumerFunction.getSubscribedPartitionsToStartOffsets());
-
- assertTrue(consumerFunction.getRestoredState() == null);
+ // verify that we do not try to fetch the partitions list and subscribe to them even though we have empty state
+ assertTrue(consumerFunction.getRestoredState().isEmpty());
+ assertTrue(consumerFunction.getSubscribedPartitionsToStartOffsets().isEmpty());
consumerOperator.close();
consumerOperator.cancel();
[6/6] flink git commit: [FLINK-7224] [kafka,
docs] Fix incorrect Javadoc / docs regarding Kafka partition metadata
querying
Posted by tz...@apache.org.
[FLINK-7224] [kafka, docs] Fix incorrect Javadoc / docs regarding Kafka partition metadata querying
Since Flink 1.3, partition metadata is no longer queried on the client
side. This commit corrects the statements of this legacy behaviour in
the Javadocs and documentation.
This closes #4363.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/cf70255d
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/cf70255d
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/cf70255d
Branch: refs/heads/release-1.3
Commit: cf70255d427c4f16cb5cda008952d28f20655da1
Parents: e9d493d
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
Authored: Wed Jul 19 14:24:27 2017 +0800
Committer: Tzu-Li (Gordon) Tai <tz...@apache.org>
Committed: Sat Jul 22 19:46:34 2017 +0800
----------------------------------------------------------------------
docs/dev/connectors/kafka.md | 6 ------
.../streaming/connectors/kafka/FlinkKafkaConsumer010.java | 4 ----
.../flink/streaming/connectors/kafka/FlinkKafkaConsumer08.java | 4 ----
.../flink/streaming/connectors/kafka/FlinkKafkaConsumer09.java | 4 ----
4 files changed, 18 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/cf70255d/docs/dev/connectors/kafka.md
----------------------------------------------------------------------
diff --git a/docs/dev/connectors/kafka.md b/docs/dev/connectors/kafka.md
index bc7e7de..e5094a2 100644
--- a/docs/dev/connectors/kafka.md
+++ b/docs/dev/connectors/kafka.md
@@ -134,12 +134,6 @@ stream = env
</div>
</div>
-The current FlinkKafkaConsumer implementation will establish a connection from the client (when calling the constructor)
-for querying the list of topics and partitions.
-
-For this to work, the consumer needs to be able to access the consumers from the machine submitting the job to the Flink cluster.
-If you experience any issues with the Kafka consumer on the client side, the client log might contain information about failed requests, etc.
-
### The `DeserializationSchema`
The Flink Kafka Consumer needs to know how to turn the binary data in Kafka into Java/Scala objects. The
http://git-wip-us.apache.org/repos/asf/flink/blob/cf70255d/flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer010.java
----------------------------------------------------------------------
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 23fc84e..d102dfc 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
@@ -53,10 +53,6 @@ import java.util.Properties;
*
* <p>Please refer to Kafka's documentation for the available configuration properties:
* http://kafka.apache.org/documentation.html#newconsumerconfigs</p>
- *
- * <p><b>NOTE:</b> The implementation currently accesses partition metadata when the consumer
- * is constructed. That means that the client that submits the program needs to be able to
- * reach the Kafka brokers or ZooKeeper.</p>
*/
public class FlinkKafkaConsumer010<T> extends FlinkKafkaConsumer09<T> {
http://git-wip-us.apache.org/repos/asf/flink/blob/cf70255d/flink-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer08.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer08.java b/flink-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer08.java
index 858a790..10b7b0e 100644
--- a/flink-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer08.java
+++ b/flink-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer08.java
@@ -93,10 +93,6 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
*
* <p>When using a Kafka topic to send data between Flink jobs, we recommend using the
* {@see TypeInformationSerializationSchema} and {@see TypeInformationKeyValueSerializationSchema}.</p>
- *
- * <p><b>NOTE:</b> The implementation currently accesses partition metadata when the consumer
- * is constructed. That means that the client that submits the program needs to be able to
- * reach the Kafka brokers or ZooKeeper.</p>
*/
public class FlinkKafkaConsumer08<T> extends FlinkKafkaConsumerBase<T> {
http://git-wip-us.apache.org/repos/asf/flink/blob/cf70255d/flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer09.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer09.java b/flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer09.java
index e638348..ea2013c 100644
--- a/flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer09.java
+++ b/flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer09.java
@@ -62,10 +62,6 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
*
* <p>Please refer to Kafka's documentation for the available configuration properties:
* http://kafka.apache.org/documentation.html#newconsumerconfigs</p>
- *
- * <p><b>NOTE:</b> The implementation currently accesses partition metadata when the consumer
- * is constructed. That means that the client that submits the program needs to be able to
- * reach the Kafka brokers or ZooKeeper.</p>
*/
public class FlinkKafkaConsumer09<T> extends FlinkKafkaConsumerBase<T> {
[3/6] flink git commit: [hotfix] [kafka] Make checkpoint methods
final in KafkaConsumerBase
Posted by tz...@apache.org.
[hotfix] [kafka] Make checkpoint methods final in KafkaConsumerBase
This prevents concrete Kafka Source implementations from accidentally
overriding the checkpointing methods. This would be problematic when not
providing tests. We test the checkpoint methods of the ConsumerBase but
derived methods would not be tested.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/ead25aa7
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/ead25aa7
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/ead25aa7
Branch: refs/heads/release-1.3
Commit: ead25aa7a591f28f675af509a5874a77f01713ac
Parents: b056432
Author: Aljoscha Krettek <al...@gmail.com>
Authored: Tue Jul 18 10:35:54 2017 +0200
Committer: Tzu-Li (Gordon) Tai <tz...@apache.org>
Committed: Sat Jul 22 19:37:09 2017 +0800
----------------------------------------------------------------------
.../streaming/connectors/kafka/FlinkKafkaConsumerBase.java | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/ead25aa7/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 d9b75bb..199755c 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
@@ -504,7 +504,7 @@ public abstract class FlinkKafkaConsumerBase<T> extends RichParallelSourceFuncti
// ------------------------------------------------------------------------
@Override
- public void initializeState(FunctionInitializationContext context) throws Exception {
+ public final void initializeState(FunctionInitializationContext context) throws Exception {
// we might have been restored via restoreState() which restores from legacy operator state
if (!restored) {
@@ -532,7 +532,7 @@ public abstract class FlinkKafkaConsumerBase<T> extends RichParallelSourceFuncti
}
@Override
- public void snapshotState(FunctionSnapshotContext context) throws Exception {
+ public final void snapshotState(FunctionSnapshotContext context) throws Exception {
if (!running) {
LOG.debug("snapshotState() called on closed source");
} else {
@@ -577,7 +577,7 @@ public abstract class FlinkKafkaConsumerBase<T> extends RichParallelSourceFuncti
}
@Override
- public void restoreState(HashMap<KafkaTopicPartition, Long> restoredOffsets) {
+ public final void restoreState(HashMap<KafkaTopicPartition, Long> restoredOffsets) {
LOG.info("{} (taskIdx={}) restoring offsets from an older version.",
getClass().getSimpleName(), getRuntimeContext().getIndexOfThisSubtask());