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/28 13:53:37 UTC
[7/7] flink git commit: [FLINK-7248] [kafka,
tests] Remove invalid checkRestoredNullCheckpointWhenFetcherNotReady
test
[FLINK-7248] [kafka, tests] Remove invalid checkRestoredNullCheckpointWhenFetcherNotReady test
This test is an invalid remnant from recent major Kafka consumer
refactorings. The actual behaviour is covered by
checkRestoredCheckpointWhenFetcherNotReady. When the fetcher is not yet
ready and exposed and a checkpoint happens, we fallback to using any
restored state as the checkpoint.
This closes #4387.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/c5221c85
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/c5221c85
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/c5221c85
Branch: refs/heads/master
Commit: c5221c85a1c532a5234ec0b4331a04cffa04ba5c
Parents: 68395cc
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
Authored: Mon Jul 24 16:34:37 2017 +0800
Committer: Tzu-Li (Gordon) Tai <tz...@apache.org>
Committed: Fri Jul 28 21:52:30 2017 +0800
----------------------------------------------------------------------
.../kafka/FlinkKafkaConsumerBaseTest.java | 49 ++++----------------
1 file changed, 9 insertions(+), 40 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/c5221c85/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 fef2820..ad08c4d 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
@@ -137,15 +137,16 @@ public class FlinkKafkaConsumerBaseTest {
}
/**
- * Tests that no checkpoints happen when the fetcher is not running.
+ * Tests that when taking a checkpoint when the fetcher is not running yet,
+ * the checkpoint correctly contains the restored state instead.
*/
@Test
public void checkRestoredCheckpointWhenFetcherNotReady() throws Exception {
OperatorStateStore operatorStateStore = mock(OperatorStateStore.class);
- TestingListState<Serializable> listState = new TestingListState<>();
- listState.add(Tuple2.of(new KafkaTopicPartition("abc", 13), 16768L));
- listState.add(Tuple2.of(new KafkaTopicPartition("def", 7), 987654321L));
+ TestingListState<Serializable> restoredListState = new TestingListState<>();
+ restoredListState.add(Tuple2.of(new KafkaTopicPartition("abc", 13), 16768L));
+ restoredListState.add(Tuple2.of(new KafkaTopicPartition("def", 7), 987654321L));
FlinkKafkaConsumerBase<String> consumer = getConsumer(null, new LinkedMap(), true);
StreamingRuntimeContext context = mock(StreamingRuntimeContext.class);
@@ -156,7 +157,7 @@ public class FlinkKafkaConsumerBaseTest {
// mock old 1.2 state (empty)
when(operatorStateStore.getSerializableListState(Matchers.any(String.class))).thenReturn(new TestingListState<Serializable>());
// mock 1.3 state
- when(operatorStateStore.getUnionListState(Matchers.any(ListStateDescriptor.class))).thenReturn(listState);
+ when(operatorStateStore.getUnionListState(Matchers.any(ListStateDescriptor.class))).thenReturn(restoredListState);
StateInitializationContext initializationContext = mock(StateInitializationContext.class);
@@ -171,17 +172,17 @@ public class FlinkKafkaConsumerBaseTest {
// ensure that the list was cleared and refilled. while this is an implementation detail, we use it here
// to figure out that snapshotState() actually did something.
- Assert.assertTrue(listState.isClearCalled());
+ Assert.assertTrue(restoredListState.isClearCalled());
Set<Serializable> expected = new HashSet<>();
- for (Serializable serializable : listState.get()) {
+ for (Serializable serializable : restoredListState.get()) {
expected.add(serializable);
}
int counter = 0;
- for (Serializable serializable : listState.get()) {
+ for (Serializable serializable : restoredListState.get()) {
assertTrue(expected.contains(serializable));
counter++;
}
@@ -189,38 +190,6 @@ public class FlinkKafkaConsumerBaseTest {
assertEquals(expected.size(), counter);
}
- /**
- * Tests that no checkpoints happen when the fetcher is not running.
- */
- @Test
- public void checkRestoredNullCheckpointWhenFetcherNotReady() throws Exception {
- FlinkKafkaConsumerBase<String> consumer = getConsumer(null, new LinkedMap(), true);
- StreamingRuntimeContext runtimeContext = mock(StreamingRuntimeContext.class);
- when(runtimeContext.getIndexOfThisSubtask()).thenReturn(0);
- when(runtimeContext.getNumberOfParallelSubtasks()).thenReturn(1);
- consumer.setRuntimeContext(runtimeContext);
-
- OperatorStateStore operatorStateStore = mock(OperatorStateStore.class);
- TestingListState<Serializable> listState = new TestingListState<>();
- // mock old 1.2 state (empty)
- when(operatorStateStore.getSerializableListState(Matchers.any(String.class))).thenReturn(new TestingListState<Serializable>());
- // mock 1.3 state
- when(operatorStateStore.getUnionListState(Matchers.any(ListStateDescriptor.class))).thenReturn(listState);
-
- StateInitializationContext initializationContext = mock(StateInitializationContext.class);
-
- when(initializationContext.getOperatorStateStore()).thenReturn(operatorStateStore);
- when(initializationContext.isRestored()).thenReturn(false);
-
- consumer.initializeState(initializationContext);
-
- consumer.open(new Configuration());
-
- consumer.snapshotState(new StateSnapshotContextSynchronousImpl(17, 17));
-
- assertFalse(listState.get().iterator().hasNext());
- }
-
@Test
public void testConfigureOnCheckpointsCommitMode() throws Exception {