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/02/06 19:03:14 UTC
[14/28] flink git commit: [FLINK-8484] [kinesis] Use an equivalence
wrapper around restored StreamShardMetadatas to bypass out-of-sync shard
infos
[FLINK-8484] [kinesis] Use an equivalence wrapper around restored StreamShardMetadatas to bypass out-of-sync shard infos
This closes #5337.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/7884a4f7
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/7884a4f7
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/7884a4f7
Branch: refs/heads/release-1.4
Commit: 7884a4f749b8e7ba7b831963cf2a3be12df877b7
Parents: f887575
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
Authored: Thu Feb 1 12:53:42 2018 +0100
Committer: Tzu-Li (Gordon) Tai <tz...@apache.org>
Committed: Tue Feb 6 17:31:32 2018 +0100
----------------------------------------------------------------------
.../kinesis/FlinkKinesisConsumer.java | 76 +++-------
.../kinesis/internals/KinesisDataFetcher.java | 5 +
.../kinesis/model/StreamShardMetadata.java | 44 ++++++
.../FlinkKinesisConsumerMigrationTest.java | 145 ++++++++++++++++++-
.../kinesis/FlinkKinesisConsumerTest.java | 52 -------
5 files changed, 209 insertions(+), 113 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/7884a4f7/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisConsumer.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisConsumer.java b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisConsumer.java
index de66f93..bf5f779 100644
--- a/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisConsumer.java
+++ b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisConsumer.java
@@ -99,7 +99,7 @@ public class FlinkKinesisConsumer<T> extends RichParallelSourceFunction<T> imple
private transient KinesisDataFetcher<T> fetcher;
/** The sequence numbers to restore to upon restore from failure. */
- private transient HashMap<StreamShardMetadata, SequenceNumber> sequenceNumsToRestore;
+ private transient HashMap<StreamShardMetadata.EquivalenceWrapper, SequenceNumber> sequenceNumsToRestore;
private volatile boolean running = true;
@@ -206,23 +206,16 @@ public class FlinkKinesisConsumer<T> extends RichParallelSourceFunction<T> imple
List<StreamShardHandle> allShards = fetcher.discoverNewShardsToSubscribe();
for (StreamShardHandle shard : allShards) {
- StreamShardMetadata kinesisStreamShard = KinesisDataFetcher.convertToStreamShardMetadata(shard);
+ StreamShardMetadata.EquivalenceWrapper kinesisStreamShard =
+ new StreamShardMetadata.EquivalenceWrapper(KinesisDataFetcher.convertToStreamShardMetadata(shard));
+
if (sequenceNumsToRestore != null) {
- // We need to do this to make sure that a shard that was closed after this restored state was taken will be properly
- // detected and have its sequence numbers restored. A shard will be closed when re-sharding, which can happen when
- // scaling up & down the Kinesis stream, and if the state is not synchronized, then the equality check of the current
- // Kinesis shard will not match the stored state, which will cause us to re-read the entire shard from the event horizon.
- if (updateKinesisShardStateWithMissingEndingSequenceNumber(kinesisStreamShard, sequenceNumsToRestore)) {
- if (LOG.isInfoEnabled()) {
- LOG.info("Updated local stored state for shard {} with a new ending number: {}", kinesisStreamShard.getShardId(), sequenceNumsToRestore.get(kinesisStreamShard));
- }
- }
if (sequenceNumsToRestore.containsKey(kinesisStreamShard)) {
// if the shard was already seen and is contained in the state,
// just use the sequence number stored in the state
fetcher.registerNewSubscribedShardState(
- new KinesisStreamShardState(kinesisStreamShard, shard, sequenceNumsToRestore.get(kinesisStreamShard)));
+ new KinesisStreamShardState(kinesisStreamShard.getShardMetadata(), shard, sequenceNumsToRestore.get(kinesisStreamShard)));
if (LOG.isInfoEnabled()) {
LOG.info("Subtask {} is seeding the fetcher with restored shard {}," +
@@ -232,7 +225,7 @@ public class FlinkKinesisConsumer<T> extends RichParallelSourceFunction<T> imple
} else {
// the shard wasn't discovered in the previous run, therefore should be consumed from the beginning
fetcher.registerNewSubscribedShardState(
- new KinesisStreamShardState(kinesisStreamShard, shard, SentinelSequenceNumber.SENTINEL_EARLIEST_SEQUENCE_NUM.get()));
+ new KinesisStreamShardState(kinesisStreamShard.getShardMetadata(), shard, SentinelSequenceNumber.SENTINEL_EARLIEST_SEQUENCE_NUM.get()));
if (LOG.isInfoEnabled()) {
LOG.info("Subtask {} is seeding the fetcher with new discovered shard {}," +
@@ -248,7 +241,7 @@ public class FlinkKinesisConsumer<T> extends RichParallelSourceFunction<T> imple
ConsumerConfigConstants.DEFAULT_STREAM_INITIAL_POSITION)).toSentinelSequenceNumber();
fetcher.registerNewSubscribedShardState(
- new KinesisStreamShardState(kinesisStreamShard, shard, startingSeqNum.get()));
+ new KinesisStreamShardState(kinesisStreamShard.getShardMetadata(), shard, startingSeqNum.get()));
if (LOG.isInfoEnabled()) {
LOG.info("Subtask {} will be seeded with initial shard {}, starting state set as sequence number {}",
@@ -275,45 +268,6 @@ public class FlinkKinesisConsumer<T> extends RichParallelSourceFunction<T> imple
sourceContext.close();
}
- /**
- * Synchronizes the Kinesis shard information from the current Kinesis shard with the restored state, if we find
- * a shard that match the shardId and streamName. If we find one, and its ending key is different that what we
- * have in our stored state, then we update the stored's shard's metadata's ending number.
- *
- * @param current the current Kinesis shard we're trying to synchronize.
- * @param sequenceNumsToRestore the (re)stored shard metadata and their sequence numbers.
- * @return {@code true} if the local state was updated with the current Kinesis shard's ending number.
- */
- @VisibleForTesting
- boolean updateKinesisShardStateWithMissingEndingSequenceNumber(StreamShardMetadata current, HashMap<StreamShardMetadata, SequenceNumber> sequenceNumsToRestore) {
- checkNotNull(current.getStreamName(), "Stream name not set on the current metadata shard");
- checkNotNull(current.getShardId(), "Shard id not set on the current metadata shard");
-
- // short-circuit: if the current shard doesn't have an ending sequence number, then there's no point in trying to update the local state
- // since that's the only property that can change.
- if (current.getEndingSequenceNumber() == null) {
- return false;
- }
-
- // try to find the matching shard based on the id & stream name
- for (Map.Entry<StreamShardMetadata, SequenceNumber> entry : sequenceNumsToRestore.entrySet()) {
- if (current.getStreamName().equals(entry.getKey().getStreamName())
- && current.getShardId().equals(entry.getKey().getShardId())) {
- // synchronize the local state if the ending sequence number is different
- if (!current.getEndingSequenceNumber().equals(entry.getKey().getEndingSequenceNumber())) {
- // ugly, but since the hashcode will change, we'll need to remove it and add it back
- sequenceNumsToRestore.remove(entry.getKey());
- entry.getKey().setEndingSequenceNumber(current.getEndingSequenceNumber());
- sequenceNumsToRestore.put(entry.getKey(), entry.getValue());
- return true;
- }
- // we already found the matching shard
- break;
- }
- }
- return false;
- }
-
@Override
public void cancel() {
running = false;
@@ -362,7 +316,13 @@ public class FlinkKinesisConsumer<T> extends RichParallelSourceFunction<T> imple
if (sequenceNumsToRestore == null) {
sequenceNumsToRestore = new HashMap<>();
for (Tuple2<StreamShardMetadata, SequenceNumber> kinesisSequenceNumber : sequenceNumsStateForCheckpoint.get()) {
- sequenceNumsToRestore.put(kinesisSequenceNumber.f0, kinesisSequenceNumber.f1);
+ sequenceNumsToRestore.put(
+ // we wrap the restored metadata inside an equivalence wrapper that checks only stream name and shard id,
+ // so that if a shard had been closed (due to a Kinesis reshard operation, for example) since
+ // the savepoint and has a different metadata than what we last stored,
+ // we will still be able to match it in sequenceNumsToRestore. Please see FLINK-8484 for details.
+ new StreamShardMetadata.EquivalenceWrapper(kinesisSequenceNumber.f0),
+ kinesisSequenceNumber.f1);
}
LOG.info("Setting restore state in the FlinkKinesisConsumer. Using the following offsets: {}",
@@ -386,16 +346,16 @@ public class FlinkKinesisConsumer<T> extends RichParallelSourceFunction<T> imple
if (fetcher == null) {
if (sequenceNumsToRestore != null) {
- for (Map.Entry<StreamShardMetadata, SequenceNumber> entry : sequenceNumsToRestore.entrySet()) {
+ for (Map.Entry<StreamShardMetadata.EquivalenceWrapper, SequenceNumber> entry : sequenceNumsToRestore.entrySet()) {
// sequenceNumsToRestore is the restored global union state;
// should only snapshot shards that actually belong to us
if (KinesisDataFetcher.isThisSubtaskShouldSubscribeTo(
- KinesisDataFetcher.convertToStreamShardHandle(entry.getKey()),
+ KinesisDataFetcher.convertToStreamShardHandle(entry.getKey().getShardMetadata()),
getRuntimeContext().getNumberOfParallelSubtasks(),
getRuntimeContext().getIndexOfThisSubtask())) {
- sequenceNumsStateForCheckpoint.add(Tuple2.of(entry.getKey(), entry.getValue()));
+ sequenceNumsStateForCheckpoint.add(Tuple2.of(entry.getKey().getShardMetadata(), entry.getValue()));
}
}
}
@@ -426,7 +386,7 @@ public class FlinkKinesisConsumer<T> extends RichParallelSourceFunction<T> imple
}
@VisibleForTesting
- HashMap<StreamShardMetadata, SequenceNumber> getRestoredState() {
+ HashMap<StreamShardMetadata.EquivalenceWrapper, SequenceNumber> getRestoredState() {
return sequenceNumsToRestore;
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7884a4f7/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/internals/KinesisDataFetcher.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/internals/KinesisDataFetcher.java b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/internals/KinesisDataFetcher.java
index a8f37a5..79e4bfd 100644
--- a/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/internals/KinesisDataFetcher.java
+++ b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/internals/KinesisDataFetcher.java
@@ -574,6 +574,11 @@ public class KinesisDataFetcher<T> {
});
}
+ @VisibleForTesting
+ public List<KinesisStreamShardState> getSubscribedShardsState() {
+ return subscribedShardsState;
+ }
+
/**
* Utility function to create an initial map of the last discovered shard id of each subscribed stream, set to null;
* This is called in the constructor; correct values will be set later on by calling advanceLastDiscoveredShardOfStream().
http://git-wip-us.apache.org/repos/asf/flink/blob/7884a4f7/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/model/StreamShardMetadata.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/model/StreamShardMetadata.java b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/model/StreamShardMetadata.java
index a158a8b..707822a 100644
--- a/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/model/StreamShardMetadata.java
+++ b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/model/StreamShardMetadata.java
@@ -20,6 +20,8 @@ package org.apache.flink.streaming.connectors.kinesis.model;
import java.io.Serializable;
import java.util.Objects;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
/**
* A serializable representation of a AWS Kinesis Stream shard. It is basically a wrapper class around the information
* disintegrated from {@link com.amazonaws.services.kinesis.model.Shard} and its nested classes. The disintegration
@@ -170,4 +172,46 @@ public class StreamShardMetadata implements Serializable {
return hash;
}
+ /** An equivalence wrapper that only checks for the stream name and shard id for equality. */
+ public static class EquivalenceWrapper {
+
+ private final StreamShardMetadata shardMetadata;
+
+ public EquivalenceWrapper(StreamShardMetadata shardMetadata) {
+ this.shardMetadata = checkNotNull(shardMetadata);
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (!(obj instanceof EquivalenceWrapper)) {
+ return false;
+ }
+
+ if (obj == this) {
+ return true;
+ }
+
+ EquivalenceWrapper other = (EquivalenceWrapper) obj;
+
+ return shardMetadata.getStreamName().equals(other.shardMetadata.getStreamName())
+ && shardMetadata.getShardId().equals(other.shardMetadata.getShardId());
+ }
+
+ @Override
+ public int hashCode() {
+ int hash = 17;
+
+ if (shardMetadata.getStreamName() != null) {
+ hash = 37 * hash + shardMetadata.getStreamName().hashCode();
+ }
+ if (shardMetadata.getShardId() != null) {
+ hash = 37 * hash + shardMetadata.getShardId().hashCode();
+ }
+ return hash;
+ }
+
+ public StreamShardMetadata getShardMetadata() {
+ return shardMetadata;
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7884a4f7/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisConsumerMigrationTest.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisConsumerMigrationTest.java b/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisConsumerMigrationTest.java
index 46010d1..eddbab0 100644
--- a/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisConsumerMigrationTest.java
+++ b/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisConsumerMigrationTest.java
@@ -24,6 +24,8 @@ import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.operators.StreamSource;
import org.apache.flink.streaming.connectors.kinesis.config.AWSConfigConstants;
import org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher;
+import org.apache.flink.streaming.connectors.kinesis.model.KinesisStreamShardState;
+import org.apache.flink.streaming.connectors.kinesis.model.SentinelSequenceNumber;
import org.apache.flink.streaming.connectors.kinesis.model.SequenceNumber;
import org.apache.flink.streaming.connectors.kinesis.model.StreamShardHandle;
import org.apache.flink.streaming.connectors.kinesis.model.StreamShardMetadata;
@@ -38,6 +40,7 @@ import org.apache.flink.streaming.util.migration.MigrationTestUtil;
import org.apache.flink.streaming.util.migration.MigrationVersion;
import org.apache.flink.streaming.util.serialization.SimpleStringSchema;
+import com.amazonaws.services.kinesis.model.SequenceNumberRange;
import com.amazonaws.services.kinesis.model.Shard;
import org.junit.Ignore;
@@ -52,10 +55,12 @@ import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicReference;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertTrue;
@@ -76,14 +81,16 @@ public class FlinkKinesisConsumerMigrationTest {
private final MigrationVersion flinkGenerateSavepointVersion = null;
private static final String TEST_STREAM_NAME = "fakeStream1";
+ private static final SequenceNumber TEST_SEQUENCE_NUMBER = new SequenceNumber("987654321");
+ private static final String TEST_SHARD_ID = KinesisShardIdGenerator.generateFromShardOrder(0);
private static final HashMap<StreamShardMetadata, SequenceNumber> TEST_STATE = new HashMap<>();
static {
StreamShardMetadata shardMetadata = new StreamShardMetadata();
shardMetadata.setStreamName(TEST_STREAM_NAME);
- shardMetadata.setShardId(KinesisShardIdGenerator.generateFromShardOrder(0));
+ shardMetadata.setShardId(TEST_SHARD_ID);
- TEST_STATE.put(shardMetadata, new SequenceNumber("987654321"));
+ TEST_STATE.put(shardMetadata, TEST_SEQUENCE_NUMBER);
}
private final MigrationVersion testMigrateVersion;
@@ -116,6 +123,10 @@ public class FlinkKinesisConsumerMigrationTest {
Shard shard = new Shard();
shard.setShardId(shardMetadata.getShardId());
+ SequenceNumberRange sequenceNumberRange = new SequenceNumberRange();
+ sequenceNumberRange.withStartingSequenceNumber("1");
+ shard.setSequenceNumberRange(sequenceNumberRange);
+
initialDiscoveryShards.add(new StreamShardHandle(shardMetadata.getStreamName(), shard));
}
@@ -147,6 +158,15 @@ public class FlinkKinesisConsumerMigrationTest {
// assert that no state was restored
assertTrue(consumerFunction.getRestoredState().isEmpty());
+ // although the restore state is empty, the fetcher should still have been registered the initial discovered shard;
+ // furthermore, the discovered shard should be considered a newly created shard while the job wasn't running,
+ // and therefore should be consumed from the earliest sequence number
+ KinesisStreamShardState restoredShardState = fetcher.getSubscribedShardsState().get(0);
+ assertEquals(TEST_STREAM_NAME, restoredShardState.getStreamShardHandle().getStreamName());
+ assertEquals(TEST_SHARD_ID, restoredShardState.getStreamShardHandle().getShard().getShardId());
+ assertFalse(restoredShardState.getStreamShardHandle().isClosed());
+ assertEquals(SentinelSequenceNumber.SENTINEL_EARLIEST_SEQUENCE_NUM.get(), restoredShardState.getLastProcessedSequenceNum());
+
consumerOperator.close();
consumerOperator.cancel();
}
@@ -158,6 +178,10 @@ public class FlinkKinesisConsumerMigrationTest {
Shard shard = new Shard();
shard.setShardId(shardMetadata.getShardId());
+ SequenceNumberRange sequenceNumberRange = new SequenceNumberRange();
+ sequenceNumberRange.withStartingSequenceNumber("1");
+ shard.setSequenceNumberRange(sequenceNumberRange);
+
initialDiscoveryShards.add(new StreamShardHandle(shardMetadata.getStreamName(), shard));
}
@@ -190,7 +214,111 @@ public class FlinkKinesisConsumerMigrationTest {
// assert that state is correctly restored
assertNotEquals(null, consumerFunction.getRestoredState());
assertEquals(1, consumerFunction.getRestoredState().size());
- assertEquals(TEST_STATE, consumerFunction.getRestoredState());
+ assertEquals(TEST_STATE, removeEquivalenceWrappers(consumerFunction.getRestoredState()));
+ assertEquals(1, fetcher.getSubscribedShardsState().size());
+ assertEquals(TEST_SEQUENCE_NUMBER, fetcher.getSubscribedShardsState().get(0).getLastProcessedSequenceNum());
+
+ KinesisStreamShardState restoredShardState = fetcher.getSubscribedShardsState().get(0);
+ assertEquals(TEST_STREAM_NAME, restoredShardState.getStreamShardHandle().getStreamName());
+ assertEquals(TEST_SHARD_ID, restoredShardState.getStreamShardHandle().getShard().getShardId());
+ assertFalse(restoredShardState.getStreamShardHandle().isClosed());
+ assertEquals(TEST_SEQUENCE_NUMBER, restoredShardState.getLastProcessedSequenceNum());
+
+ consumerOperator.close();
+ consumerOperator.cancel();
+ }
+
+ @Test
+ public void testRestoreWithReshardedStream() throws Exception {
+ final List<StreamShardHandle> initialDiscoveryShards = new ArrayList<>(TEST_STATE.size());
+ for (StreamShardMetadata shardMetadata : TEST_STATE.keySet()) {
+ // setup the closed shard
+ Shard closedShard = new Shard();
+ closedShard.setShardId(shardMetadata.getShardId());
+
+ SequenceNumberRange closedSequenceNumberRange = new SequenceNumberRange();
+ closedSequenceNumberRange.withStartingSequenceNumber("1");
+ closedSequenceNumberRange.withEndingSequenceNumber("1087654321"); // this represents a closed shard
+ closedShard.setSequenceNumberRange(closedSequenceNumberRange);
+
+ initialDiscoveryShards.add(new StreamShardHandle(shardMetadata.getStreamName(), closedShard));
+
+ // setup the new shards
+ Shard newSplitShard1 = new Shard();
+ newSplitShard1.setShardId(KinesisShardIdGenerator.generateFromShardOrder(1));
+
+ SequenceNumberRange newSequenceNumberRange1 = new SequenceNumberRange();
+ newSequenceNumberRange1.withStartingSequenceNumber("1087654322");
+ newSplitShard1.setSequenceNumberRange(newSequenceNumberRange1);
+
+ newSplitShard1.setParentShardId(TEST_SHARD_ID);
+
+ Shard newSplitShard2 = new Shard();
+ newSplitShard2.setShardId(KinesisShardIdGenerator.generateFromShardOrder(2));
+
+ SequenceNumberRange newSequenceNumberRange2 = new SequenceNumberRange();
+ newSequenceNumberRange2.withStartingSequenceNumber("2087654322");
+ newSplitShard2.setSequenceNumberRange(newSequenceNumberRange2);
+
+ newSplitShard2.setParentShardId(TEST_SHARD_ID);
+
+ initialDiscoveryShards.add(new StreamShardHandle(shardMetadata.getStreamName(), newSplitShard1));
+ initialDiscoveryShards.add(new StreamShardHandle(shardMetadata.getStreamName(), newSplitShard2));
+ }
+
+ final TestFetcher<String> fetcher = new TestFetcher<>(
+ Collections.singletonList(TEST_STREAM_NAME),
+ new TestSourceContext<>(),
+ getMockRuntimeContext(1, 0),
+ getStandardProperties(),
+ new KinesisDeserializationSchemaWrapper<>(new SimpleStringSchema()),
+ null,
+ initialDiscoveryShards);
+
+ final DummyFlinkKinesisConsumer<String> consumerFunction = new DummyFlinkKinesisConsumer<>(
+ fetcher, new KinesisDeserializationSchemaWrapper<>(new SimpleStringSchema()));
+
+ StreamSource<String, DummyFlinkKinesisConsumer<String>> consumerOperator =
+ new StreamSource<>(consumerFunction);
+
+ final AbstractStreamOperatorTestHarness<String> testHarness =
+ new AbstractStreamOperatorTestHarness<>(consumerOperator, 1, 1, 0);
+
+ testHarness.setup();
+ MigrationTestUtil.restoreFromSnapshot(
+ testHarness,
+ "src/test/resources/kinesis-consumer-migration-test-flink" + testMigrateVersion + "-snapshot", testMigrateVersion);
+ testHarness.open();
+
+ consumerFunction.run(new TestSourceContext<>());
+
+ // assert that state is correctly restored
+ assertNotEquals(null, consumerFunction.getRestoredState());
+ assertEquals(1, consumerFunction.getRestoredState().size());
+ assertEquals(TEST_STATE, removeEquivalenceWrappers(consumerFunction.getRestoredState()));
+
+ // assert that the fetcher is registered with all shards, including new shards
+ assertEquals(3, fetcher.getSubscribedShardsState().size());
+
+ KinesisStreamShardState restoredClosedShardState = fetcher.getSubscribedShardsState().get(0);
+ assertEquals(TEST_STREAM_NAME, restoredClosedShardState.getStreamShardHandle().getStreamName());
+ assertEquals(TEST_SHARD_ID, restoredClosedShardState.getStreamShardHandle().getShard().getShardId());
+ assertTrue(restoredClosedShardState.getStreamShardHandle().isClosed());
+ assertEquals(TEST_SEQUENCE_NUMBER, restoredClosedShardState.getLastProcessedSequenceNum());
+
+ KinesisStreamShardState restoredNewSplitShard1 = fetcher.getSubscribedShardsState().get(1);
+ assertEquals(TEST_STREAM_NAME, restoredNewSplitShard1.getStreamShardHandle().getStreamName());
+ assertEquals(KinesisShardIdGenerator.generateFromShardOrder(1), restoredNewSplitShard1.getStreamShardHandle().getShard().getShardId());
+ assertFalse(restoredNewSplitShard1.getStreamShardHandle().isClosed());
+ // new shards should be consumed from the beginning
+ assertEquals(SentinelSequenceNumber.SENTINEL_EARLIEST_SEQUENCE_NUM.get(), restoredNewSplitShard1.getLastProcessedSequenceNum());
+
+ KinesisStreamShardState restoredNewSplitShard2 = fetcher.getSubscribedShardsState().get(2);
+ assertEquals(TEST_STREAM_NAME, restoredNewSplitShard2.getStreamShardHandle().getStreamName());
+ assertEquals(KinesisShardIdGenerator.generateFromShardOrder(2), restoredNewSplitShard2.getStreamShardHandle().getShard().getShardId());
+ assertFalse(restoredNewSplitShard2.getStreamShardHandle().isClosed());
+ // new shards should be consumed from the beginning
+ assertEquals(SentinelSequenceNumber.SENTINEL_EARLIEST_SEQUENCE_NUM.get(), restoredNewSplitShard2.getLastProcessedSequenceNum());
consumerOperator.close();
consumerOperator.cancel();
@@ -344,4 +472,15 @@ public class FlinkKinesisConsumerMigrationTest {
return runtimeContext;
}
+
+ private static Map<StreamShardMetadata, SequenceNumber> removeEquivalenceWrappers(
+ Map<StreamShardMetadata.EquivalenceWrapper, SequenceNumber> equivalenceWrappedMap) {
+
+ Map<StreamShardMetadata, SequenceNumber> unwrapped = new HashMap<>();
+ for (Map.Entry<StreamShardMetadata.EquivalenceWrapper, SequenceNumber> wrapped : equivalenceWrappedMap.entrySet()) {
+ unwrapped.put(wrapped.getKey().getShardMetadata(), wrapped.getValue());
+ }
+
+ return unwrapped;
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7884a4f7/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisConsumerTest.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisConsumerTest.java b/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisConsumerTest.java
index a142b26..11b951b 100644
--- a/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisConsumerTest.java
+++ b/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisConsumerTest.java
@@ -63,14 +63,12 @@ import org.powermock.modules.junit4.PowerMockRunner;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
-import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.UUID;
import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.mock;
@@ -597,56 +595,6 @@ public class FlinkKinesisConsumerTest {
closedStreamShardHandle, fakeRestoredState.get(closedStreamShardHandle)));
}
- @Test
- public void testUpdateKinesisShardStateWithMissingEndingSequenceNumber() {
- final String streamName = "fakeStream1";
- final String shardId = "shard-000001";
-
- Properties config = TestUtils.getStandardProperties();
- FlinkKinesisConsumer<String> consumer = new FlinkKinesisConsumer<>(streamName, new SimpleStringSchema(), config);
-
- // having the current Kinesis shard we're trying to find the restored state for
- final StreamShardMetadata current = new StreamShardMetadata();
- current.setShardId(shardId);
- current.setStreamName(streamName);
- current.setEndingSequenceNumber(null);
-
- final HashMap<StreamShardMetadata, SequenceNumber> sequenceNumsToRestore = new LinkedHashMap<>();
- assertFalse("Current shard is open, expecting a short-circuit", consumer.updateKinesisShardStateWithMissingEndingSequenceNumber(current, sequenceNumsToRestore));
-
- // create some non-matching shards
- final StreamShardMetadata differentStreamName = new StreamShardMetadata();
- differentStreamName.setStreamName("fakeStream2");
- differentStreamName.setShardId(shardId);
-
- final StreamShardMetadata differentShardId = new StreamShardMetadata();
- differentShardId.setStreamName(streamName);
- differentShardId.setShardId("shard-000002");
-
- // create the matching shard
- final StreamShardMetadata match = new StreamShardMetadata();
- match.setStreamName(streamName);
- match.setShardId(shardId);
- // ensure the sequence number isn't set (shard is considered in open state)
- match.setEndingSequenceNumber(null);
-
- sequenceNumsToRestore.put(differentStreamName, new SequenceNumber("123"));
- sequenceNumsToRestore.put(differentShardId, new SequenceNumber("456"));
- sequenceNumsToRestore.put(match, new SequenceNumber("789"));
-
- assertFalse("No ending sequence number was set, so no synchronisation was done.", consumer.updateKinesisShardStateWithMissingEndingSequenceNumber(current, sequenceNumsToRestore));
-
- // alter the ending sequence number (indicating the shard is now closed)
- final String endingSequenceNumber = "99999";
- current.setEndingSequenceNumber(endingSequenceNumber);
- assertTrue("Shard was closed (ending number set).", consumer.updateKinesisShardStateWithMissingEndingSequenceNumber(current, sequenceNumsToRestore));
- assertEquals(endingSequenceNumber, match.getEndingSequenceNumber());
- assertEquals(current, match);
- assertTrue("Make sure we can still find back our match in the restored sequences ", sequenceNumsToRestore.containsKey(match));
-
- assertFalse("Ending number was already set, no more need for synchronisation", consumer.updateKinesisShardStateWithMissingEndingSequenceNumber(current, sequenceNumsToRestore));
- }
-
private static final class TestingListState<T> implements ListState<T> {
private final List<T> list = new ArrayList<>();