[FLINK-32097][Connectors/Kinesis] Implement support for Kinesis deaggregation #188

Lzgpom · 2025-02-01T15:35:37Z

Purpose of the change

This PR implements the Kinesis deaggregation. The implementation was heavily based on the old kinesis connector.

Verifying this change

This change added tests and can be verified as follows:

Added integration tests for with record deaggregation
Added unit tests
Manually verified by running the Kinesis connector on a local Flink cluster.

Further ToDos and Follow-ups

Checkpoints do not take into consideration the sequence number of the deaggregated records.

Brief change log

767f576 Implement record deaggregation using AggregatorUtil from the KCL 3.x
ddcb1a9 Implemented tests for RecordBatch regarding deaggregation. There was a need for aggregated records to test. So it was used https://github.com/awslabs/kinesis-aggregation to do so. Unfortunatly, it does have a version in the maven repository compatible with KCL 3.x so it was used a workaround.
fce6d03 Added integration tests.
ddcb1a9 fce6d03 Added unit tests.

Significant changes

Dependencies have been added or upgraded
Public API has been changed (Public API is any class annotated with @Public(Evolving))
Serializers have been changed
New feature has been introduced
- If yes, how is this documented? (not applicable / docs / JavaDocs / not documented)

In order to implement record deaggregation, the RecordBatch uses the AggregatorUtil, from the KCL 3, with the subscribed shard to deaggregate. The deaggregated records are now instance of KinesisClientRecord since the AggregatorUtil requires it. Therefore, there is a bit of refactor to update the class of the record that was received. Also, the RecordBatch class was extracted out of KinesisShardSplitReaderBase

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

…gations.

… for aggregation tests.

…plitReader for aggregated records.

…m stream to loop.

boring-cyborg · 2025-02-01T15:35:40Z

Thanks for opening this pull request! Please check out our contributing guidelines. (https://flink.apache.org/contributing/how-to-contribute.html)

Lzgpom · 2025-02-05T13:28:39Z

@hlteoh37 could you take a look?

nicusX · 2025-02-07T15:12:27Z

@Lzgpom Building this locally and I've got a spotless-check format violation.
mvn spotless:apply fixes it

nicusX · 2025-02-07T17:12:34Z

I tested it locally, with a stream produced by KPL with aggregation enabled, and it worked fine.

Lzgpom · 2025-02-07T17:26:07Z

@Lzgpom Building this locally and I've got a spotless-check format violation. mvn spotless:apply fixes it

I tried to run mvn spotless:apply but it didn't have any changes. It allows reported that Spotless apply skipped

nicusX · 2025-02-07T17:33:59Z

Uhm, spotless:apply it did some changes for me, and after that mvn package succeeded

nicusX · 2025-02-07T17:43:27Z

If I run a simple mvn clean package on your branch I get spotess-check violations in RecordBatchTest.java + 10 other files. And mvn spotless:apply changes 12 files.
All style definitions are part of the repo.
Not sure I understand the different result

Lzgpom · 2025-02-07T19:01:25Z

@nicusX I was using the java 17 and the spotless did not work. I changed to java 11 and it worked.
Thanks for notifying me about this.

nicusX

Changes LGTM
I verified that it now does not fail checkstyle checks.

Note for maintainers: There is a breaking change in the public API that forces a major version change.

nicusX · 2025-02-18T09:44:23Z

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

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


This has Apache 2.0 licence, all good so
https://github.com/awslabs/kinesis-aggregation

nicusX · 2025-02-18T09:45:32Z

...va/org/apache/flink/connector/kinesis/source/serialization/KinesisDeserializationSchema.java

@@ -60,7 +60,7 @@ default void open(DeserializationSchema.InitializationContext context) throws Ex
     * @param output the identifier of the shard the record was sent to
     * @throws IOException exception when deserializing record
     */
-    void deserialize(Record record, String stream, String shardId, Collector<T> output)
+    void deserialize(KinesisClientRecord record, String stream, String shardId, Collector<T> output)


This is a breaking change to the public interface

+1

@Lzgpom Have we considered alternatives to changing the interface? I can see that Hong suggested wrapping internally. Have we considered alternatives such as having a separate KinesisClientRecordDeserializationSchema?

hlteoh37 · 2025-03-03T08:51:25Z

.../main/java/org/apache/flink/connector/kinesis/source/reader/KinesisShardSplitReaderBase.java

@@ -50,10 +49,10 @@
 /** Base implementation of the SplitReader for reading from KinesisShardSplits. */
 @Internal
 public abstract class KinesisShardSplitReaderBase
-        implements SplitReader<Record, KinesisShardSplit> {
+        implements SplitReader<KinesisClientRecord, KinesisShardSplit> {


This change actually changes the interface exposed for the KinesisSource, so it would be a backwards incompatible change. Is there a way we can wrap this internally?

I have been looking at the implications of not changing from Record into KinesisClientRecord and this is what I have found:

The deaggreated records would have to be converted from KinesisClientRecord, since AggregatorUtil only uses KinesisClientRecord, to Record adding a slight overhead.

The KinesisClientRecord has the subSequenceNumber that could and should be used for checkpointing.

Apart from that I don't see any more issues. Either way I don't mind changing it back to Record.

@Lzgpom Can you elaborate on what the user experience would be with the current changes in PR? Will it break compilation for existing apps that already depend on KinesisSource without further code changes?

@Lzgpom I think the gist is, can you please change the interface back to using Record in all @Public and @PublicEvolving public methods?
This would allow including the change in a minor version of the connector, and users of 5.0.0 would be able to just update the connector version with no code changes

leekeiabstraction · 2025-04-14T14:17:39Z

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

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


Would it be better to work with the owner of the repo to get this published instead?

leekeiabstraction · 2025-04-14T14:23:35Z

.../main/java/org/apache/flink/connector/kinesis/source/reader/KinesisShardSplitReaderBase.java

@@ -50,10 +49,10 @@
 /** Base implementation of the SplitReader for reading from KinesisShardSplits. */
 @Internal
 public abstract class KinesisShardSplitReaderBase
-        implements SplitReader<Record, KinesisShardSplit> {
+        implements SplitReader<KinesisClientRecord, KinesisShardSplit> {


@Lzgpom Can you elaborate on what the user experience would be with the current changes in PR? Will it break compilation for existing apps that already depend on KinesisSource without further code changes?

leekeiabstraction · 2025-04-14T14:28:22Z

...va/org/apache/flink/connector/kinesis/source/serialization/KinesisDeserializationSchema.java

@@ -60,7 +60,7 @@ default void open(DeserializationSchema.InitializationContext context) throws Ex
     * @param output the identifier of the shard the record was sent to
     * @throws IOException exception when deserializing record
     */
-    void deserialize(Record record, String stream, String shardId, Collector<T> output)
+    void deserialize(KinesisClientRecord record, String stream, String shardId, Collector<T> output)


+1

@Lzgpom Have we considered alternatives to changing the interface? I can see that Hong suggested wrapping internally. Have we considered alternatives such as having a separate KinesisClientRecordDeserializationSchema?

dannycranmer · 2025-04-29T07:31:25Z

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

@@ -52,6 +60,11 @@ under the License.
            <artifactId>kinesis</artifactId>
        </dependency>

+        <dependency>
+            <groupId>software.amazon.kinesis</groupId>
+            <artifactId>amazon-kinesis-client</artifactId>


This is a shame, can we do this without KCL? It's a heavy weight dependency for the 1 class(?) needed for deaggregation. We have been talking about removing this dependency with the new connector for years, but never really solved this bit. Can we ask Kinesis to publish a separate library for this? Or, we could just copy that file (assuming it is still a small asking of code)

@dannycranmer KinesisClientRecord has some transitive dependencies.
However, the KCL can be substantially lightened removing most of transitive dependencies
See this example which is an alternative solution, implementing de-aggregation in the Deserialization Schema

Lzgpom added 6 commits February 1, 2025 14:57

[FLINK-32097][Connectors/Kinesis] Added integration tests for deaggre…

fce6d03

…gations.

[FLINK-32097][Connectors/Kinesis] Improved KinesisStreamsSourceITCase…

82c6ed2

… for aggregation tests.

[FLINK-32097][Connectors/Kinesis] Added tests to PollingKinesisShardS…

e1e274f

…plitReader for aggregated records.

[FLINK-32097][Connectors/Kinesis] Changed RecordBatch deaggregate fro…

156dc6b

…m stream to loop.

[FLINK-32097][Connectors/Kinesis] Spotless apply

ae629b4

nicusX reviewed Feb 18, 2025

View reviewed changes

hlteoh37 reviewed Mar 3, 2025

View reviewed changes

leekeiabstraction reviewed Apr 14, 2025

View reviewed changes

dannycranmer reviewed Apr 29, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-32097][Connectors/Kinesis] Implement support for Kinesis deaggregation #188

[FLINK-32097][Connectors/Kinesis] Implement support for Kinesis deaggregation #188

Lzgpom commented Feb 1, 2025

boring-cyborg bot commented Feb 1, 2025

Lzgpom commented Feb 5, 2025

nicusX commented Feb 7, 2025

nicusX commented Feb 7, 2025

Lzgpom commented Feb 7, 2025

nicusX commented Feb 7, 2025

nicusX commented Feb 7, 2025

Lzgpom commented Feb 7, 2025

nicusX left a comment •

edited

Loading

nicusX Feb 18, 2025

nicusX Feb 18, 2025

leekeiabstraction Apr 14, 2025

hlteoh37 Mar 3, 2025

Lzgpom Mar 5, 2025

leekeiabstraction Apr 14, 2025

nicusX Apr 24, 2025

leekeiabstraction Apr 14, 2025

leekeiabstraction Apr 14, 2025

leekeiabstraction Apr 14, 2025

dannycranmer Apr 29, 2025

nicusX Apr 29, 2025

[FLINK-32097][Connectors/Kinesis] Implement support for Kinesis deaggregation #188

Are you sure you want to change the base?

[FLINK-32097][Connectors/Kinesis] Implement support for Kinesis deaggregation #188

Conversation

Lzgpom commented Feb 1, 2025

Purpose of the change

Verifying this change

Further ToDos and Follow-ups

Brief change log

Significant changes

boring-cyborg bot commented Feb 1, 2025

Lzgpom commented Feb 5, 2025

nicusX commented Feb 7, 2025

nicusX commented Feb 7, 2025

Lzgpom commented Feb 7, 2025

nicusX commented Feb 7, 2025

nicusX commented Feb 7, 2025

Lzgpom commented Feb 7, 2025

nicusX left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nicusX left a comment •

edited

Loading