Github user pluppens commented on a diff in the pull request:
https://github.com/apache/flink/pull/5337#discussion_r163226460
--- Diff:
flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisConsumerTest.java
---
@@ -515,6 +515,56 @@ public void
testStreamShardMetadataSerializedUsingPojoSerializer() {
assertTrue(typeInformation.createSerializer(new
ExecutionConfig()) instanceof PojoSerializer);
}
+ /**
+ * FLINK-8484: ensure that a state change in the StreamShardMetadata
other than {@link StreamShardMetadata#shardId} or
+ * {@link StreamShardMetadata#streamName} does not result in the shard
not being able to be restored.
+ * This handles the corner case where the stored shard metadata is open
(no ending sequence number), but after the
+ * job restore, the shard has been closed (ending number set) due to
re-sharding, and we can no longer rely on
+ * {@link StreamShardMetadata#equals(Object)} to find back the sequence
number in the collection of restored shard metadata.
+ */
+ @Test
+ public void testFindSequenceNumberToRestoreFrom() {
--- End diff --
Makes sense. I'll look into it and see if I can find a way to test it as a
whole.
---
