You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Lzgpom (via GitHub)" <gi...@apache.org> on 2025/02/01 15:35:39 UTC
[PR] [FLINK-32097][Connectors/Kinesis] Implement support for Kinesis deaggregation [flink-connector-aws]
Lzgpom opened a new pull request, #188:
URL: https://github.com/apache/flink-connector-aws/pull/188
## 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
- 767f576d4295aa6c0ad04ac34ebb183cc57bd0cd Implement record deaggregation using `AggregatorUtil` from the KCL 3.x
- ddcb1a9cf291cd9eab73829925a08da76b2dc107 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.
- fce6d035270509f31fd4cf3d9a13b0737cb985a1 Added integration tests.
- ddcb1a9cf291cd9eab73829925a08da76b2dc107 fce6d035270509f31fd4cf3d9a13b0737cb985a1 Added unit tests.
## Significant changes
- [x] 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)
--
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
Re: [PR] [FLINK-32097][Connectors/Kinesis] Implement support for Kinesis deaggregation [flink-connector-aws]
Posted by "nicusX (via GitHub)" <gi...@apache.org>.
nicusX commented on PR #188:
URL: https://github.com/apache/flink-connector-aws/pull/188#issuecomment-2643519339
I tested it locally, with a stream produced by KPL with aggregation enabled, and it worked fine.
--
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
Re: [PR] [FLINK-32097][Connectors/Kinesis] Implement support for Kinesis deaggregation [flink-connector-aws]
Posted by "Lzgpom (via GitHub)" <gi...@apache.org>.
Lzgpom commented on PR #188:
URL: https://github.com/apache/flink-connector-aws/pull/188#issuecomment-2643549489
> @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`
--
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
Re: [PR] [FLINK-32097][Connectors/Kinesis] Implement support for Kinesis deaggregation [flink-connector-aws]
Posted by "nicusX (via GitHub)" <gi...@apache.org>.
nicusX commented on code in PR #188:
URL: https://github.com/apache/flink-connector-aws/pull/188#discussion_r2066006151
##########
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>
Review Comment:
@dannycranmer `KinesisClientRecord` has some transitive dependencies.
However, the KCL can be substantially lightened removing most of transitive dependencies
See [this example](https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/blob/5e4873883c64c28484261125b3a2e20df3455c84/java/KinesisSourceDeaggregation/flink-app/pom.xml#L85) which is an alternative solution, implementing de-aggregation in the Deserialization Schema
--
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
Re: [PR] [FLINK-32097][Connectors/Kinesis] Implement support for Kinesis deaggregation [flink-connector-aws]
Posted by "dannycranmer (via GitHub)" <gi...@apache.org>.
dannycranmer commented on code in PR #188:
URL: https://github.com/apache/flink-connector-aws/pull/188#discussion_r2065697710
##########
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>
Review Comment:
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)
--
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
Re: [PR] [FLINK-32097][Connectors/Kinesis] Implement support for Kinesis deaggregation [flink-connector-aws]
Posted by "Lzgpom (via GitHub)" <gi...@apache.org>.
Lzgpom commented on PR #188:
URL: https://github.com/apache/flink-connector-aws/pull/188#issuecomment-2643757305
@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.
--
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
Re: [PR] [FLINK-32097][Connectors/Kinesis] Implement support for Kinesis deaggregation [flink-connector-aws]
Posted by "Lzgpom (via GitHub)" <gi...@apache.org>.
Lzgpom commented on code in PR #188:
URL: https://github.com/apache/flink-connector-aws/pull/188#discussion_r2093202714
##########
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>
Review Comment:
@leekeiabstraction It seems that there is a problem publishing the artifact as it is stated in this issue https://github.com/awslabs/kinesis-aggregation/issues/120
--
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
Re: [PR] [FLINK-32097][Connectors/Kinesis] Implement support for Kinesis deaggregation [flink-connector-aws]
Posted by "nicusX (via GitHub)" <gi...@apache.org>.
nicusX commented on code in PR #188:
URL: https://github.com/apache/flink-connector-aws/pull/188#discussion_r2058511129
##########
flink-connector-aws/flink-connector-aws-kinesis-streams/src/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> {
Review Comment:
@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
--
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
Re: [PR] [FLINK-32097][Connectors/Kinesis] Implement support for Kinesis deaggregation [flink-connector-aws]
Posted by "nicusX (via GitHub)" <gi...@apache.org>.
nicusX commented on PR #188:
URL: https://github.com/apache/flink-connector-aws/pull/188#issuecomment-2643565743
Uhm, `spotless:apply` it did some changes for me, and after that `mvn package` succeeded
--
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
Re: [PR] [FLINK-32097][Connectors/Kinesis] Implement support for Kinesis deaggregation [flink-connector-aws]
Posted by "hlteoh37 (via GitHub)" <gi...@apache.org>.
hlteoh37 commented on code in PR #188:
URL: https://github.com/apache/flink-connector-aws/pull/188#discussion_r1977101009
##########
flink-connector-aws/flink-connector-aws-kinesis-streams/src/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> {
Review Comment:
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?
--
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
Re: [PR] [FLINK-32097][Connectors/Kinesis] Implement support for Kinesis deaggregation [flink-connector-aws]
Posted by "nicusX (via GitHub)" <gi...@apache.org>.
nicusX commented on PR #188:
URL: https://github.com/apache/flink-connector-aws/pull/188#issuecomment-2643211998
@Lzgpom Building this locally and I've got a spotless-check format violation.
`mvn spotless:apply` fixes it
--
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
Re: [PR] [FLINK-32097][Connectors/Kinesis] Implement support for Kinesis deaggregation [flink-connector-aws]
Posted by "nicusX (via GitHub)" <gi...@apache.org>.
nicusX commented on PR #188:
URL: https://github.com/apache/flink-connector-aws/pull/188#issuecomment-2643588383
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
--
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
Re: [PR] [FLINK-32097][Connectors/Kinesis] Implement support for Kinesis deaggregation [flink-connector-aws]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #188:
URL: https://github.com/apache/flink-connector-aws/pull/188#issuecomment-2628998219
Thanks for opening this pull request! Please check out our contributing guidelines. (https://flink.apache.org/contributing/how-to-contribute.html)
--
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
Re: [PR] [FLINK-32097][Connectors/Kinesis] Implement support for Kinesis deaggregation [flink-connector-aws]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction commented on code in PR #188:
URL: https://github.com/apache/flink-connector-aws/pull/188#discussion_r2042250368
##########
flink-connector-aws/flink-connector-aws-kinesis-streams/src/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> {
Review Comment:
@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?
##########
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>
Review Comment:
Would it be better to work with the owner of the repo to get this published instead?
##########
flink-connector-aws/flink-connector-aws-kinesis-streams/src/main/java/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)
Review Comment:
+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?
--
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
Re: [PR] [FLINK-32097][Connectors/Kinesis] Implement support for Kinesis deaggregation [flink-connector-aws]
Posted by "Lzgpom (via GitHub)" <gi...@apache.org>.
Lzgpom commented on code in PR #188:
URL: https://github.com/apache/flink-connector-aws/pull/188#discussion_r1981100443
##########
flink-connector-aws/flink-connector-aws-kinesis-streams/src/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> {
Review Comment:
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`.
--
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
Re: [PR] [FLINK-32097][Connectors/Kinesis] Implement support for Kinesis deaggregation [flink-connector-aws]
Posted by "nicusX (via GitHub)" <gi...@apache.org>.
nicusX commented on code in PR #188:
URL: https://github.com/apache/flink-connector-aws/pull/188#discussion_r1959395783
##########
flink-connector-aws/flink-connector-aws-kinesis-streams/pom.xml:
##########
@@ -102,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>
Review Comment:
This has Apache 2.0 licence, all good so
https://github.com/awslabs/kinesis-aggregation
##########
flink-connector-aws/flink-connector-aws-kinesis-streams/src/main/java/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)
Review Comment:
This is a breaking change to the public interface
--
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
Re: [PR] [FLINK-32097][Connectors/Kinesis] Implement support for Kinesis deaggregation [flink-connector-aws]
Posted by "Lzgpom (via GitHub)" <gi...@apache.org>.
Lzgpom commented on PR #188:
URL: https://github.com/apache/flink-connector-aws/pull/188#issuecomment-2636857333
@hlteoh37 could you take a look?
--
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