[FLINK-24228][connectors/firehose] - Unified Async Sink for Kinesis Firehose #18314

CrynetLogistics · 2022-01-10T10:58:47Z

What is the purpose of the change

Allows users to write to KDF directly through this sink based on the unified sink (FLIP-141).

Brief change log

(for example:)

Kinesis Data Firehose datastream sink
Minor refactor of KDS sink components to reuse them in KDF sink

Verifying this change

Please make sure both new and modified tests in this PR follows the conventions defined in our code quality guide: https://flink.apache.org/contributing/code-style-and-quality-common.html#testing

(Please pick either of the following options)

This change added tests and can be verified as follows:

Added integration tests for end-to-end inserts into Firehose
Unit tests
Will separately have a Jira/PR for e2e tests hitting live Firehose

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (yes / no)y
The public API, i.e., is any changed class annotated with @Public(Evolving): (yes / no)y
The serializers: (yes / no / don't know)n
The runtime per-record code paths (performance sensitive): (yes / no / don't know)n
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / no / don't know)n
The S3 file system connector: (yes / no / don't know)n

Documentation

Does this pull request introduce a new feature? (yes / no)y
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)docs&JavaDocs

flinkbot · 2022-01-10T11:02:23Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit 0cb7e0c (Mon Jan 10 11:02:23 UTC 2022)

Warnings:

4 pom.xml files were touched: Check for build and licensing issues.
No documentation files were touched! Remember to keep the Flink docs up to date!

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

flinkbot · 2022-01-10T11:02:47Z

CI report:

826fd84 UNKNOWN
cabbe4d Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

...in/java/org/apache/flink/connector/kinesis/sink/KinesisDataFirehoseSinkElementConverter.java

...se/src/test/java/org/apache/flink/connector/firehose/sink/testutils/LocalstackContainer.java

CrynetLogistics · 2022-01-18T10:26:23Z

Documentation for this work: https://issues.apache.org/jira/browse/FLINK-25692

flink-connectors/flink-connector-aws-base/pom.xml

...base/src/main/java/org/apache/flink/connector/aws/config/AWSUnifiedSinksConfigConstants.java

flink-connectors/flink-connector-aws-kinesis-data-firehose/pom.xml

...se/src/main/java/org/apache/flink/connector/firehose/sink/KinesisDataFirehoseSinkWriter.java

...se/src/test/java/org/apache/flink/connector/firehose/sink/KinesisDataFirehoseSinkITCase.java

...st/java/org/apache/flink/connector/firehose/sink/testutils/KinesisDataFirehoseTestUtils.java

...se/src/test/java/org/apache/flink/connector/firehose/sink/testutils/LocalstackContainer.java

CrynetLogistics · 2022-01-20T16:20:20Z

@CrynetLogistics remember to change the default maxInFlightRequests to 50 and maxBufferedRequests to something sensible.

CrynetLogistics · 2022-01-20T16:21:39Z

docs/content.zh/docs/connectors/datastream/firehose.md

...irehose/src/main/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseException.java

dannycranmer · 2022-01-21T14:03:59Z

...rehose/src/test/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkITCase.java

+        env.execute("Integration Test");
+
+        List<S3Object> objects = listBucketObjects(s3AsyncClient, BUCKET_NAME);
+        assertEquals(NUMBER_OF_ELEMENTS, objects.size());


Please update this assertion to use assertJ

dannycranmer · 2022-01-21T14:04:25Z

...se/src/test/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkWriterTest.java

+        assertEquals(
+                testString.getBytes(StandardCharsets.US_ASCII).length,
+                sinkWriter.getSizeInBytes(record));


Change to assertJ and other instances

dannycranmer · 2022-01-21T14:05:24Z

...rehose/src/test/java/org/apache/flink/connector/firehose/sink/examples/SinkIntoFirehose.java

+                env.fromSequence(1, 10_000_000L)
+                        .map(Object::toString)
+                        .returns(String.class)
+                        .map(data -> mapper.writeValueAsString(ImmutableMap.of("data", data)));


nit: Why not use a Json serialisation schema instead of map to string?

Sorry, really struggling to get it working... tried casting to RowData and using a JsonRowDataSerializationSchema but getting issues. I can implement my own SerializationSchema, but I feel that might distract from the main point... any help would be much appreciated...

Is it that I need to have
env.fromSequence(...).<somehow cast to RowData>; and then pass in a JsonRowDataSerializationSchema to the sink?

As discussed, please remove this. We should not be maintaining sample code in the test package of a module. Let's raise a follow up to remove the KDS sample too

...se/src/test/java/org/apache/flink/connector/firehose/sink/testutils/LocalstackContainer.java

...-connector-aws-base/src/main/java/org/apache/flink/connector/aws/util/AWSAsyncSinksUtil.java

...ehose/src/main/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkBuilder.java

.../main/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkElementConverter.java

...rehose/src/main/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkWriter.java

dannycranmer · 2022-01-27T16:57:12Z

...rehose/src/test/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkITCase.java

+    private S3AsyncClient s3AsyncClient;
+    private FirehoseAsyncClient firehoseAsyncClient;
+    private IamAsyncClient iamAsyncClient;


Move these below the static fields

...rehose/src/test/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkITCase.java

…based on the Async Sink Base implemented: - Refactored AWSUnifiedSinksUtil into a class that caters for Kinesis and Firehose - Extracted commonalities between KDS & KDF sinks into flink-connector-aws-base - Implemented integration test based on Localstack container - Changing host/container ports to be different, changing HTTP1.1 to being the default, localstack issue fixed - Added docs page, changed type in Firehose, turned logging off, removed unused dependencies.

…r the new Firehose sink.

…lization schema rather than an ElementConverter, thereby encapsulating the Firehose `Record` from the user, verifying stream objects in KinesisFirehoseITCase

dannycranmer · 2022-02-01T15:35:10Z

Thanks @CrynetLogistics, LGTM, merging

rmetzger added the component=Connectors/Common label Jan 10, 2022

CrynetLogistics force-pushed the FLINK-24228 branch 3 times, most recently from 848e61a to 1bd1a6c Compare January 10, 2022 17:34

CrynetLogistics changed the title ~~[connectors/firehose] FLINK-24228 - Unified Async Sink for Kinesis Firehose~~ [FLINK-24228][connectors/firehose] - Unified Async Sink for Kinesis Firehose Jan 11, 2022

CrynetLogistics force-pushed the FLINK-24228 branch from 1bd1a6c to f803a8f Compare January 11, 2022 14:15

vahmed-hamdy reviewed Jan 12, 2022

View reviewed changes

...in/java/org/apache/flink/connector/kinesis/sink/KinesisDataFirehoseSinkElementConverter.java Outdated Show resolved Hide resolved

CrynetLogistics commented Jan 14, 2022

View reviewed changes

...se/src/test/java/org/apache/flink/connector/firehose/sink/testutils/LocalstackContainer.java Outdated Show resolved Hide resolved

CrynetLogistics force-pushed the FLINK-24228 branch 2 times, most recently from 4161e82 to d782d9c Compare January 19, 2022 17:18

dannycranmer requested changes Jan 20, 2022

View reviewed changes

CrynetLogistics force-pushed the FLINK-24228 branch from d782d9c to 93cc3c5 Compare January 20, 2022 20:13

dannycranmer requested changes Jan 21, 2022

View reviewed changes

CrynetLogistics force-pushed the FLINK-24228 branch 8 times, most recently from 1f858a4 to 909ba3b Compare January 26, 2022 18:46

vahmed-hamdy approved these changes Jan 27, 2022

View reviewed changes

dannycranmer reviewed Jan 27, 2022

View reviewed changes

CrynetLogistics force-pushed the FLINK-24228 branch from 909ba3b to 162d5e3 Compare January 28, 2022 16:12

CrynetLogistics added 3 commits February 1, 2022 12:01

[FLINK-24228][connectors/firehose] Added documentation (&.zh docs) fo…

de0d9a7

…r the new Firehose sink.

[FLINK-24228][connectors/firehose] Allows end users to supply a seria…

cabbe4d

…lization schema rather than an ElementConverter, thereby encapsulating the Firehose `Record` from the user, verifying stream objects in KinesisFirehoseITCase

CrynetLogistics force-pushed the FLINK-24228 branch from 162d5e3 to cabbe4d Compare February 1, 2022 12:02

dannycranmer approved these changes Feb 1, 2022

View reviewed changes

dannycranmer merged commit 9050fd5 into apache:master Feb 1, 2022

flinkbot added component=Connectors/Kinesis and removed component=Connectors/Common labels Mar 17, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-24228][connectors/firehose] - Unified Async Sink for Kinesis Firehose #18314

[FLINK-24228][connectors/firehose] - Unified Async Sink for Kinesis Firehose #18314

CrynetLogistics commented Jan 10, 2022

flinkbot commented Jan 10, 2022

flinkbot commented Jan 10, 2022 •

edited

Loading

CrynetLogistics commented Jan 18, 2022

CrynetLogistics commented Jan 20, 2022

CrynetLogistics commented Jan 20, 2022

dannycranmer Jan 21, 2022

dannycranmer Jan 21, 2022

dannycranmer Jan 21, 2022

CrynetLogistics Jan 21, 2022

dannycranmer Feb 1, 2022

dannycranmer Jan 27, 2022

dannycranmer commented Feb 1, 2022

[FLINK-24228][connectors/firehose] - Unified Async Sink for Kinesis Firehose #18314

[FLINK-24228][connectors/firehose] - Unified Async Sink for Kinesis Firehose #18314

Conversation

CrynetLogistics commented Jan 10, 2022

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

flinkbot commented Jan 10, 2022

Automated Checks

Review Progress

flinkbot commented Jan 10, 2022 • edited Loading

CI report:

CrynetLogistics commented Jan 18, 2022

CrynetLogistics commented Jan 20, 2022

CrynetLogistics commented Jan 20, 2022

dannycranmer Jan 21, 2022

Choose a reason for hiding this comment

dannycranmer Jan 21, 2022

Choose a reason for hiding this comment

dannycranmer Jan 21, 2022

Choose a reason for hiding this comment

CrynetLogistics Jan 21, 2022

Choose a reason for hiding this comment

dannycranmer Feb 1, 2022

Choose a reason for hiding this comment

dannycranmer Jan 27, 2022

Choose a reason for hiding this comment

dannycranmer commented Feb 1, 2022

flinkbot commented Jan 10, 2022 •

edited

Loading