Re: [PR] [FLINK-37627][BugFix][Connectors/Kinesis] Restarting from a checkpoint/savepoint which coincides with shard split causes data loss [flink-connector-aws]

["leekeiabstraction (via GitHub)" <gi...@apache.org>]

leekeiabstraction commented on code in PR #198:
URL: https://github.com/apache/flink-connector-aws/pull/198#discussion_r2093275543


##########
flink-connector-aws/flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/source/reader/KinesisStreamsSourceReader.java:
##########
@@ -49,6 +54,8 @@ public class KinesisStreamsSourceReader<T>
 
     private static final Logger LOG = LoggerFactory.getLogger(KinesisStreamsSourceReader.class);
     private final Map<String, KinesisShardMetrics> shardMetricGroupMap;
+    private final NavigableMap<Long, Set<KinesisShardSplit>> splitFinishedEvents;

Review Comment:
   Instead of `splitFinishedEvents`, can we call this `finishedSplits` as these are just KinesisShardSplit object with finished set to true? This is so that it's consistent with the class name `KinesisShardSplit`. In SourceReaderBase, these are referred to as splits.
   
   SplitFinishedEvent is an entirely different pojo.



##########
flink-connector-aws/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/source/split/KinesisShardSplitSerializerTest.java:
##########
@@ -117,6 +117,54 @@ void testDeserializeWrongVersion() throws Exception {
                 .withMessageContaining(String.valueOf(wrongVersionSerializer.getVersion()));
     }
 
+    @Test
+    void testSerializeAndDeserializeWithFinishedSplits() throws Exception {
+        final KinesisShardSplit initialSplit =
+                new KinesisShardSplit(
+                        STREAM_ARN,
+                        generateShardId(10),
+                        StartingPosition.continueFromSequenceNumber("some-sequence-number"),
+                        new HashSet<>(Arrays.asList(generateShardId(2), generateShardId(5))),
+                        STARTING_HASH_KEY_TEST_VALUE,
+                        ENDING_HASH_KEY_TEST_VALUE,
+                        true);
+
+        KinesisShardSplitSerializer serializer = new KinesisShardSplitSerializer();
+
+        byte[] serialized = serializer.serialize(initialSplit);
+        KinesisShardSplit deserializedSplit =
+                serializer.deserialize(serializer.getVersion(), serialized);
+
+        assertThat(deserializedSplit).usingRecursiveComparison().isEqualTo(initialSplit);
+        assertThat(deserializedSplit.isFinished()).isTrue();
+    }
+
+    @Test
+    void testDeserializeVersion1() throws Exception {
+        final KinesisShardSplitSerializer serializer = new KinesisShardSplitSerializer();
+
+        final KinesisShardSplit initialSplit =
+                new KinesisShardSplit(
+                        STREAM_ARN,
+                        generateShardId(10),
+                        StartingPosition.continueFromSequenceNumber("some-sequence-number"),
+                        new HashSet<>(Arrays.asList(generateShardId(2), generateShardId(5))),
+                        STARTING_HASH_KEY_TEST_VALUE,
+                        ENDING_HASH_KEY_TEST_VALUE,
+                        true);

Review Comment:
   Let's use the constructor without the `finished` argument.



##########
flink-connector-aws/flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/source/reader/KinesisStreamsSourceReader.java:
##########
@@ -79,8 +138,17 @@ protected KinesisShardSplit toSplitType(String splitId, KinesisShardSplitState s
 
     @Override
     public void addSplits(List<KinesisShardSplit> splits) {
-        splits.forEach(this::registerShardMetricGroup);
-        super.addSplits(splits);
+        List<KinesisShardSplit> kinesisShardSplits = new ArrayList<>();

Review Comment:
   nit: We can name this `unfinishedSplits`



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org