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 2018/01/13 07:53:25 UTC
[1/6] flink git commit: [FLINK-6951] [kinesis] Shade httpcomponents
dependency for Kinesis connector
Repository: flink
Updated Branches:
refs/heads/release-1.4 b32ac3aaf -> c91544d14
[FLINK-6951] [kinesis] Shade httpcomponents dependency for Kinesis connector
This closes #4150.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/b607a8b9
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/b607a8b9
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/b607a8b9
Branch: refs/heads/release-1.4
Commit: b607a8b95d11d80f68cdd167289a52d1f4c49d24
Parents: b32ac3a
Author: Bowen Li <bo...@gmail.com>
Authored: Wed Jun 21 15:55:19 2017 -0700
Committer: Tzu-Li (Gordon) Tai <tz...@apache.org>
Committed: Sat Jan 13 11:33:34 2018 +0800
----------------------------------------------------------------------
flink-connectors/flink-connector-kinesis/pom.xml | 5 +++++
1 file changed, 5 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/b607a8b9/flink-connectors/flink-connector-kinesis/pom.xml
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kinesis/pom.xml b/flink-connectors/flink-connector-kinesis/pom.xml
index 83934f6..f64b831 100644
--- a/flink-connectors/flink-connector-kinesis/pom.xml
+++ b/flink-connectors/flink-connector-kinesis/pom.xml
@@ -154,6 +154,7 @@ under the License.
<includes>
<include>com.amazonaws:*</include>
<include>com.google.protobuf:*</include>
+ <include>org.apache.httpcomponents:*</include>
</includes>
</artifactSet>
<relocations combine.children="override">
@@ -166,6 +167,10 @@ under the License.
<pattern>com.amazonaws</pattern>
<shadedPattern>org.apache.flink.kinesis.shaded.com.amazonaws</shadedPattern>
</relocation>
+ <relocation>
+ <pattern>org.apache.http</pattern>
+ <shadedPattern>org.apache.flink.kinesis.shaded.org.apache.http</shadedPattern>
+ </relocation>
</relocations>
</configuration>
</execution>
[3/6] flink git commit: [hotfix] Fix typo in AbstractMetricGroup.java
Posted by tz...@apache.org.
[hotfix] Fix typo in AbstractMetricGroup.java
This closes #5280.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/b7f8a5a2
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/b7f8a5a2
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/b7f8a5a2
Branch: refs/heads/release-1.4
Commit: b7f8a5a2b03d5a5c8ca2f119486e17c24befbc99
Parents: ae00cad
Author: maqingxiang-it <ma...@dev05v.sys.corp.qihoo.net>
Authored: Thu Jan 11 13:44:04 2018 +0800
Committer: Tzu-Li (Gordon) Tai <tz...@apache.org>
Committed: Sat Jan 13 11:33:50 2018 +0800
----------------------------------------------------------------------
.../apache/flink/runtime/metrics/groups/AbstractMetricGroup.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/b7f8a5a2/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroup.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroup.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroup.java
index 66eace5..23b969a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroup.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroup.java
@@ -352,7 +352,7 @@ public abstract class AbstractMetricGroup<A extends AbstractMetricGroup<?>> impl
// add the metric only if the group is still open
synchronized (this) {
if (!closed) {
- // immediately put without a 'contains' check to optimize the common case (no collition)
+ // immediately put without a 'contains' check to optimize the common case (no collision)
// collisions are resolved later
Metric prior = metrics.put(name, metric);
[2/6] flink git commit: [hotfix] [doc] Fixed doc typo in DataStream
API
Posted by tz...@apache.org.
[hotfix] [doc] Fixed doc typo in DataStream API
This closes #5283.
This closes #5191.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/ae00cad3
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/ae00cad3
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/ae00cad3
Branch: refs/heads/release-1.4
Commit: ae00cad38c868ee294d1d0930731974bfefcb727
Parents: b607a8b
Author: Alejandro Alcalde <al...@gmail.com>
Authored: Thu Jan 11 13:49:29 2018 +0100
Committer: Tzu-Li (Gordon) Tai <tz...@apache.org>
Committed: Sat Jan 13 11:33:42 2018 +0800
----------------------------------------------------------------------
docs/dev/datastream_api.md | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/ae00cad3/docs/dev/datastream_api.md
----------------------------------------------------------------------
diff --git a/docs/dev/datastream_api.md b/docs/dev/datastream_api.md
index 09580b1..d7ab2e7 100644
--- a/docs/dev/datastream_api.md
+++ b/docs/dev/datastream_api.md
@@ -161,7 +161,7 @@ File-based:
*IMPLEMENTATION:*
- Under the hood, Flink splits the file reading process into two sub-tasks, namely *directory monitoring* and *data reading*. Each of these sub-tasks is implemented by a separate entity. Monitoring is implemented by a single, **non-parallel** (parallelism = 1) task, while reading is performed by multiple tasks running in parallel. The parallelism of the latter is equal to the job parallelism. The role of the single monitoring task is to scan the directory (periodically or only once depending on the `watchType`), find the files to be processed, divide them in *splits*, and assign these splits to the downstream readers. The readers are the ones who will read the actual data. Each split is read by only one reader, while a reader can read muplitple splits, one-by-one.
+ Under the hood, Flink splits the file reading process into two sub-tasks, namely *directory monitoring* and *data reading*. Each of these sub-tasks is implemented by a separate entity. Monitoring is implemented by a single, **non-parallel** (parallelism = 1) task, while reading is performed by multiple tasks running in parallel. The parallelism of the latter is equal to the job parallelism. The role of the single monitoring task is to scan the directory (periodically or only once depending on the `watchType`), find the files to be processed, divide them in *splits*, and assign these splits to the downstream readers. The readers are the ones who will read the actual data. Each split is read by only one reader, while a reader can read multiple splits, one-by-one.
*IMPORTANT NOTES:*
@@ -219,7 +219,7 @@ File-based:
*IMPLEMENTATION:*
- Under the hood, Flink splits the file reading process into two sub-tasks, namely *directory monitoring* and *data reading*. Each of these sub-tasks is implemented by a separate entity. Monitoring is implemented by a single, **non-parallel** (parallelism = 1) task, while reading is performed by multiple tasks running in parallel. The parallelism of the latter is equal to the job parallelism. The role of the single monitoring task is to scan the directory (periodically or only once depending on the `watchType`), find the files to be processed, divide them in *splits*, and assign these splits to the downstream readers. The readers are the ones who will read the actual data. Each split is read by only one reader, while a reader can read muplitple splits, one-by-one.
+ Under the hood, Flink splits the file reading process into two sub-tasks, namely *directory monitoring* and *data reading*. Each of these sub-tasks is implemented by a separate entity. Monitoring is implemented by a single, **non-parallel** (parallelism = 1) task, while reading is performed by multiple tasks running in parallel. The parallelism of the latter is equal to the job parallelism. The role of the single monitoring task is to scan the directory (periodically or only once depending on the `watchType`), find the files to be processed, divide them in *splits*, and assign these splits to the downstream readers. The readers are the ones who will read the actual data. Each split is read by only one reader, while a reader can read multiple splits, one-by-one.
*IMPORTANT NOTES:*
[6/6] flink git commit: [hotfix] [kafka] Remove stale comment on
publishing procedures of AbstractFetcher
Posted by tz...@apache.org.
[hotfix] [kafka] Remove stale comment on publishing procedures of AbstractFetcher
The previous comment mentioned "only now will the fetcher return at
least the restored offsets when calling snapshotCurrentState()". This is
a remnant of the previous fetcher initialization behaviour, where in the
past the fetcher wasn't directly seeded with restored offsets on
instantiation.
Since this is no longer true, this commit fixes the stale comment to
avoid confusion.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/c91544d1
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/c91544d1
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/c91544d1
Branch: refs/heads/release-1.4
Commit: c91544d140668a1b6c4c59909cc9374cc890ccf3
Parents: c4bfc7d
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
Authored: Wed Dec 20 11:54:40 2017 -0800
Committer: Tzu-Li (Gordon) Tai <tz...@apache.org>
Committed: Sat Jan 13 11:34:28 2018 +0800
----------------------------------------------------------------------
.../kafka/FlinkKafkaConsumerBase.java | 20 +++++++++-----------
1 file changed, 9 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/c91544d1/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 44e0d54..d71827f 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
@@ -545,8 +545,12 @@ public abstract class FlinkKafkaConsumerBase<T> extends RichParallelSourceFuncti
sourceContext.markAsTemporarilyIdle();
}
- // create the fetcher that will communicate with the Kafka brokers
- final AbstractFetcher<T, ?> fetcher = createFetcher(
+ // from this point forward:
+ // - 'snapshotState' will draw offsets from the fetcher,
+ // instead of being built from `subscribedPartitionsToStartOffsets`
+ // - 'notifyCheckpointComplete' will start to do work (i.e. commit offsets to
+ // Kafka through the fetcher, if configured to do so)
+ this.kafkaFetcher = createFetcher(
sourceContext,
subscribedPartitionsToStartOffsets,
periodicWatermarkAssigner,
@@ -554,12 +558,6 @@ public abstract class FlinkKafkaConsumerBase<T> extends RichParallelSourceFuncti
(StreamingRuntimeContext) getRuntimeContext(),
offsetCommitMode);
- // publish the reference, for snapshot-, commit-, and cancel calls
- // IMPORTANT: We can only do that now, because only now will calls to
- // the fetchers 'snapshotCurrentState()' method return at least
- // the restored offsets
- this.kafkaFetcher = fetcher;
-
if (!running) {
return;
}
@@ -598,7 +596,7 @@ public abstract class FlinkKafkaConsumerBase<T> extends RichParallelSourceFuncti
// no need to add the discovered partitions if we were closed during the meantime
if (running && !discoveredPartitions.isEmpty()) {
- fetcher.addDiscoveredPartitions(discoveredPartitions);
+ kafkaFetcher.addDiscoveredPartitions(discoveredPartitions);
}
// do not waste any time sleeping if we're not running anymore
@@ -621,7 +619,7 @@ public abstract class FlinkKafkaConsumerBase<T> extends RichParallelSourceFuncti
});
discoveryLoopThread.start();
- fetcher.runFetchLoop();
+ kafkaFetcher.runFetchLoop();
// --------------------------------------------------------------------
@@ -638,7 +636,7 @@ public abstract class FlinkKafkaConsumerBase<T> extends RichParallelSourceFuncti
// won't be using the discoverer
partitionDiscoverer.close();
- fetcher.runFetchLoop();
+ kafkaFetcher.runFetchLoop();
}
}
[5/6] flink git commit: [FLINK-8306] [kafka,
tests] Fix mock verifications on final method
Posted by tz...@apache.org.
[FLINK-8306] [kafka, tests] Fix mock verifications on final method
Previously, offset commit behavioural tests relied on verifying on
AbstractFetcher::commitInternalOffsetsToKafka(). That method is actually
final, and could not be mocked.
This commit fixes that by implementing a proper mock AbstractFetcher,
which keeps track of the offset commits that go through.
This closes #5284.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/c4bfc7de
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/c4bfc7de
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/c4bfc7de
Branch: refs/heads/release-1.4
Commit: c4bfc7de36201d7a144ae995931ffd3a079ed649
Parents: 968683f
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
Authored: Fri Jan 12 08:45:32 2018 +0800
Committer: Tzu-Li (Gordon) Tai <tz...@apache.org>
Committed: Sat Jan 13 11:34:20 2018 +0800
----------------------------------------------------------------------
.../kafka/FlinkKafkaConsumerBaseTest.java | 203 ++++++++++---------
.../kafka/internals/AbstractFetcherTest.java | 63 +-----
.../kafka/testutils/TestSourceContext.java | 87 ++++++++
3 files changed, 199 insertions(+), 154 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/c4bfc7de/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 4361c67..f091c08 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
@@ -19,7 +19,6 @@
package org.apache.flink.streaming.connectors.kafka;
import org.apache.flink.api.common.ExecutionConfig;
-import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.api.common.state.KeyedStateStore;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
@@ -48,7 +47,9 @@ import org.apache.flink.streaming.connectors.kafka.internals.KafkaCommitCallback
import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicsDescriptor;
import org.apache.flink.streaming.connectors.kafka.testutils.TestPartitionDiscoverer;
+import org.apache.flink.streaming.connectors.kafka.testutils.TestSourceContext;
import org.apache.flink.streaming.runtime.tasks.OperatorStateHandles;
+import org.apache.flink.streaming.runtime.tasks.TestProcessingTimeService;
import org.apache.flink.streaming.util.AbstractStreamOperatorTestHarness;
import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema;
import org.apache.flink.util.Preconditions;
@@ -56,11 +57,13 @@ import org.apache.flink.util.SerializedValue;
import org.junit.Assert;
import org.junit.Test;
-import org.mockito.Matchers;
-import org.mockito.Mockito;
+
+import javax.annotation.Nonnull;
import java.io.Serializable;
+import java.util.ArrayDeque;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
@@ -68,6 +71,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
+import static org.apache.flink.util.Preconditions.checkState;
import static org.hamcrest.Matchers.everyItem;
import static org.hamcrest.Matchers.hasSize;
import static org.hamcrest.collection.IsIn.isIn;
@@ -75,14 +79,11 @@ 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.assertNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
-import static org.mockito.Matchers.anyMap;
import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.never;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
/**
* Tests for the {@link FlinkKafkaConsumerBase}.
@@ -129,7 +130,11 @@ public class FlinkKafkaConsumerBaseTest {
@Test
public void ignoreCheckpointWhenNotRunning() throws Exception {
@SuppressWarnings("unchecked")
- final FlinkKafkaConsumerBase<String> consumer = new DummyFlinkKafkaConsumer<>();
+ final MockFetcher<String> fetcher = new MockFetcher<>();
+ final FlinkKafkaConsumerBase<String> consumer = new DummyFlinkKafkaConsumer<>(
+ fetcher,
+ mock(AbstractPartitionDiscoverer.class),
+ false);
final TestingListState<Tuple2<KafkaTopicPartition, Long>> listState = new TestingListState<>();
setupConsumer(consumer, false, listState, true, 0, 1);
@@ -139,6 +144,11 @@ public class FlinkKafkaConsumerBaseTest {
// no state should have been checkpointed
assertFalse(listState.get().iterator().hasNext());
+
+ // acknowledgement of the checkpoint should also not result in any offset commits
+ consumer.notifyCheckpointComplete(1L);
+ assertNull(fetcher.getAndClearLastCommittedOffsets());
+ assertEquals(0, fetcher.getCommitCount());
}
/**
@@ -265,10 +275,7 @@ public class FlinkKafkaConsumerBaseTest {
// --------------------------------------------------------------------
- final OneShotLatch runLatch = new OneShotLatch();
- final OneShotLatch stopLatch = new OneShotLatch();
- final AbstractFetcher<String, ?> fetcher = getRunnableMockFetcher(runLatch, stopLatch);
- when(fetcher.snapshotCurrentState()).thenReturn(state1, state2, state3);
+ final MockFetcher<String> fetcher = new MockFetcher<>(state1, state2, state3);
final FlinkKafkaConsumerBase<String> consumer = new DummyFlinkKafkaConsumer<>(
fetcher,
@@ -283,17 +290,11 @@ public class FlinkKafkaConsumerBaseTest {
final CheckedThread runThread = new CheckedThread() {
@Override
public void go() throws Exception {
- consumer.run(mock(SourceFunction.SourceContext.class));
- }
-
- @Override
- public void sync() throws Exception {
- stopLatch.trigger();
- super.sync();
+ consumer.run(new TestSourceContext<>());
}
};
runThread.start();
- runLatch.await();
+ fetcher.waitUntilRun();
assertEquals(0, consumer.getPendingOffsetsToCommit().size());
@@ -329,6 +330,8 @@ public class FlinkKafkaConsumerBaseTest {
consumer.notifyCheckpointComplete(138L);
assertEquals(1, consumer.getPendingOffsetsToCommit().size());
assertTrue(consumer.getPendingOffsetsToCommit().containsKey(140L));
+ assertEquals(state1, fetcher.getAndClearLastCommittedOffsets());
+ assertEquals(1, fetcher.getCommitCount());
// checkpoint 3
consumer.snapshotState(new StateSnapshotContextSynchronousImpl(141, 141));
@@ -347,28 +350,15 @@ public class FlinkKafkaConsumerBaseTest {
// ack checkpoint 3, subsumes number 2
consumer.notifyCheckpointComplete(141L);
assertEquals(0, consumer.getPendingOffsetsToCommit().size());
+ assertEquals(state3, fetcher.getAndClearLastCommittedOffsets());
+ assertEquals(2, fetcher.getCommitCount());
consumer.notifyCheckpointComplete(666); // invalid checkpoint
assertEquals(0, consumer.getPendingOffsetsToCommit().size());
+ assertNull(fetcher.getAndClearLastCommittedOffsets());
+ assertEquals(2, fetcher.getCommitCount());
- // create 500 snapshots
- for (int i = 100; i < 600; i++) {
- consumer.snapshotState(new StateSnapshotContextSynchronousImpl(i, i));
- listState.clear();
- }
- assertEquals(FlinkKafkaConsumerBase.MAX_NUM_PENDING_CHECKPOINTS, consumer.getPendingOffsetsToCommit().size());
-
- // commit only the second last
- consumer.notifyCheckpointComplete(598);
- assertEquals(1, consumer.getPendingOffsetsToCommit().size());
-
- // access invalid checkpoint
- consumer.notifyCheckpointComplete(590);
-
- // and the last
- consumer.notifyCheckpointComplete(599);
- assertEquals(0, consumer.getPendingOffsetsToCommit().size());
-
+ consumer.cancel();
runThread.sync();
}
@@ -393,10 +383,7 @@ public class FlinkKafkaConsumerBaseTest {
// --------------------------------------------------------------------
- final OneShotLatch runLatch = new OneShotLatch();
- final OneShotLatch stopLatch = new OneShotLatch();
- final AbstractFetcher<String, ?> fetcher = getRunnableMockFetcher(runLatch, stopLatch);
- when(fetcher.snapshotCurrentState()).thenReturn(state1, state2, state3);
+ final MockFetcher<String> fetcher = new MockFetcher<>(state1, state2, state3);
final FlinkKafkaConsumerBase<String> consumer = new DummyFlinkKafkaConsumer<>(
fetcher,
@@ -412,17 +399,11 @@ public class FlinkKafkaConsumerBaseTest {
final CheckedThread runThread = new CheckedThread() {
@Override
public void go() throws Exception {
- consumer.run(mock(SourceFunction.SourceContext.class));
- }
-
- @Override
- public void sync() throws Exception {
- stopLatch.trigger();
- super.sync();
+ consumer.run(new TestSourceContext<>());
}
};
runThread.start();
- runLatch.await();
+ fetcher.waitUntilRun();
assertEquals(0, consumer.getPendingOffsetsToCommit().size());
@@ -454,7 +435,8 @@ public class FlinkKafkaConsumerBaseTest {
// ack checkpoint 1
consumer.notifyCheckpointComplete(138L);
- verify(fetcher, never()).commitInternalOffsetsToKafka(anyMap(), Matchers.any(KafkaCommitCallback.class)); // no offsets should be committed
+ assertEquals(0, fetcher.getCommitCount());
+ assertNull(fetcher.getAndClearLastCommittedOffsets()); // no offsets should be committed
// checkpoint 3
consumer.snapshotState(new StateSnapshotContextSynchronousImpl(141, 141));
@@ -471,29 +453,15 @@ public class FlinkKafkaConsumerBaseTest {
// ack checkpoint 3, subsumes number 2
consumer.notifyCheckpointComplete(141L);
- verify(fetcher, never()).commitInternalOffsetsToKafka(anyMap(), Matchers.any(KafkaCommitCallback.class)); // no offsets should be committed
+ assertEquals(0, fetcher.getCommitCount());
+ assertNull(fetcher.getAndClearLastCommittedOffsets()); // no offsets should be committed
consumer.notifyCheckpointComplete(666); // invalid checkpoint
- verify(fetcher, never()).commitInternalOffsetsToKafka(anyMap(), Matchers.any(KafkaCommitCallback.class)); // no offsets should be committed
-
- // create 500 snapshots
- for (int i = 100; i < 600; i++) {
- consumer.snapshotState(new StateSnapshotContextSynchronousImpl(i, i));
- listState.clear();
- }
- assertEquals(0, consumer.getPendingOffsetsToCommit().size()); // pending offsets to commit should not be updated
+ assertEquals(0, fetcher.getCommitCount());
+ assertNull(fetcher.getAndClearLastCommittedOffsets()); // no offsets should be committed
- // commit only the second last
- consumer.notifyCheckpointComplete(598);
- verify(fetcher, never()).commitInternalOffsetsToKafka(anyMap(), Matchers.any(KafkaCommitCallback.class)); // no offsets should be committed
-
- // access invalid checkpoint
- consumer.notifyCheckpointComplete(590);
- verify(fetcher, never()).commitInternalOffsetsToKafka(anyMap(), Matchers.any(KafkaCommitCallback.class)); // no offsets should be committed
-
- // and the last
- consumer.notifyCheckpointComplete(599);
- verify(fetcher, never()).commitInternalOffsetsToKafka(anyMap(), Matchers.any(KafkaCommitCallback.class)); // no offsets should be committed
+ consumer.cancel();
+ runThread.sync();
}
@Test
@@ -734,26 +702,6 @@ public class FlinkKafkaConsumerBaseTest {
}
}
- /**
- * Returns a mock {@link AbstractFetcher}, with run / stop latches injected in
- * the {@link AbstractFetcher#runFetchLoop()} method.
- */
- private static <T> AbstractFetcher<T, ?> getRunnableMockFetcher(
- OneShotLatch runLatch,
- OneShotLatch stopLatch) throws Exception {
-
- @SuppressWarnings("unchecked")
- final AbstractFetcher<T, ?> fetcher = mock(AbstractFetcher.class);
-
- Mockito.doAnswer(invocationOnMock -> {
- runLatch.trigger();
- stopLatch.await();
- return null;
- }).when(fetcher).runFetchLoop();
-
- return fetcher;
- }
-
@SuppressWarnings("unchecked")
private static <T, S> void setupConsumer(
FlinkKafkaConsumerBase<T> consumer,
@@ -769,6 +717,77 @@ public class FlinkKafkaConsumerBaseTest {
consumer.open(new Configuration());
}
+ private static class MockFetcher<T> extends AbstractFetcher<T, Object> {
+
+ private final OneShotLatch runLatch = new OneShotLatch();
+ private final OneShotLatch stopLatch = new OneShotLatch();
+
+ private final ArrayDeque<HashMap<KafkaTopicPartition, Long>> stateSnapshotsToReturn = new ArrayDeque<>();
+
+ private Map<KafkaTopicPartition, Long> lastCommittedOffsets;
+ private int commitCount = 0;
+
+ @SafeVarargs
+ private MockFetcher(HashMap<KafkaTopicPartition, Long>... stateSnapshotsToReturn) throws Exception {
+ super(
+ new TestSourceContext<>(),
+ new HashMap<>(),
+ null,
+ null,
+ new TestProcessingTimeService(),
+ 0,
+ MockFetcher.class.getClassLoader(),
+ false);
+
+ this.stateSnapshotsToReturn.addAll(Arrays.asList(stateSnapshotsToReturn));
+ }
+
+ @Override
+ protected void doCommitInternalOffsetsToKafka(
+ Map<KafkaTopicPartition, Long> offsets,
+ @Nonnull KafkaCommitCallback commitCallback) throws Exception {
+ this.lastCommittedOffsets = offsets;
+ this.commitCount++;
+ commitCallback.onSuccess();
+ }
+
+ @Override
+ public void runFetchLoop() throws Exception {
+ runLatch.trigger();
+ stopLatch.await();
+ }
+
+ @Override
+ public HashMap<KafkaTopicPartition, Long> snapshotCurrentState() {
+ checkState(!stateSnapshotsToReturn.isEmpty());
+ return stateSnapshotsToReturn.poll();
+ }
+
+ @Override
+ protected Object createKafkaPartitionHandle(KafkaTopicPartition partition) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void cancel() {
+ stopLatch.trigger();
+ }
+
+ private void waitUntilRun() throws InterruptedException {
+ runLatch.await();
+ }
+
+ private Map<KafkaTopicPartition, Long> getAndClearLastCommittedOffsets() {
+ Map<KafkaTopicPartition, Long> offsets = this.lastCommittedOffsets;
+ this.lastCommittedOffsets = null;
+ return offsets;
+ }
+
+ private int getCommitCount() {
+ return commitCount;
+ }
+ }
+
private static class MockRuntimeContext extends StreamingRuntimeContext {
private final boolean isCheckpointingEnabled;
@@ -784,7 +803,7 @@ public class FlinkKafkaConsumerBaseTest {
super(
new MockStreamOperator(),
new MockEnvironment("mockTask", 4 * MemoryManager.DEFAULT_PAGE_SIZE, null, 16),
- Collections.<String, Accumulator<?, ?>>emptyMap());
+ Collections.emptyMap());
this.isCheckpointingEnabled = isCheckpointingEnabled;
this.numParallelSubtasks = numParallelSubtasks;
http://git-wip-us.apache.org/repos/asf/flink/blob/c4bfc7de/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 e4a58dd..6fe1d6f 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
@@ -22,7 +22,7 @@ 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.SourceContext;
import org.apache.flink.streaming.api.watermark.Watermark;
-import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.connectors.kafka.testutils.TestSourceContext;
import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
import org.apache.flink.streaming.runtime.tasks.TestProcessingTimeService;
import org.apache.flink.util.SerializedValue;
@@ -444,67 +444,6 @@ public class AbstractFetcherTest {
// ------------------------------------------------------------------------
- private static final class TestSourceContext<T> implements SourceContext<T> {
-
- private final Object checkpointLock = new Object();
- private final Object watermarkLock = new Object();
-
- private volatile StreamRecord<T> latestElement;
- private volatile Watermark currentWatermark;
-
- @Override
- public void collect(T element) {
- this.latestElement = new StreamRecord<>(element);
- }
-
- @Override
- public void collectWithTimestamp(T element, long timestamp) {
- this.latestElement = new StreamRecord<>(element, timestamp);
- }
-
- @Override
- public void emitWatermark(Watermark mark) {
- synchronized (watermarkLock) {
- currentWatermark = mark;
- watermarkLock.notifyAll();
- }
- }
-
- @Override
- public void markAsTemporarilyIdle() {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public Object getCheckpointLock() {
- return checkpointLock;
- }
-
- @Override
- public void close() {}
-
- public StreamRecord<T> getLatestElement() {
- return latestElement;
- }
-
- public boolean hasWatermark() {
- return currentWatermark != null;
- }
-
- public Watermark getLatestWatermark() throws InterruptedException {
- synchronized (watermarkLock) {
- while (currentWatermark == null) {
- watermarkLock.wait();
- }
- Watermark wm = currentWatermark;
- currentWatermark = null;
- return wm;
- }
- }
- }
-
- // ------------------------------------------------------------------------
-
private static class PeriodicTestExtractor implements AssignerWithPeriodicWatermarks<Long> {
private volatile long maxTimestamp = Long.MIN_VALUE;
http://git-wip-us.apache.org/repos/asf/flink/blob/c4bfc7de/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/TestSourceContext.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/TestSourceContext.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/TestSourceContext.java
new file mode 100644
index 0000000..2a96a68
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/TestSourceContext.java
@@ -0,0 +1,87 @@
+/*
+ * 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.testutils;
+
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+/**
+ * Test {@link org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext}.
+ */
+public final class TestSourceContext<T> implements SourceFunction.SourceContext<T> {
+
+ private final Object checkpointLock = new Object();
+ private final Object watermarkLock = new Object();
+
+ private volatile StreamRecord<T> latestElement;
+ private volatile Watermark currentWatermark;
+
+ @Override
+ public void collect(T element) {
+ this.latestElement = new StreamRecord<>(element);
+ }
+
+ @Override
+ public void collectWithTimestamp(T element, long timestamp) {
+ this.latestElement = new StreamRecord<>(element, timestamp);
+ }
+
+ @Override
+ public void emitWatermark(Watermark mark) {
+ synchronized (watermarkLock) {
+ currentWatermark = mark;
+ watermarkLock.notifyAll();
+ }
+ }
+
+ @Override
+ public void markAsTemporarilyIdle() {
+ // do nothing
+ }
+
+ @Override
+ public Object getCheckpointLock() {
+ return checkpointLock;
+ }
+
+ @Override
+ public void close() {
+ // do nothing
+ }
+
+ public StreamRecord<T> getLatestElement() {
+ return latestElement;
+ }
+
+ public boolean hasWatermark() {
+ return currentWatermark != null;
+ }
+
+ public Watermark getLatestWatermark() throws InterruptedException {
+ synchronized (watermarkLock) {
+ while (currentWatermark == null) {
+ watermarkLock.wait();
+ }
+ Watermark wm = currentWatermark;
+ currentWatermark = null;
+ return wm;
+ }
+ }
+}
[4/6] flink git commit: [FLINK-8296] [kafka] Rework
FlinkKafkaConsumerBaseTest to not rely on Java reflection
Posted by tz...@apache.org.
[FLINK-8296] [kafka] Rework FlinkKafkaConsumerBaseTest to not rely on Java reflection
Reflection was mainly used to inject mocks into private fields of the
FlinkKafkaConsumerBase, without the need to fully execute all operator
life cycle methods. This, however, caused the unit tests to be too
implementation-specific.
This commit reworks the FlinkKafkaConsumerBaseTest to remove test
consumer instantiation methods that rely on reflection for dependency
injection. All tests now instantiate dummy test consumers normally, and
let all tests properly execute all operator life cycle methods
regardless of the tested logic.
This closes #5188.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/968683f1
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/968683f1
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/968683f1
Branch: refs/heads/release-1.4
Commit: 968683f15378122478e34bbba6f9c021b8b086ae
Parents: b7f8a5a
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
Authored: Tue Dec 19 16:10:44 2017 -0800
Committer: Tzu-Li (Gordon) Tai <tz...@apache.org>
Committed: Sat Jan 13 11:34:10 2018 +0800
----------------------------------------------------------------------
.../kafka/FlinkKafkaConsumerBase.java | 5 +
.../kafka/FlinkKafkaConsumerBaseTest.java | 532 ++++++++++++-------
2 files changed, 338 insertions(+), 199 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/968683f1/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 865d66c..44e0d54 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
@@ -882,4 +882,9 @@ public abstract class FlinkKafkaConsumerBase<T> extends RichParallelSourceFuncti
OffsetCommitMode getOffsetCommitMode() {
return offsetCommitMode;
}
+
+ @VisibleForTesting
+ LinkedMap getPendingOffsetsToCommit() {
+ return pendingOffsetsToCommit;
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/968683f1/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 6ccfeb1..4361c67 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
@@ -18,17 +18,27 @@
package org.apache.flink.streaming.connectors.kafka;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.accumulators.Accumulator;
+import org.apache.flink.api.common.state.KeyedStateStore;
import org.apache.flink.api.common.state.ListState;
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.StateInitializationContext;
+import org.apache.flink.core.testutils.CheckedThread;
+import org.apache.flink.core.testutils.OneShotLatch;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.runtime.operators.testutils.MockEnvironment;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
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.AbstractStreamOperator;
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;
@@ -44,16 +54,13 @@ import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.SerializedValue;
-import org.apache.commons.collections.map.LinkedMap;
import org.junit.Assert;
import org.junit.Test;
import org.mockito.Matchers;
import org.mockito.Mockito;
import java.io.Serializable;
-import java.lang.reflect.Field;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
@@ -86,20 +93,19 @@ public class FlinkKafkaConsumerBaseTest {
* Tests that not both types of timestamp extractors / watermark generators can be used.
*/
@Test
+ @SuppressWarnings("unchecked")
public void testEitherWatermarkExtractor() {
try {
- new DummyFlinkKafkaConsumer<>().assignTimestampsAndWatermarks((AssignerWithPeriodicWatermarks<Object>) null);
+ new DummyFlinkKafkaConsumer<String>().assignTimestampsAndWatermarks((AssignerWithPeriodicWatermarks<String>) null);
fail();
} catch (NullPointerException ignored) {}
try {
- new DummyFlinkKafkaConsumer<>().assignTimestampsAndWatermarks((AssignerWithPunctuatedWatermarks<Object>) null);
+ new DummyFlinkKafkaConsumer<String>().assignTimestampsAndWatermarks((AssignerWithPunctuatedWatermarks<String>) null);
fail();
} catch (NullPointerException ignored) {}
- @SuppressWarnings("unchecked")
final AssignerWithPeriodicWatermarks<String> periodicAssigner = mock(AssignerWithPeriodicWatermarks.class);
- @SuppressWarnings("unchecked")
final AssignerWithPunctuatedWatermarks<String> punctuatedAssigner = mock(AssignerWithPunctuatedWatermarks.class);
DummyFlinkKafkaConsumer<String> c1 = new DummyFlinkKafkaConsumer<>();
@@ -123,17 +129,16 @@ public class FlinkKafkaConsumerBaseTest {
@Test
public void ignoreCheckpointWhenNotRunning() throws Exception {
@SuppressWarnings("unchecked")
- final AbstractFetcher<String, ?> fetcher = mock(AbstractFetcher.class);
+ final FlinkKafkaConsumerBase<String> consumer = new DummyFlinkKafkaConsumer<>();
- FlinkKafkaConsumerBase<String> consumer = getConsumer(fetcher, new LinkedMap(), false);
- OperatorStateStore operatorStateStore = mock(OperatorStateStore.class);
- TestingListState<Tuple2<KafkaTopicPartition, Long>> listState = new TestingListState<>();
- when(operatorStateStore.getListState(Matchers.any(ListStateDescriptor.class))).thenReturn(listState);
+ final TestingListState<Tuple2<KafkaTopicPartition, Long>> listState = new TestingListState<>();
+ setupConsumer(consumer, false, listState, true, 0, 1);
+ // snapshot before the fetcher starts running
consumer.snapshotState(new StateSnapshotContextSynchronousImpl(1, 1));
+ // no state should have been checkpointed
assertFalse(listState.get().iterator().hasNext());
- consumer.notifyCheckpointComplete(66L);
}
/**
@@ -142,32 +147,13 @@ public class FlinkKafkaConsumerBaseTest {
*/
@Test
public void checkRestoredCheckpointWhenFetcherNotReady() throws Exception {
- OperatorStateStore operatorStateStore = mock(OperatorStateStore.class);
-
- 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);
- when(context.getNumberOfParallelSubtasks()).thenReturn(1);
- when(context.getIndexOfThisSubtask()).thenReturn(0);
- consumer.setRuntimeContext(context);
-
- // 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(restoredListState);
-
- StateInitializationContext initializationContext = mock(StateInitializationContext.class);
-
- when(initializationContext.getOperatorStateStore()).thenReturn(operatorStateStore);
- when(initializationContext.isRestored()).thenReturn(true);
-
- consumer.initializeState(initializationContext);
+ @SuppressWarnings("unchecked")
+ final FlinkKafkaConsumerBase<String> consumer = new DummyFlinkKafkaConsumer<>();
- consumer.open(new Configuration());
+ final TestingListState<Tuple2<KafkaTopicPartition, Long>> restoredListState = new TestingListState<>();
+ setupConsumer(consumer, true, restoredListState, true, 0, 1);
+ // snapshot before the fetcher starts running
consumer.snapshotState(new StateSnapshotContextSynchronousImpl(17, 17));
// ensure that the list was cleared and refilled. while this is an implementation detail, we use it here
@@ -192,67 +178,68 @@ public class FlinkKafkaConsumerBaseTest {
@Test
public void testConfigureOnCheckpointsCommitMode() throws Exception {
+ @SuppressWarnings("unchecked")
+ // auto-commit enabled; this should be ignored in this case
+ final DummyFlinkKafkaConsumer<String> consumer = new DummyFlinkKafkaConsumer<>(true);
- DummyFlinkKafkaConsumer consumer = new DummyFlinkKafkaConsumer();
- consumer.setIsAutoCommitEnabled(true); // this should be ignored
-
- StreamingRuntimeContext context = mock(StreamingRuntimeContext.class);
- when(context.getIndexOfThisSubtask()).thenReturn(0);
- when(context.getNumberOfParallelSubtasks()).thenReturn(1);
- when(context.isCheckpointingEnabled()).thenReturn(true); // enable checkpointing, auto commit should be ignored
- consumer.setRuntimeContext(context);
+ setupConsumer(
+ consumer,
+ false,
+ null,
+ true, // enable checkpointing; auto commit should be ignored
+ 0,
+ 1);
- consumer.open(new Configuration());
assertEquals(OffsetCommitMode.ON_CHECKPOINTS, consumer.getOffsetCommitMode());
}
@Test
public void testConfigureAutoCommitMode() throws Exception {
+ @SuppressWarnings("unchecked")
+ final DummyFlinkKafkaConsumer<String> consumer = new DummyFlinkKafkaConsumer<>(true);
- DummyFlinkKafkaConsumer consumer = new DummyFlinkKafkaConsumer();
- consumer.setIsAutoCommitEnabled(true);
-
- StreamingRuntimeContext context = mock(StreamingRuntimeContext.class);
- when(context.getIndexOfThisSubtask()).thenReturn(0);
- when(context.getNumberOfParallelSubtasks()).thenReturn(1);
- when(context.isCheckpointingEnabled()).thenReturn(false); // disable checkpointing, auto commit should be respected
- consumer.setRuntimeContext(context);
+ setupConsumer(
+ consumer,
+ false,
+ null,
+ false, // disable checkpointing; auto commit should be respected
+ 0,
+ 1);
- consumer.open(new Configuration());
assertEquals(OffsetCommitMode.KAFKA_PERIODIC, consumer.getOffsetCommitMode());
}
@Test
public void testConfigureDisableOffsetCommitWithCheckpointing() throws Exception {
-
- DummyFlinkKafkaConsumer consumer = new DummyFlinkKafkaConsumer();
- consumer.setIsAutoCommitEnabled(true); // this should be ignored
-
- StreamingRuntimeContext context = mock(StreamingRuntimeContext.class);
- when(context.getIndexOfThisSubtask()).thenReturn(0);
- when(context.getNumberOfParallelSubtasks()).thenReturn(1);
- when(context.isCheckpointingEnabled()).thenReturn(true); // enable checkpointing, auto commit should be ignored
- consumer.setRuntimeContext(context);
-
+ @SuppressWarnings("unchecked")
+ // auto-commit enabled; this should be ignored in this case
+ final DummyFlinkKafkaConsumer<String> consumer = new DummyFlinkKafkaConsumer<>(true);
consumer.setCommitOffsetsOnCheckpoints(false); // disabling offset committing should override everything
- consumer.open(new Configuration());
+ setupConsumer(
+ consumer,
+ false,
+ null,
+ true, // enable checkpointing; auto commit should be ignored
+ 0,
+ 1);
+
assertEquals(OffsetCommitMode.DISABLED, consumer.getOffsetCommitMode());
}
@Test
public void testConfigureDisableOffsetCommitWithoutCheckpointing() throws Exception {
+ @SuppressWarnings("unchecked")
+ final DummyFlinkKafkaConsumer<String> consumer = new DummyFlinkKafkaConsumer<>(false);
- DummyFlinkKafkaConsumer consumer = new DummyFlinkKafkaConsumer();
- consumer.setIsAutoCommitEnabled(false);
-
- StreamingRuntimeContext context = mock(StreamingRuntimeContext.class);
- when(context.getIndexOfThisSubtask()).thenReturn(0);
- when(context.getNumberOfParallelSubtasks()).thenReturn(1);
- when(context.isCheckpointingEnabled()).thenReturn(false); // disable checkpointing, auto commit should be respected
- consumer.setRuntimeContext(context);
+ setupConsumer(
+ consumer,
+ false,
+ null,
+ false, // disable checkpointing; auto commit should be respected
+ 0,
+ 1);
- consumer.open(new Configuration());
assertEquals(OffsetCommitMode.DISABLED, consumer.getOffsetCommitMode());
}
@@ -278,36 +265,37 @@ public class FlinkKafkaConsumerBaseTest {
// --------------------------------------------------------------------
- final AbstractFetcher<String, ?> fetcher = mock(AbstractFetcher.class);
+ final OneShotLatch runLatch = new OneShotLatch();
+ final OneShotLatch stopLatch = new OneShotLatch();
+ final AbstractFetcher<String, ?> fetcher = getRunnableMockFetcher(runLatch, stopLatch);
when(fetcher.snapshotCurrentState()).thenReturn(state1, state2, state3);
- final LinkedMap pendingOffsetsToCommit = new LinkedMap();
-
- FlinkKafkaConsumerBase<String> consumer = getConsumer(fetcher, pendingOffsetsToCommit, true);
- StreamingRuntimeContext mockRuntimeContext = mock(StreamingRuntimeContext.class);
- when(mockRuntimeContext.isCheckpointingEnabled()).thenReturn(true); // enable checkpointing
- when(mockRuntimeContext.getIndexOfThisSubtask()).thenReturn(0);
- when(mockRuntimeContext.getNumberOfParallelSubtasks()).thenReturn(1);
- consumer.setRuntimeContext(mockRuntimeContext);
-
- assertEquals(0, pendingOffsetsToCommit.size());
-
- OperatorStateStore backend = mock(OperatorStateStore.class);
+ final FlinkKafkaConsumerBase<String> consumer = new DummyFlinkKafkaConsumer<>(
+ fetcher,
+ mock(AbstractPartitionDiscoverer.class),
+ false);
- TestingListState<Serializable> listState = new TestingListState<>();
- // mock old 1.2 state (empty)
- when(backend.getSerializableListState(Matchers.any(String.class))).thenReturn(new TestingListState<Serializable>());
- // mock 1.3 state
- when(backend.getUnionListState(Matchers.any(ListStateDescriptor.class))).thenReturn(listState);
+ final TestingListState<Serializable> listState = new TestingListState<>();
- StateInitializationContext initializationContext = mock(StateInitializationContext.class);
+ // setup and run the consumer; wait until the consumer reaches the main fetch loop before continuing test
+ setupConsumer(consumer, false, listState, true, 0, 1);
- when(initializationContext.getOperatorStateStore()).thenReturn(backend);
- when(initializationContext.isRestored()).thenReturn(false, true, true, true);
+ final CheckedThread runThread = new CheckedThread() {
+ @Override
+ public void go() throws Exception {
+ consumer.run(mock(SourceFunction.SourceContext.class));
+ }
- consumer.initializeState(initializationContext);
+ @Override
+ public void sync() throws Exception {
+ stopLatch.trigger();
+ super.sync();
+ }
+ };
+ runThread.start();
+ runLatch.await();
- consumer.open(new Configuration());
+ assertEquals(0, consumer.getPendingOffsetsToCommit().size());
// checkpoint 1
consumer.snapshotState(new StateSnapshotContextSynchronousImpl(138, 138));
@@ -320,8 +308,8 @@ public class FlinkKafkaConsumerBaseTest {
}
assertEquals(state1, snapshot1);
- assertEquals(1, pendingOffsetsToCommit.size());
- assertEquals(state1, pendingOffsetsToCommit.get(138L));
+ assertEquals(1, consumer.getPendingOffsetsToCommit().size());
+ assertEquals(state1, consumer.getPendingOffsetsToCommit().get(138L));
// checkpoint 2
consumer.snapshotState(new StateSnapshotContextSynchronousImpl(140, 140));
@@ -334,13 +322,13 @@ public class FlinkKafkaConsumerBaseTest {
}
assertEquals(state2, snapshot2);
- assertEquals(2, pendingOffsetsToCommit.size());
- assertEquals(state2, pendingOffsetsToCommit.get(140L));
+ assertEquals(2, consumer.getPendingOffsetsToCommit().size());
+ assertEquals(state2, consumer.getPendingOffsetsToCommit().get(140L));
// ack checkpoint 1
consumer.notifyCheckpointComplete(138L);
- assertEquals(1, pendingOffsetsToCommit.size());
- assertTrue(pendingOffsetsToCommit.containsKey(140L));
+ assertEquals(1, consumer.getPendingOffsetsToCommit().size());
+ assertTrue(consumer.getPendingOffsetsToCommit().containsKey(140L));
// checkpoint 3
consumer.snapshotState(new StateSnapshotContextSynchronousImpl(141, 141));
@@ -353,37 +341,35 @@ public class FlinkKafkaConsumerBaseTest {
}
assertEquals(state3, snapshot3);
- assertEquals(2, pendingOffsetsToCommit.size());
- assertEquals(state3, pendingOffsetsToCommit.get(141L));
+ assertEquals(2, consumer.getPendingOffsetsToCommit().size());
+ assertEquals(state3, consumer.getPendingOffsetsToCommit().get(141L));
// ack checkpoint 3, subsumes number 2
consumer.notifyCheckpointComplete(141L);
- assertEquals(0, pendingOffsetsToCommit.size());
+ assertEquals(0, consumer.getPendingOffsetsToCommit().size());
consumer.notifyCheckpointComplete(666); // invalid checkpoint
- assertEquals(0, pendingOffsetsToCommit.size());
-
- OperatorStateStore operatorStateStore = mock(OperatorStateStore.class);
- listState = new TestingListState<>();
- when(operatorStateStore.getListState(Matchers.any(ListStateDescriptor.class))).thenReturn(listState);
+ assertEquals(0, consumer.getPendingOffsetsToCommit().size());
// create 500 snapshots
for (int i = 100; i < 600; i++) {
consumer.snapshotState(new StateSnapshotContextSynchronousImpl(i, i));
listState.clear();
}
- assertEquals(FlinkKafkaConsumerBase.MAX_NUM_PENDING_CHECKPOINTS, pendingOffsetsToCommit.size());
+ assertEquals(FlinkKafkaConsumerBase.MAX_NUM_PENDING_CHECKPOINTS, consumer.getPendingOffsetsToCommit().size());
// commit only the second last
consumer.notifyCheckpointComplete(598);
- assertEquals(1, pendingOffsetsToCommit.size());
+ assertEquals(1, consumer.getPendingOffsetsToCommit().size());
// access invalid checkpoint
consumer.notifyCheckpointComplete(590);
// and the last
consumer.notifyCheckpointComplete(599);
- assertEquals(0, pendingOffsetsToCommit.size());
+ assertEquals(0, consumer.getPendingOffsetsToCommit().size());
+
+ runThread.sync();
}
@Test
@@ -407,38 +393,38 @@ public class FlinkKafkaConsumerBaseTest {
// --------------------------------------------------------------------
- final AbstractFetcher<String, ?> fetcher = mock(AbstractFetcher.class);
+ final OneShotLatch runLatch = new OneShotLatch();
+ final OneShotLatch stopLatch = new OneShotLatch();
+ final AbstractFetcher<String, ?> fetcher = getRunnableMockFetcher(runLatch, stopLatch);
when(fetcher.snapshotCurrentState()).thenReturn(state1, state2, state3);
- final LinkedMap pendingOffsetsToCommit = new LinkedMap();
-
- FlinkKafkaConsumerBase<String> consumer = getConsumer(fetcher, pendingOffsetsToCommit, true);
- StreamingRuntimeContext mockRuntimeContext = mock(StreamingRuntimeContext.class);
- when(mockRuntimeContext.isCheckpointingEnabled()).thenReturn(true); // enable checkpointing
- when(mockRuntimeContext.getIndexOfThisSubtask()).thenReturn(0);
- when(mockRuntimeContext.getNumberOfParallelSubtasks()).thenReturn(1);
- consumer.setRuntimeContext(mockRuntimeContext);
-
+ final FlinkKafkaConsumerBase<String> consumer = new DummyFlinkKafkaConsumer<>(
+ fetcher,
+ mock(AbstractPartitionDiscoverer.class),
+ false);
consumer.setCommitOffsetsOnCheckpoints(false); // disable offset committing
- assertEquals(0, pendingOffsetsToCommit.size());
-
- OperatorStateStore backend = mock(OperatorStateStore.class);
+ final TestingListState<Serializable> listState = new TestingListState<>();
- TestingListState<Serializable> listState = new TestingListState<>();
- // mock old 1.2 state (empty)
- when(backend.getSerializableListState(Matchers.any(String.class))).thenReturn(new TestingListState<Serializable>());
- // mock 1.3 state
- when(backend.getUnionListState(Matchers.any(ListStateDescriptor.class))).thenReturn(listState);
+ // setup and run the consumer; wait until the consumer reaches the main fetch loop before continuing test
+ setupConsumer(consumer, false, listState, true, 0, 1);
- StateInitializationContext initializationContext = mock(StateInitializationContext.class);
-
- when(initializationContext.getOperatorStateStore()).thenReturn(backend);
- when(initializationContext.isRestored()).thenReturn(false, true, true, true);
+ final CheckedThread runThread = new CheckedThread() {
+ @Override
+ public void go() throws Exception {
+ consumer.run(mock(SourceFunction.SourceContext.class));
+ }
- consumer.initializeState(initializationContext);
+ @Override
+ public void sync() throws Exception {
+ stopLatch.trigger();
+ super.sync();
+ }
+ };
+ runThread.start();
+ runLatch.await();
- consumer.open(new Configuration());
+ assertEquals(0, consumer.getPendingOffsetsToCommit().size());
// checkpoint 1
consumer.snapshotState(new StateSnapshotContextSynchronousImpl(138, 138));
@@ -451,7 +437,7 @@ public class FlinkKafkaConsumerBaseTest {
}
assertEquals(state1, snapshot1);
- assertEquals(0, pendingOffsetsToCommit.size()); // pending offsets to commit should not be updated
+ assertEquals(0, consumer.getPendingOffsetsToCommit().size()); // pending offsets to commit should not be updated
// checkpoint 2
consumer.snapshotState(new StateSnapshotContextSynchronousImpl(140, 140));
@@ -464,7 +450,7 @@ public class FlinkKafkaConsumerBaseTest {
}
assertEquals(state2, snapshot2);
- assertEquals(0, pendingOffsetsToCommit.size()); // pending offsets to commit should not be updated
+ assertEquals(0, consumer.getPendingOffsetsToCommit().size()); // pending offsets to commit should not be updated
// ack checkpoint 1
consumer.notifyCheckpointComplete(138L);
@@ -481,7 +467,7 @@ public class FlinkKafkaConsumerBaseTest {
}
assertEquals(state3, snapshot3);
- assertEquals(0, pendingOffsetsToCommit.size()); // pending offsets to commit should not be updated
+ assertEquals(0, consumer.getPendingOffsetsToCommit().size()); // pending offsets to commit should not be updated
// ack checkpoint 3, subsumes number 2
consumer.notifyCheckpointComplete(141L);
@@ -490,16 +476,12 @@ public class FlinkKafkaConsumerBaseTest {
consumer.notifyCheckpointComplete(666); // invalid checkpoint
verify(fetcher, never()).commitInternalOffsetsToKafka(anyMap(), Matchers.any(KafkaCommitCallback.class)); // no offsets should be committed
- OperatorStateStore operatorStateStore = mock(OperatorStateStore.class);
- listState = new TestingListState<>();
- when(operatorStateStore.getListState(Matchers.any(ListStateDescriptor.class))).thenReturn(listState);
-
// create 500 snapshots
for (int i = 100; i < 600; i++) {
consumer.snapshotState(new StateSnapshotContextSynchronousImpl(i, i));
listState.clear();
}
- assertEquals(0, pendingOffsetsToCommit.size()); // pending offsets to commit should not be updated
+ assertEquals(0, consumer.getPendingOffsetsToCommit().size()); // pending offsets to commit should not be updated
// commit only the second last
consumer.notifyCheckpointComplete(598);
@@ -532,7 +514,7 @@ public class FlinkKafkaConsumerBaseTest {
* of topics fetched from Kafka.
*/
@SuppressWarnings("unchecked")
- void testRescaling(
+ private void testRescaling(
final int initialParallelism,
final int numPartitions,
final int restoredParallelism,
@@ -554,8 +536,14 @@ public class FlinkKafkaConsumerBaseTest {
new AbstractStreamOperatorTestHarness[initialParallelism];
for (int i = 0; i < initialParallelism; i++) {
- consumers[i] = new DummyFlinkKafkaConsumer<>(
- Collections.singletonList("test-topic"), mockFetchedPartitionsOnStartup);
+ TestPartitionDiscoverer partitionDiscoverer = new TestPartitionDiscoverer(
+ new KafkaTopicsDescriptor(Collections.singletonList("test-topic"), null),
+ i,
+ initialParallelism,
+ TestPartitionDiscoverer.createMockGetAllTopicsSequenceFromFixedReturn(Collections.singletonList("test-topic")),
+ TestPartitionDiscoverer.createMockGetAllPartitionsFromTopicsSequenceFromFixedReturn(mockFetchedPartitionsOnStartup));
+
+ consumers[i] = new DummyFlinkKafkaConsumer<>(mock(AbstractFetcher.class), partitionDiscoverer, false);
testHarnesses[i] = createTestHarness(consumers[i], initialParallelism, i);
// initializeState() is always called, null signals that we didn't restore
@@ -602,8 +590,14 @@ public class FlinkKafkaConsumerBaseTest {
new AbstractStreamOperatorTestHarness[restoredParallelism];
for (int i = 0; i < restoredParallelism; i++) {
- restoredConsumers[i] = new DummyFlinkKafkaConsumer<>(
- Collections.singletonList("test-topic"), mockFetchedPartitionsAfterRestore);
+ TestPartitionDiscoverer partitionDiscoverer = new TestPartitionDiscoverer(
+ new KafkaTopicsDescriptor(Collections.singletonList("test-topic"), null),
+ i,
+ restoredParallelism,
+ TestPartitionDiscoverer.createMockGetAllTopicsSequenceFromFixedReturn(Collections.singletonList("test-topic")),
+ TestPartitionDiscoverer.createMockGetAllPartitionsFromTopicsSequenceFromFixedReturn(mockFetchedPartitionsAfterRestore));
+
+ restoredConsumers[i] = new DummyFlinkKafkaConsumer<>(mock(AbstractFetcher.class), partitionDiscoverer, false);
restoredTestHarnesses[i] = createTestHarness(restoredConsumers[i], restoredParallelism, i);
// initializeState() is always called, null signals that we didn't restore
@@ -630,28 +624,6 @@ public class FlinkKafkaConsumerBaseTest {
// ------------------------------------------------------------------------
- private static <T> FlinkKafkaConsumerBase<T> getConsumer(
- AbstractFetcher<T, ?> fetcher, LinkedMap pendingOffsetsToCommit, boolean running) throws Exception {
- FlinkKafkaConsumerBase<T> consumer = new DummyFlinkKafkaConsumer<>();
- StreamingRuntimeContext mockRuntimeContext = mock(StreamingRuntimeContext.class);
- Mockito.when(mockRuntimeContext.isCheckpointingEnabled()).thenReturn(true);
- consumer.setRuntimeContext(mockRuntimeContext);
-
- Field fetcherField = FlinkKafkaConsumerBase.class.getDeclaredField("kafkaFetcher");
- fetcherField.setAccessible(true);
- fetcherField.set(consumer, fetcher);
-
- Field mapField = FlinkKafkaConsumerBase.class.getDeclaredField("pendingOffsetsToCommit");
- mapField.setAccessible(true);
- mapField.set(consumer, pendingOffsetsToCommit);
-
- Field runningField = FlinkKafkaConsumerBase.class.getDeclaredField("running");
- runningField.setAccessible(true);
- runningField.set(consumer, running);
-
- return consumer;
- }
-
private static <T> AbstractStreamOperatorTestHarness<T> createTestHarness(
SourceFunction<T> source, int numSubtasks, int subtaskIndex) throws Exception {
@@ -667,25 +639,43 @@ public class FlinkKafkaConsumerBaseTest {
// ------------------------------------------------------------------------
+ /**
+ * An instantiable dummy {@link FlinkKafkaConsumerBase} that supports injecting
+ * mocks for {@link FlinkKafkaConsumerBase#kafkaFetcher}, {@link FlinkKafkaConsumerBase#partitionDiscoverer},
+ * and {@link FlinkKafkaConsumerBase#getIsAutoCommitEnabled()}.
+ */
private static class DummyFlinkKafkaConsumer<T> extends FlinkKafkaConsumerBase<T> {
private static final long serialVersionUID = 1L;
- boolean isAutoCommitEnabled = false;
+ private AbstractFetcher<T, ?> testFetcher;
+ private AbstractPartitionDiscoverer testPartitionDiscoverer;
+ private boolean isAutoCommitEnabled;
- private List<String> fixedMockGetAllTopicsReturnSequence;
- private List<KafkaTopicPartition> fixedMockGetAllPartitionsForTopicsReturnSequence;
+ @SuppressWarnings("unchecked")
+ DummyFlinkKafkaConsumer() {
+ this(false);
+ }
- public DummyFlinkKafkaConsumer() {
- this(Collections.singletonList("dummy-topic"), Collections.singletonList(new KafkaTopicPartition("dummy-topic", 0)));
+ @SuppressWarnings("unchecked")
+ DummyFlinkKafkaConsumer(boolean isAutoCommitEnabled) {
+ this(mock(AbstractFetcher.class), mock(AbstractPartitionDiscoverer.class), isAutoCommitEnabled);
}
@SuppressWarnings("unchecked")
- public DummyFlinkKafkaConsumer(
- List<String> fixedMockGetAllTopicsReturnSequence,
- List<KafkaTopicPartition> fixedMockGetAllPartitionsForTopicsReturnSequence) {
- super(Arrays.asList("dummy-topic"), null, (KeyedDeserializationSchema < T >) mock(KeyedDeserializationSchema.class), 0);
- this.fixedMockGetAllTopicsReturnSequence = Preconditions.checkNotNull(fixedMockGetAllTopicsReturnSequence);
- this.fixedMockGetAllPartitionsForTopicsReturnSequence = Preconditions.checkNotNull(fixedMockGetAllPartitionsForTopicsReturnSequence);
+ DummyFlinkKafkaConsumer(
+ AbstractFetcher<T, ?> testFetcher,
+ AbstractPartitionDiscoverer testPartitionDiscoverer,
+ boolean isAutoCommitEnabled) {
+
+ super(
+ Collections.singletonList("dummy-topic"),
+ null,
+ (KeyedDeserializationSchema < T >) mock(KeyedDeserializationSchema.class),
+ PARTITION_DISCOVERY_DISABLED);
+
+ this.testFetcher = testFetcher;
+ this.testPartitionDiscoverer = testPartitionDiscoverer;
+ this.isAutoCommitEnabled = isAutoCommitEnabled;
}
@Override
@@ -697,7 +687,7 @@ public class FlinkKafkaConsumerBaseTest {
SerializedValue<AssignerWithPunctuatedWatermarks<T>> watermarksPunctuated,
StreamingRuntimeContext runtimeContext,
OffsetCommitMode offsetCommitMode) throws Exception {
- return mock(AbstractFetcher.class);
+ return this.testFetcher;
}
@Override
@@ -705,21 +695,12 @@ public class FlinkKafkaConsumerBaseTest {
KafkaTopicsDescriptor topicsDescriptor,
int indexOfThisSubtask,
int numParallelSubtasks) {
- return new TestPartitionDiscoverer(
- topicsDescriptor,
- indexOfThisSubtask,
- numParallelSubtasks,
- TestPartitionDiscoverer.createMockGetAllTopicsSequenceFromFixedReturn(fixedMockGetAllTopicsReturnSequence),
- TestPartitionDiscoverer.createMockGetAllPartitionsFromTopicsSequenceFromFixedReturn(fixedMockGetAllPartitionsForTopicsReturnSequence));
+ return this.testPartitionDiscoverer;
}
@Override
protected boolean getIsAutoCommitEnabled() {
- return isAutoCommitEnabled;
- }
-
- public void setIsAutoCommitEnabled(boolean isAutoCommitEnabled) {
- this.isAutoCommitEnabled = isAutoCommitEnabled;
+ return this.isAutoCommitEnabled;
}
}
@@ -748,8 +729,161 @@ public class FlinkKafkaConsumerBaseTest {
return list;
}
- public boolean isClearCalled() {
+ boolean isClearCalled() {
return clearCalled;
}
}
+
+ /**
+ * Returns a mock {@link AbstractFetcher}, with run / stop latches injected in
+ * the {@link AbstractFetcher#runFetchLoop()} method.
+ */
+ private static <T> AbstractFetcher<T, ?> getRunnableMockFetcher(
+ OneShotLatch runLatch,
+ OneShotLatch stopLatch) throws Exception {
+
+ @SuppressWarnings("unchecked")
+ final AbstractFetcher<T, ?> fetcher = mock(AbstractFetcher.class);
+
+ Mockito.doAnswer(invocationOnMock -> {
+ runLatch.trigger();
+ stopLatch.await();
+ return null;
+ }).when(fetcher).runFetchLoop();
+
+ return fetcher;
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <T, S> void setupConsumer(
+ FlinkKafkaConsumerBase<T> consumer,
+ boolean isRestored,
+ ListState<S> restoredListState,
+ boolean isCheckpointingEnabled,
+ int subtaskIndex,
+ int totalNumSubtasks) throws Exception {
+
+ // run setup procedure in operator life cycle
+ consumer.setRuntimeContext(new MockRuntimeContext(isCheckpointingEnabled, totalNumSubtasks, subtaskIndex));
+ consumer.initializeState(new MockFunctionInitializationContext(isRestored, new MockOperatorStateStore(restoredListState)));
+ consumer.open(new Configuration());
+ }
+
+ private static class MockRuntimeContext extends StreamingRuntimeContext {
+
+ private final boolean isCheckpointingEnabled;
+
+ private final int numParallelSubtasks;
+ private final int subtaskIndex;
+
+ private MockRuntimeContext(
+ boolean isCheckpointingEnabled,
+ int numParallelSubtasks,
+ int subtaskIndex) {
+
+ super(
+ new MockStreamOperator(),
+ new MockEnvironment("mockTask", 4 * MemoryManager.DEFAULT_PAGE_SIZE, null, 16),
+ Collections.<String, Accumulator<?, ?>>emptyMap());
+
+ this.isCheckpointingEnabled = isCheckpointingEnabled;
+ this.numParallelSubtasks = numParallelSubtasks;
+ this.subtaskIndex = subtaskIndex;
+ }
+
+ @Override
+ public MetricGroup getMetricGroup() {
+ return new UnregisteredMetricsGroup();
+ }
+
+ @Override
+ public boolean isCheckpointingEnabled() {
+ return isCheckpointingEnabled;
+ }
+
+ @Override
+ public int getIndexOfThisSubtask() {
+ return subtaskIndex;
+ }
+
+ @Override
+ public int getNumberOfParallelSubtasks() {
+ return numParallelSubtasks;
+ }
+
+ // ------------------------------------------------------------------------
+
+ private static class MockStreamOperator extends AbstractStreamOperator<Integer> {
+ private static final long serialVersionUID = -1153976702711944427L;
+
+ @Override
+ public ExecutionConfig getExecutionConfig() {
+ return new ExecutionConfig();
+ }
+ }
+ }
+
+ private static class MockOperatorStateStore implements OperatorStateStore {
+
+ private final ListState<?> mockRestoredUnionListState;
+
+ private MockOperatorStateStore(ListState<?> restoredUnionListState) {
+ this.mockRestoredUnionListState = restoredUnionListState;
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <S> ListState<S> getUnionListState(ListStateDescriptor<S> stateDescriptor) throws Exception {
+ return (ListState<S>) mockRestoredUnionListState;
+ }
+
+ @Override
+ public <T extends Serializable> ListState<T> getSerializableListState(String stateName) throws Exception {
+ // return empty state for the legacy 1.2 Kafka consumer state
+ return new TestingListState<>();
+ }
+
+ // ------------------------------------------------------------------------
+
+ @Override
+ public <S> ListState<S> getOperatorState(ListStateDescriptor<S> stateDescriptor) throws Exception {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public <S> ListState<S> getListState(ListStateDescriptor<S> stateDescriptor) throws Exception {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Set<String> getRegisteredStateNames() {
+ throw new UnsupportedOperationException();
+ }
+ }
+
+ private static class MockFunctionInitializationContext implements FunctionInitializationContext {
+
+ private final boolean isRestored;
+ private final OperatorStateStore operatorStateStore;
+
+ private MockFunctionInitializationContext(boolean isRestored, OperatorStateStore operatorStateStore) {
+ this.isRestored = isRestored;
+ this.operatorStateStore = operatorStateStore;
+ }
+
+ @Override
+ public boolean isRestored() {
+ return isRestored;
+ }
+
+ @Override
+ public OperatorStateStore getOperatorStateStore() {
+ return operatorStateStore;
+ }
+
+ @Override
+ public KeyedStateStore getKeyedStateStore() {
+ throw new UnsupportedOperationException();
+ }
+ }
}