[FLINK-32097][Connectors/Kinesis] Added tests for RecordBatch regarding deaggregation.

There was a need to create aggregated records to test if the records are being deaggregated correctly.
For that there is a dependency that can create aggregated records, but unfortunately it does not have a version available in the maven repository that is compatible with the kcl 3.x. So it uses a release of github that is compatible.

Also, the protobuf version set in the dependency management was not compatible with the kcl 3.x

Author: Lzgpom

flink-connector-aws/flink-connector-aws-kinesis-streams/pom.xml

@@ -33,6 +33,14 @@ under the License.
     <name>Flink : Connectors : AWS : Amazon Kinesis Data Streams Connector v2</name>
     <packaging>jar</packaging>
 
+    <repositories>
+        <!-- used for the kinesis aggregator dependency since it is not available in maven central -->
+        <repository>
+            <id>jitpack.io</id>
+            <url>https://jitpack.io</url>
+        </repository>
+    </repositories>
+
     <dependencies>
         <dependency>
             <groupId>org.apache.flink</groupId>
@@ -107,6 +115,15 @@ under the License.
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <!-- the kinesis aggregator dependency since it is not available in maven central -->
+            <!-- look into issue https://github.com/awslabs/kinesis-aggregation/issues/120 -->
+            <groupId>com.github.awslabs.kinesis-aggregation</groupId>
+            <artifactId>amazon-kinesis-aggregator</artifactId>
+            <version>2.0.3</version>
+            <scope>test</scope>
+        </dependency>
+
         <dependency>
             <groupId>nl.jqno.equalsverifier</groupId>
             <artifactId>equalsverifier</artifactId>

flink-connector-aws/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/source/reader/RecordBatchTest.java

@@ -0,0 +1,101 @@
+/*
+ * 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.connector.kinesis.source.reader;
+
+import com.amazonaws.kinesis.agg.AggRecord;
+import com.amazonaws.kinesis.agg.RecordAggregator;
+import org.junit.jupiter.api.Test;
+import software.amazon.awssdk.core.SdkBytes;
+import software.amazon.awssdk.services.kinesis.model.Record;
+
+import java.time.Instant;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.apache.flink.connector.kinesis.source.util.TestUtil.getTestRecord;
+import static org.apache.flink.connector.kinesis.source.util.TestUtil.getTestSplit;
+import static org.assertj.core.api.Assertions.assertThat;
+
+class RecordBatchTest {
+
+    @Test
+    public void testDeaggregateRecordsPassThrough() {
+        List<Record> records =
+                Stream.of(getTestRecord("data-1"), getTestRecord("data-2"), getTestRecord("data-3"))
+                        .collect(Collectors.toList());
+
+        RecordBatch result =
+                new RecordBatch(records, getTestSplit(), 100L, true);
+
+        assertThat(result.getDeaggregatedRecords().size()).isEqualTo(3);
+    }
+
+    @Test
+    public void testDeaggregateRecordsWithAggregatedRecords() {
+        List<Record> records =
+                Stream.of(getTestRecord("data-1"), getTestRecord("data-2"), getTestRecord("data-3"))
+                        .collect(Collectors.toList());
+
+        Record aggregatedRecord = createAggregatedRecord(records);
+
+        RecordBatch result =
+                new RecordBatch(Collections.singletonList(aggregatedRecord), getTestSplit(), 100L, true);
+
+        assertThat(result.getDeaggregatedRecords().size()).isEqualTo(3);
+    }
+
+    @Test
+    public void testGetMillisBehindLatest() {
+        RecordBatch result =
+                new RecordBatch(
+                        Collections.singletonList(getTestRecord("data-1")), getTestSplit(), 100L, true);
+
+        assertThat(result.getMillisBehindLatest()).isEqualTo(100L);
+    }
+
+    @Test
+    public void testIsCompleted() {
+        RecordBatch result =
+                new RecordBatch(
+                        Collections.singletonList(getTestRecord("data-1")), getTestSplit(), 100L, true);
+
+        assertThat(result.isCompleted()).isTrue();
+    }
+
+    private static Record createAggregatedRecord(List<Record> records) {
+        RecordAggregator recordAggregator = new RecordAggregator();
+
+        for (Record record : records) {
+            try {
+                recordAggregator.addUserRecord("key", record.data().asByteArray());
+            } catch (Exception e) {
+                throw new RuntimeException("Failed to add record to aggregator", e);
+            }
+        }
+
+        AggRecord aggRecord = recordAggregator.clearAndGet();
+
+        return Record.builder()
+                .data(SdkBytes.fromByteArray(aggRecord.toRecordBytes()))
+                .approximateArrivalTimestamp(Instant.now())
+                .build();
+    }
+}

pom.xml

@@ -332,11 +332,6 @@ under the License.
                 <artifactId>javassist</artifactId>
                 <version>3.24.0-GA</version>
             </dependency>
-            <dependency>
-                <groupId>com.google.protobuf</groupId>
-                <artifactId>protobuf-java</artifactId>
-                <version>3.25.5</version>
-            </dependency>
             <dependency>
                 <groupId>com.google.guava</groupId>
                 <artifactId>guava</artifactId>