[GitHub] [flink] dannycranmer commented on a diff in pull request #20945: [FLINK-29395][Connector/Kinesis] Handle empty deaggregated records…

[GitBox <gi...@apache.org>]

dannycranmer commented on code in PR #20945:
URL: https://github.com/apache/flink/pull/20945#discussion_r991006302


##########
flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/internals/publisher/fanout/FanOutRecordPublisherTest.java:
##########
@@ -167,6 +173,42 @@ public void testToSdkV2StartingPositionAtTimeStamp() throws Exception {
         assertThat(kinesis.getStartingPositionForSubscription(0).type()).isEqualTo(AT_TIMESTAMP);
     }
 
+    @Test
+    public void testToSdkV2StartingPositionAtTimestampWithDroppedAggregatedRecord()
+            throws Exception {
+        // Create Aggregate Record with explicit hash keys 0 and 1
+        RecordAggregator recordAggregator = new RecordAggregator();
+        recordAggregator.addUserRecord("pk", "0", randomAlphabetic(32).getBytes(UTF_8));
+        recordAggregator.addUserRecord("pk", "1", randomAlphabetic(32).getBytes(UTF_8));
+        Record record =
+                Record.builder()
+                        .approximateArrivalTimestamp(Instant.now())
+                        .data(
+                                SdkBytes.fromByteArray(
+                                        recordAggregator.clearAndGet().toRecordBytes()))
+                        .sequenceNumber("1")
+                        .partitionKey("pk")
+                        .build();
+
+        KinesisProxyV2Interface kinesis = singletonShard(createSubscribeToShardEvent(record));
+        Date now = new Date();
+
+        // Create ShardHandle with HashKeyRange excluding single UserRecord with hash key 0
+        HashKeyRange hashKeyRange =
+                new HashKeyRange().withStartingHashKey("1").withEndingHashKey("100");
+
+        RecordPublisher publisher =
+                new FanOutRecordPublisher(
+                        StartingPosition.fromTimestamp(now),
+                        "arn",
+                        createDummyStreamShardHandle("stream-name", "shardId-00000", hashKeyRange),
+                        kinesis,
+                        createConfiguration(),
+                        new FullJitterBackoff());
+        publisher.run(
+                new TestConsumer(SENTINEL_AT_TIMESTAMP_SEQUENCE_NUM.get().getSequenceNumber()));

Review Comment:
   Can we assert that the correct sequence number is actually reinstated somehow?



-- 
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