You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by da...@apache.org on 2022/02/01 15:35:49 UTC
[flink] 01/03: [FLINK-24228][connectors/firehose] Amazon Kinesis Data Firehose sink 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 [...]
This is an automated email from the ASF dual-hosted git repository.
dannycranmer pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 0e908be6e5344f6220a1329580d9e4535cd6ceb6
Author: Zichen Liu <zi...@amazon.com>
AuthorDate: Thu Dec 23 11:57:09 2021 +0000
[FLINK-24228][connectors/firehose] Amazon Kinesis Data Firehose sink 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.
---
flink-connectors/flink-connector-aws-base/pom.xml | 21 +++
.../connector/aws/util/AWSAsyncSinkUtil.java} | 107 ++++++++-------
.../aws/testutils/AWSServicesTestUtils.java | 129 ++++++++++++++++++
.../aws/testutils/LocalstackContainer.java | 78 +++++++++++
.../connector/aws/util/AWSAsyncSinkUtilTest.java} | 150 ++++++++++++---------
.../pom.xml | 12 --
.../KinesisDataStreamsConfigConstants.java} | 9 +-
.../kinesis/sink/KinesisDataStreamsSinkWriter.java | 10 +-
.../src/test/resources/log4j2-test.properties | 2 +-
.../pom.xml | 43 +++---
.../sink/KinesisFirehoseConfigConstants.java} | 17 ++-
.../firehose/sink/KinesisFirehoseException.java | 54 ++++++++
.../firehose/sink/KinesisFirehoseSink.java | 115 ++++++++++++++++
.../firehose/sink/KinesisFirehoseSinkBuilder.java | 149 ++++++++++++++++++++
.../sink/KinesisFirehoseSinkElementConverter.java | 76 +++++++++++
.../firehose/sink/KinesisFirehoseSinkWriter.java} | 120 +++++++++--------
.../src/main/resources/log4j2.properties | 25 ++++
.../sink/KinesisFirehoseSinkBuilderTest.java | 70 ++++++++++
.../KinesisFirehoseSinkElementConverterTest.java | 54 ++++++++
.../firehose/sink/KinesisFirehoseSinkITCase.java | 124 +++++++++++++++++
.../firehose/sink/KinesisFirehoseSinkTest.java | 76 +++++++++++
.../sink/KinesisFirehoseSinkWriterTest.java | 107 +++++++++++++++
.../firehose/sink/examples/SinkIntoFirehose.java | 74 ++++++++++
.../sink/testutils/KinesisFirehoseTestUtils.java | 73 ++++++++++
.../src/test/resources/log4j2-test.properties | 2 +-
flink-connectors/flink-connector-base/pom.xml | 16 +++
.../base/sink/writer/AsyncSinkWriterTest.java | 120 +----------------
.../base/sink/writer/TestSinkInitContext.java | 139 +++++++++++++++++++
.../connectors/kinesis/proxy/KinesisProxyV2.java | 4 +-
.../kinesis/proxy/KinesisProxyV2Factory.java | 18 ++-
.../streaming/connectors/kinesis/util/AWSUtil.java | 11 +-
.../connectors/kinesis/util/AwsV2Util.java | 8 ++
.../connectors/kinesis/util/AwsV2UtilTest.java | 34 +++++
flink-connectors/pom.xml | 1 +
.../org/apache/flink/util/DockerImageVersions.java | 2 +
tools/ci/stage.sh | 1 +
36 files changed, 1697 insertions(+), 354 deletions(-)
diff --git a/flink-connectors/flink-connector-aws-base/pom.xml b/flink-connectors/flink-connector-aws-base/pom.xml
index b774b2e..870e487 100644
--- a/flink-connectors/flink-connector-aws-base/pom.xml
+++ b/flink-connectors/flink-connector-aws-base/pom.xml
@@ -63,6 +63,27 @@ under the License.
<artifactId>sts</artifactId>
<version>${aws.sdk.version}</version>
</dependency>
+
+ <!-- Test dependencies -->
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>testcontainers</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>s3</artifactId>
+ <version>${aws.sdk.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>iam</artifactId>
+ <version>${aws.sdk.version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
diff --git a/flink-connectors/flink-connector-aws-kinesis-data-streams/src/main/java/org/apache/flink/connector/kinesis/util/AWSKinesisDataStreamsUtil.java b/flink-connectors/flink-connector-aws-base/src/main/java/org/apache/flink/connector/aws/util/AWSAsyncSinkUtil.java
similarity index 61%
rename from flink-connectors/flink-connector-aws-kinesis-data-streams/src/main/java/org/apache/flink/connector/kinesis/util/AWSKinesisDataStreamsUtil.java
rename to flink-connectors/flink-connector-aws-base/src/main/java/org/apache/flink/connector/aws/util/AWSAsyncSinkUtil.java
index 1ec75aa..1256142 100644
--- a/flink-connectors/flink-connector-aws-kinesis-data-streams/src/main/java/org/apache/flink/connector/kinesis/util/AWSKinesisDataStreamsUtil.java
+++ b/flink-connectors/flink-connector-aws-base/src/main/java/org/apache/flink/connector/aws/util/AWSAsyncSinkUtil.java
@@ -15,24 +15,21 @@
* limitations under the License.
*/
-package org.apache.flink.connector.kinesis.util;
+package org.apache.flink.connector.aws.util;
import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.connector.aws.config.AWSConfigConstants;
-import org.apache.flink.connector.aws.util.AWSGeneralUtil;
-import org.apache.flink.connector.kinesis.config.AWSKinesisDataStreamsConfigConstants;
import org.apache.flink.runtime.util.EnvironmentInformation;
+import software.amazon.awssdk.awscore.client.builder.AwsAsyncClientBuilder;
+import software.amazon.awssdk.awscore.client.builder.AwsClientBuilder;
+import software.amazon.awssdk.core.SdkClient;
import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration;
import software.amazon.awssdk.core.client.config.SdkAdvancedClientOption;
import software.amazon.awssdk.core.client.config.SdkClientConfiguration;
import software.amazon.awssdk.core.client.config.SdkClientOption;
import software.amazon.awssdk.http.async.SdkAsyncHttpClient;
-import software.amazon.awssdk.services.kinesis.KinesisAsyncClient;
-import software.amazon.awssdk.services.kinesis.KinesisAsyncClientBuilder;
-import software.amazon.awssdk.services.kinesis.model.LimitExceededException;
-import software.amazon.awssdk.services.kinesis.model.ProvisionedThroughputExceededException;
import java.net.URI;
import java.util.Optional;
@@ -40,11 +37,10 @@ import java.util.Properties;
/** Some utilities specific to Amazon Web Service. */
@Internal
-public class AWSKinesisDataStreamsUtil extends AWSGeneralUtil {
+public class AWSAsyncSinkUtil extends AWSGeneralUtil {
- /** Used for formatting Flink-specific user agent string when creating Kinesis client. */
- private static final String USER_AGENT_FORMAT =
- AWSKinesisDataStreamsConfigConstants.BASE_KINESIS_USER_AGENT_PREFIX_FORMAT + " V2";
+ /** V2 suffix to denote the unified sinks. V1 sinks are based on KPL etc. */
+ static final String V2_USER_AGENT_SUFFIX = " V2";
/**
* Creates a user agent prefix for Flink. This can be used by HTTP Clients.
@@ -62,51 +58,61 @@ public class AWSKinesisDataStreamsUtil extends AWSGeneralUtil {
/**
* @param configProps configuration properties
- * @param httpClient the underlying HTTP client used to talk to Kinesis
- * @return a new Amazon Kinesis Client
+ * @param httpClient the underlying HTTP client used to talk to AWS
+ * @return a new AWS Client
*/
- public static KinesisAsyncClient createKinesisAsyncClient(
- final Properties configProps, final SdkAsyncHttpClient httpClient) {
+ public static <
+ S extends SdkClient,
+ T extends
+ AwsAsyncClientBuilder<? extends T, S>
+ & AwsClientBuilder<? extends T, S>>
+ S createAwsAsyncClient(
+ final Properties configProps,
+ final SdkAsyncHttpClient httpClient,
+ final T clientBuilder,
+ final String awsUserAgentPrefixFormat,
+ final String awsClientUserAgentPrefix) {
SdkClientConfiguration clientConfiguration = SdkClientConfiguration.builder().build();
- return createKinesisAsyncClient(configProps, clientConfiguration, httpClient);
+ return createAwsAsyncClient(
+ configProps,
+ clientConfiguration,
+ httpClient,
+ clientBuilder,
+ awsUserAgentPrefixFormat,
+ awsClientUserAgentPrefix);
}
/**
* @param configProps configuration properties
* @param clientConfiguration the AWS SDK v2 config to instantiate the client
- * @param httpClient the underlying HTTP client used to talk to Kinesis
- * @return a new Amazon Kinesis Client
+ * @param httpClient the underlying HTTP client used to talk to AWS
+ * @return a new AWS Client
*/
- public static KinesisAsyncClient createKinesisAsyncClient(
- final Properties configProps,
- final SdkClientConfiguration clientConfiguration,
- final SdkAsyncHttpClient httpClient) {
+ public static <
+ S extends SdkClient,
+ T extends
+ AwsAsyncClientBuilder<? extends T, S>
+ & AwsClientBuilder<? extends T, S>>
+ S createAwsAsyncClient(
+ final Properties configProps,
+ final SdkClientConfiguration clientConfiguration,
+ final SdkAsyncHttpClient httpClient,
+ final T clientBuilder,
+ final String awsUserAgentPrefixFormat,
+ final String awsClientUserAgentPrefix) {
String flinkUserAgentPrefix =
- Optional.ofNullable(
- configProps.getProperty(
- AWSKinesisDataStreamsConfigConstants
- .KINESIS_CLIENT_USER_AGENT_PREFIX))
- .orElse(formatFlinkUserAgentPrefix(USER_AGENT_FORMAT));
+ Optional.ofNullable(configProps.getProperty(awsClientUserAgentPrefix))
+ .orElse(
+ formatFlinkUserAgentPrefix(
+ awsUserAgentPrefixFormat + V2_USER_AGENT_SUFFIX));
final ClientOverrideConfiguration overrideConfiguration =
createClientOverrideConfiguration(
clientConfiguration,
ClientOverrideConfiguration.builder(),
flinkUserAgentPrefix);
- final KinesisAsyncClientBuilder clientBuilder = KinesisAsyncClient.builder();
- return createKinesisAsyncClient(
- configProps, clientBuilder, httpClient, overrideConfiguration);
- }
-
- @VisibleForTesting
- static ClientOverrideConfiguration createClientOverrideConfiguration(
- final SdkClientConfiguration config,
- final ClientOverrideConfiguration.Builder overrideConfigurationBuilder) {
- return createClientOverrideConfiguration(
- config,
- overrideConfigurationBuilder,
- formatFlinkUserAgentPrefix(USER_AGENT_FORMAT));
+ return createAwsAsyncClient(configProps, clientBuilder, httpClient, overrideConfiguration);
}
@VisibleForTesting
@@ -131,11 +137,16 @@ public class AWSKinesisDataStreamsUtil extends AWSGeneralUtil {
}
@VisibleForTesting
- static KinesisAsyncClient createKinesisAsyncClient(
- final Properties configProps,
- final KinesisAsyncClientBuilder clientBuilder,
- final SdkAsyncHttpClient httpClient,
- final ClientOverrideConfiguration overrideConfiguration) {
+ static <
+ S extends SdkClient,
+ T extends
+ AwsAsyncClientBuilder<? extends T, S>
+ & AwsClientBuilder<? extends T, S>>
+ S createAwsAsyncClient(
+ final Properties configProps,
+ final T clientBuilder,
+ final SdkAsyncHttpClient httpClient,
+ final ClientOverrideConfiguration overrideConfiguration) {
if (configProps.containsKey(AWSConfigConstants.AWS_ENDPOINT)) {
final URI endpointOverride =
@@ -150,10 +161,4 @@ public class AWSKinesisDataStreamsUtil extends AWSGeneralUtil {
.region(getRegion(configProps))
.build();
}
-
- public static boolean isRecoverableException(Exception e) {
- Throwable cause = e.getCause();
- return cause instanceof LimitExceededException
- || cause instanceof ProvisionedThroughputExceededException;
- }
}
diff --git a/flink-connectors/flink-connector-aws-base/src/test/java/org/apache/flink/connector/aws/testutils/AWSServicesTestUtils.java b/flink-connectors/flink-connector-aws-base/src/test/java/org/apache/flink/connector/aws/testutils/AWSServicesTestUtils.java
new file mode 100644
index 0000000..543d81b
--- /dev/null
+++ b/flink-connectors/flink-connector-aws-base/src/test/java/org/apache/flink/connector/aws/testutils/AWSServicesTestUtils.java
@@ -0,0 +1,129 @@
+/*
+ * 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.aws.testutils;
+
+import org.apache.flink.connector.aws.config.AWSConfigConstants;
+import org.apache.flink.connector.aws.util.AWSGeneralUtil;
+
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.core.waiters.WaiterResponse;
+import software.amazon.awssdk.http.async.SdkAsyncHttpClient;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.iam.IamAsyncClient;
+import software.amazon.awssdk.services.iam.model.CreateRoleRequest;
+import software.amazon.awssdk.services.iam.model.CreateRoleResponse;
+import software.amazon.awssdk.services.s3.S3AsyncClient;
+import software.amazon.awssdk.services.s3.model.CreateBucketRequest;
+import software.amazon.awssdk.services.s3.model.HeadBucketRequest;
+import software.amazon.awssdk.services.s3.model.HeadBucketResponse;
+import software.amazon.awssdk.services.s3.model.ListObjectsRequest;
+import software.amazon.awssdk.services.s3.model.ListObjectsResponse;
+import software.amazon.awssdk.services.s3.model.S3Object;
+import software.amazon.awssdk.services.s3.waiters.S3AsyncWaiter;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+
+import static org.apache.flink.connector.aws.config.AWSConfigConstants.AWS_CREDENTIALS_PROVIDER;
+import static org.apache.flink.connector.aws.config.AWSConfigConstants.AWS_ENDPOINT;
+import static org.apache.flink.connector.aws.config.AWSConfigConstants.AWS_REGION;
+import static org.apache.flink.connector.aws.config.AWSConfigConstants.TRUST_ALL_CERTIFICATES;
+
+/**
+ * A set of static methods that can be used to call common AWS services on the Localstack container.
+ */
+public class AWSServicesTestUtils {
+
+ private static final String ACCESS_KEY_ID = "accessKeyId";
+ private static final String SECRET_ACCESS_KEY = "secretAccessKey";
+
+ public static S3AsyncClient getS3Client(String endpoint) throws URISyntaxException {
+ return S3AsyncClient.builder()
+ .httpClient(getHttpClient(endpoint))
+ .region(Region.AP_SOUTHEAST_1)
+ .endpointOverride(new URI(endpoint))
+ .credentialsProvider(getDefaultCredentials())
+ .build();
+ }
+
+ public static IamAsyncClient getIamClient(String endpoint) throws URISyntaxException {
+ return IamAsyncClient.builder()
+ .httpClient(getHttpClient(endpoint))
+ .region(Region.AWS_GLOBAL)
+ .endpointOverride(new URI(endpoint))
+ .credentialsProvider(getDefaultCredentials())
+ .build();
+ }
+
+ public static AwsCredentialsProvider getDefaultCredentials() {
+ return StaticCredentialsProvider.create(
+ AwsBasicCredentials.create(ACCESS_KEY_ID, SECRET_ACCESS_KEY));
+ }
+
+ public static Properties getConfig(String endpoint) {
+ Properties config = new Properties();
+ config.setProperty(AWS_REGION, Region.AP_SOUTHEAST_1.toString());
+ config.setProperty(AWS_ENDPOINT, endpoint);
+ config.setProperty(AWSConfigConstants.accessKeyId(AWS_CREDENTIALS_PROVIDER), ACCESS_KEY_ID);
+ config.setProperty(
+ AWSConfigConstants.secretKey(AWS_CREDENTIALS_PROVIDER), SECRET_ACCESS_KEY);
+ config.setProperty(TRUST_ALL_CERTIFICATES, "true");
+ return config;
+ }
+
+ public static SdkAsyncHttpClient getHttpClient(String endpoint) {
+ return AWSGeneralUtil.createAsyncHttpClient(getConfig(endpoint));
+ }
+
+ public static void createBucket(S3AsyncClient s3Client, String bucketName)
+ throws ExecutionException, InterruptedException {
+ CreateBucketRequest bucketRequest =
+ CreateBucketRequest.builder().bucket(bucketName).build();
+ s3Client.createBucket(bucketRequest);
+
+ HeadBucketRequest bucketRequestWait =
+ HeadBucketRequest.builder().bucket(bucketName).build();
+
+ S3AsyncWaiter s3Waiter = s3Client.waiter();
+ CompletableFuture<WaiterResponse<HeadBucketResponse>> waiterResponseFuture =
+ s3Waiter.waitUntilBucketExists(bucketRequestWait);
+
+ waiterResponseFuture.get();
+ }
+
+ public static void createIAMRole(IamAsyncClient iam, String roleName)
+ throws ExecutionException, InterruptedException {
+ CreateRoleRequest request = CreateRoleRequest.builder().roleName(roleName).build();
+
+ CompletableFuture<CreateRoleResponse> responseFuture = iam.createRole(request);
+ responseFuture.get();
+ }
+
+ public static List<S3Object> listBucketObjects(S3AsyncClient s3, String bucketName)
+ throws ExecutionException, InterruptedException {
+ ListObjectsRequest listObjects = ListObjectsRequest.builder().bucket(bucketName).build();
+ CompletableFuture<ListObjectsResponse> res = s3.listObjects(listObjects);
+ return res.get().contents();
+ }
+}
diff --git a/flink-connectors/flink-connector-aws-base/src/test/java/org/apache/flink/connector/aws/testutils/LocalstackContainer.java b/flink-connectors/flink-connector-aws-base/src/test/java/org/apache/flink/connector/aws/testutils/LocalstackContainer.java
new file mode 100644
index 0000000..bdc0517
--- /dev/null
+++ b/flink-connectors/flink-connector-aws-base/src/test/java/org/apache/flink/connector/aws/testutils/LocalstackContainer.java
@@ -0,0 +1,78 @@
+/*
+ * 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.aws.testutils;
+
+import org.rnorth.ducttape.ratelimits.RateLimiter;
+import org.rnorth.ducttape.ratelimits.RateLimiterBuilder;
+import org.rnorth.ducttape.unreliables.Unreliables;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.AbstractWaitStrategy;
+import org.testcontainers.utility.DockerImageName;
+import software.amazon.awssdk.services.s3.S3AsyncClient;
+import software.amazon.awssdk.services.s3.model.S3Object;
+
+import java.net.URISyntaxException;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+
+/**
+ * A class wrapping the Localstack container that provides mock implementations of many common AWS
+ * services.
+ */
+public class LocalstackContainer extends GenericContainer<LocalstackContainer> {
+
+ private static final int CONTAINER_PORT = 4566;
+
+ public LocalstackContainer(DockerImageName imageName) {
+ super(imageName);
+ withExposedPorts(CONTAINER_PORT);
+ waitingFor(new ListBucketObjectsWaitStrategy());
+ }
+
+ public String getEndpoint() {
+ return String.format("https://%s:%s", getHost(), getMappedPort(CONTAINER_PORT));
+ }
+
+ private class ListBucketObjectsWaitStrategy extends AbstractWaitStrategy {
+ private static final int TRANSACTIONS_PER_SECOND = 1;
+
+ private final RateLimiter rateLimiter =
+ RateLimiterBuilder.newBuilder()
+ .withRate(TRANSACTIONS_PER_SECOND, SECONDS)
+ .withConstantThroughput()
+ .build();
+
+ @Override
+ protected void waitUntilReady() {
+ Unreliables.retryUntilSuccess(
+ (int) startupTimeout.getSeconds(),
+ SECONDS,
+ () -> rateLimiter.getWhenReady(this::list));
+ }
+
+ private List<S3Object> list()
+ throws ExecutionException, InterruptedException, URISyntaxException {
+ String bucketName = "bucket-name-not-to-be-used";
+ S3AsyncClient client = AWSServicesTestUtils.getS3Client(getEndpoint());
+ AWSServicesTestUtils.createBucket(client, bucketName);
+ return AWSServicesTestUtils.listBucketObjects(client, bucketName);
+ }
+ }
+}
diff --git a/flink-connectors/flink-connector-aws-kinesis-data-streams/src/test/java/org/apache/flink/connector/kinesis/util/AWSKinesisDataStreamsUtilTest.java b/flink-connectors/flink-connector-aws-base/src/test/java/org/apache/flink/connector/aws/util/AWSAsyncSinkUtilTest.java
similarity index 62%
rename from flink-connectors/flink-connector-aws-kinesis-data-streams/src/test/java/org/apache/flink/connector/kinesis/util/AWSKinesisDataStreamsUtilTest.java
rename to flink-connectors/flink-connector-aws-base/src/test/java/org/apache/flink/connector/aws/util/AWSAsyncSinkUtilTest.java
index 0d6eae8..ccecdb8 100644
--- a/flink-connectors/flink-connector-aws-kinesis-data-streams/src/test/java/org/apache/flink/connector/kinesis/util/AWSKinesisDataStreamsUtilTest.java
+++ b/flink-connectors/flink-connector-aws-base/src/test/java/org/apache/flink/connector/aws/util/AWSAsyncSinkUtilTest.java
@@ -15,13 +15,15 @@
* limitations under the License.
*/
-package org.apache.flink.connector.kinesis.util;
-
-import org.apache.flink.connector.aws.util.TestUtil;
-import org.apache.flink.connector.kinesis.config.AWSKinesisDataStreamsConfigConstants;
+package org.apache.flink.connector.aws.util;
import org.junit.Test;
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
+import software.amazon.awssdk.awscore.client.builder.AwsAsyncClientBuilder;
+import software.amazon.awssdk.awscore.client.builder.AwsClientBuilder;
+import software.amazon.awssdk.core.SdkClient;
+import software.amazon.awssdk.core.client.config.ClientAsyncConfiguration;
import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration;
import software.amazon.awssdk.core.client.config.SdkAdvancedClientOption;
import software.amazon.awssdk.core.client.config.SdkClientConfiguration;
@@ -29,18 +31,14 @@ import software.amazon.awssdk.core.client.config.SdkClientOption;
import software.amazon.awssdk.http.async.SdkAsyncHttpClient;
import software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient;
import software.amazon.awssdk.regions.Region;
-import software.amazon.awssdk.services.kinesis.KinesisAsyncClientBuilder;
-import software.amazon.awssdk.services.kinesis.model.LimitExceededException;
import java.net.URI;
import java.time.Duration;
import java.util.Properties;
-import java.util.concurrent.ExecutionException;
import static org.apache.flink.connector.aws.config.AWSConfigConstants.AWS_ENDPOINT;
import static org.apache.flink.connector.aws.config.AWSConfigConstants.AWS_REGION;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
+import static org.apache.flink.connector.aws.util.AWSAsyncSinkUtil.formatFlinkUserAgentPrefix;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.argThat;
import static org.mockito.Mockito.mock;
@@ -48,20 +46,23 @@ import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
-/** Tests for {@link AWSKinesisDataStreamsUtil}. */
-public class AWSKinesisDataStreamsUtilTest {
+/** Tests for {@link AWSAsyncSinkUtil}. */
+public class AWSAsyncSinkUtilTest {
+
private static final String DEFAULT_USER_AGENT_PREFIX_FORMAT =
- AWSKinesisDataStreamsConfigConstants.BASE_KINESIS_USER_AGENT_PREFIX_FORMAT + " V2";
+ "Apache Flink %s (%s) *Destination* Connector";
+ private static final String DEFAULT_USER_AGENT_PREFIX_FORMAT_V2 =
+ "Apache Flink %s (%s) *Destination* Connector V2";
@Test
public void testCreateKinesisAsyncClient() {
Properties properties = TestUtil.properties(AWS_REGION, "eu-west-2");
- KinesisAsyncClientBuilder builder = mockKinesisAsyncClientBuilder();
+ MockAsyncClientBuilder builder = mockKinesisAsyncClientBuilder();
ClientOverrideConfiguration clientOverrideConfiguration =
ClientOverrideConfiguration.builder().build();
SdkAsyncHttpClient httpClient = NettyNioAsyncHttpClient.builder().build();
- AWSKinesisDataStreamsUtil.createKinesisAsyncClient(
+ AWSAsyncSinkUtil.createAwsAsyncClient(
properties, builder, httpClient, clientOverrideConfiguration);
verify(builder).overrideConfiguration(clientOverrideConfiguration);
@@ -77,12 +78,12 @@ public class AWSKinesisDataStreamsUtilTest {
Properties properties = TestUtil.properties(AWS_REGION, "eu-west-2");
properties.setProperty(AWS_ENDPOINT, "https://localhost");
- KinesisAsyncClientBuilder builder = mockKinesisAsyncClientBuilder();
+ MockAsyncClientBuilder builder = mockKinesisAsyncClientBuilder();
ClientOverrideConfiguration clientOverrideConfiguration =
ClientOverrideConfiguration.builder().build();
SdkAsyncHttpClient httpClient = NettyNioAsyncHttpClient.builder().build();
- AWSKinesisDataStreamsUtil.createKinesisAsyncClient(
+ AWSAsyncSinkUtil.createAwsAsyncClient(
properties, builder, httpClient, clientOverrideConfiguration);
verify(builder).endpointOverride(URI.create("https://localhost"));
@@ -94,14 +95,17 @@ public class AWSKinesisDataStreamsUtilTest {
ClientOverrideConfiguration.Builder builder = mockClientOverrideConfigurationBuilder();
- AWSKinesisDataStreamsUtil.createClientOverrideConfiguration(clientConfiguration, builder);
+ AWSAsyncSinkUtil.createClientOverrideConfiguration(
+ clientConfiguration,
+ builder,
+ formatFlinkUserAgentPrefix(
+ DEFAULT_USER_AGENT_PREFIX_FORMAT + AWSAsyncSinkUtil.V2_USER_AGENT_SUFFIX));
verify(builder).build();
verify(builder)
.putAdvancedOption(
SdkAdvancedClientOption.USER_AGENT_PREFIX,
- AWSKinesisDataStreamsUtil.formatFlinkUserAgentPrefix(
- DEFAULT_USER_AGENT_PREFIX_FORMAT));
+ formatFlinkUserAgentPrefix(DEFAULT_USER_AGENT_PREFIX_FORMAT_V2));
verify(builder).putAdvancedOption(SdkAdvancedClientOption.USER_AGENT_SUFFIX, null);
verify(builder, never()).apiCallAttemptTimeout(any());
verify(builder, never()).apiCallTimeout(any());
@@ -116,7 +120,11 @@ public class AWSKinesisDataStreamsUtilTest {
ClientOverrideConfiguration.Builder builder = mockClientOverrideConfigurationBuilder();
- AWSKinesisDataStreamsUtil.createClientOverrideConfiguration(clientConfiguration, builder);
+ AWSAsyncSinkUtil.createClientOverrideConfiguration(
+ clientConfiguration,
+ builder,
+ formatFlinkUserAgentPrefix(
+ DEFAULT_USER_AGENT_PREFIX_FORMAT + AWSAsyncSinkUtil.V2_USER_AGENT_SUFFIX));
verify(builder).putAdvancedOption(SdkAdvancedClientOption.USER_AGENT_SUFFIX, "suffix");
}
@@ -130,7 +138,12 @@ public class AWSKinesisDataStreamsUtilTest {
ClientOverrideConfiguration.Builder builder = mockClientOverrideConfigurationBuilder();
- AWSKinesisDataStreamsUtil.createClientOverrideConfiguration(clientConfiguration, builder);
+ AWSAsyncSinkUtil.createClientOverrideConfiguration(
+ clientConfiguration,
+ builder,
+ formatFlinkUserAgentPrefix(
+ DEFAULT_USER_AGENT_PREFIX_FORMAT_V2
+ + AWSAsyncSinkUtil.V2_USER_AGENT_SUFFIX));
verify(builder).apiCallAttemptTimeout(Duration.ofMillis(500));
}
@@ -144,49 +157,18 @@ public class AWSKinesisDataStreamsUtilTest {
ClientOverrideConfiguration.Builder builder = mockClientOverrideConfigurationBuilder();
- AWSKinesisDataStreamsUtil.createClientOverrideConfiguration(clientConfiguration, builder);
+ AWSAsyncSinkUtil.createClientOverrideConfiguration(
+ clientConfiguration,
+ builder,
+ formatFlinkUserAgentPrefix(
+ DEFAULT_USER_AGENT_PREFIX_FORMAT_V2
+ + AWSAsyncSinkUtil.V2_USER_AGENT_SUFFIX));
verify(builder).apiCallTimeout(Duration.ofMillis(600));
}
- @Test
- public void testIsRecoverableExceptionForRecoverable() {
- Exception recoverable = LimitExceededException.builder().build();
- assertTrue(
- AWSKinesisDataStreamsUtil.isRecoverableException(
- new ExecutionException(recoverable)));
- }
-
- @Test
- public void testIsRecoverableExceptionForNonRecoverable() {
- Exception nonRecoverable = new IllegalArgumentException("abc");
- assertFalse(
- AWSKinesisDataStreamsUtil.isRecoverableException(
- new ExecutionException(nonRecoverable)));
- }
-
- @Test
- public void testIsRecoverableExceptionForRuntimeExceptionWrappingRecoverable() {
- Exception recoverable = LimitExceededException.builder().build();
- Exception runtime = new RuntimeException("abc", recoverable);
- assertTrue(AWSKinesisDataStreamsUtil.isRecoverableException(runtime));
- }
-
- @Test
- public void testIsRecoverableExceptionForRuntimeExceptionWrappingNonRecoverable() {
- Exception nonRecoverable = new IllegalArgumentException("abc");
- Exception runtime = new RuntimeException("abc", nonRecoverable);
- assertFalse(AWSKinesisDataStreamsUtil.isRecoverableException(runtime));
- }
-
- @Test
- public void testIsRecoverableExceptionForNullCause() {
- Exception nonRecoverable = new IllegalArgumentException("abc");
- assertFalse(AWSKinesisDataStreamsUtil.isRecoverableException(nonRecoverable));
- }
-
- private KinesisAsyncClientBuilder mockKinesisAsyncClientBuilder() {
- KinesisAsyncClientBuilder builder = mock(KinesisAsyncClientBuilder.class);
+ private MockAsyncClientBuilder mockKinesisAsyncClientBuilder() {
+ MockAsyncClientBuilder builder = mock(MockAsyncClientBuilder.class);
when(builder.overrideConfiguration(any(ClientOverrideConfiguration.class)))
.thenReturn(builder);
when(builder.httpClient(any())).thenReturn(builder);
@@ -205,4 +187,52 @@ public class AWSKinesisDataStreamsUtilTest {
return builder;
}
+
+ private static class MockAsyncClientBuilder
+ implements AwsAsyncClientBuilder<MockAsyncClientBuilder, SdkClient>,
+ AwsClientBuilder<MockAsyncClientBuilder, SdkClient> {
+
+ @Override
+ public MockAsyncClientBuilder asyncConfiguration(
+ ClientAsyncConfiguration clientAsyncConfiguration) {
+ return null;
+ }
+
+ @Override
+ public MockAsyncClientBuilder httpClient(SdkAsyncHttpClient sdkAsyncHttpClient) {
+ return null;
+ }
+
+ @Override
+ public MockAsyncClientBuilder httpClientBuilder(SdkAsyncHttpClient.Builder builder) {
+ return null;
+ }
+
+ @Override
+ public MockAsyncClientBuilder credentialsProvider(
+ AwsCredentialsProvider awsCredentialsProvider) {
+ return null;
+ }
+
+ @Override
+ public MockAsyncClientBuilder region(Region region) {
+ return null;
+ }
+
+ @Override
+ public MockAsyncClientBuilder overrideConfiguration(
+ ClientOverrideConfiguration clientOverrideConfiguration) {
+ return null;
+ }
+
+ @Override
+ public MockAsyncClientBuilder endpointOverride(URI uri) {
+ return null;
+ }
+
+ @Override
+ public SdkClient build() {
+ return null;
+ }
+ }
}
diff --git a/flink-connectors/flink-connector-aws-kinesis-data-streams/pom.xml b/flink-connectors/flink-connector-aws-kinesis-data-streams/pom.xml
index 235fc3c..be50a2e 100644
--- a/flink-connectors/flink-connector-aws-kinesis-data-streams/pom.xml
+++ b/flink-connectors/flink-connector-aws-kinesis-data-streams/pom.xml
@@ -70,18 +70,6 @@ under the License.
<version>${aws.sdk.version}</version>
</dependency>
- <dependency>
- <groupId>software.amazon.awssdk</groupId>
- <artifactId>netty-nio-client</artifactId>
- <version>${aws.sdk.version}</version>
- </dependency>
-
- <dependency>
- <groupId>software.amazon.awssdk</groupId>
- <artifactId>sts</artifactId>
- <version>${aws.sdk.version}</version>
- </dependency>
-
<!--Table API dependencies-->
<dependency>
<groupId>org.apache.flink</groupId>
diff --git a/flink-connectors/flink-connector-aws-kinesis-data-streams/src/main/java/org/apache/flink/connector/kinesis/config/AWSKinesisDataStreamsConfigConstants.java b/flink-connectors/flink-connector-aws-kinesis-data-streams/src/main/java/org/apache/flink/connector/kinesis/sink/KinesisDataStreamsConfigConstants.java
similarity index 80%
copy from flink-connectors/flink-connector-aws-kinesis-data-streams/src/main/java/org/apache/flink/connector/kinesis/config/AWSKinesisDataStreamsConfigConstants.java
copy to flink-connectors/flink-connector-aws-kinesis-data-streams/src/main/java/org/apache/flink/connector/kinesis/sink/KinesisDataStreamsConfigConstants.java
index b318292..a5a6020 100644
--- a/flink-connectors/flink-connector-aws-kinesis-data-streams/src/main/java/org/apache/flink/connector/kinesis/config/AWSKinesisDataStreamsConfigConstants.java
+++ b/flink-connectors/flink-connector-aws-kinesis-data-streams/src/main/java/org/apache/flink/connector/kinesis/sink/KinesisDataStreamsConfigConstants.java
@@ -15,19 +15,18 @@
* limitations under the License.
*/
-package org.apache.flink.connector.kinesis.config;
+package org.apache.flink.connector.kinesis.sink;
import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.connector.kinesis.util.AWSKinesisDataStreamsUtil;
-/** Defaults for {@link AWSKinesisDataStreamsUtil}. */
+/** Defaults for {@link KinesisDataStreamsSinkWriter}. */
@PublicEvolving
-public class AWSKinesisDataStreamsConfigConstants {
+public class KinesisDataStreamsConfigConstants {
public static final String BASE_KINESIS_USER_AGENT_PREFIX_FORMAT =
"Apache Flink %s (%s) Kinesis Connector";
- /** Identifier for user agent prefix. */
+ /** Kinesis identifier for user agent prefix. */
public static final String KINESIS_CLIENT_USER_AGENT_PREFIX =
"aws.kinesis.client.user-agent-prefix";
}
diff --git a/flink-connectors/flink-connector-aws-kinesis-data-streams/src/main/java/org/apache/flink/connector/kinesis/sink/KinesisDataStreamsSinkWriter.java b/flink-connectors/flink-connector-aws-kinesis-data-streams/src/main/java/org/apache/flink/connector/kinesis/sink/KinesisDataStreamsSinkWriter.java
index b720e5e..b291c12 100644
--- a/flink-connectors/flink-connector-aws-kinesis-data-streams/src/main/java/org/apache/flink/connector/kinesis/sink/KinesisDataStreamsSinkWriter.java
+++ b/flink-connectors/flink-connector-aws-kinesis-data-streams/src/main/java/org/apache/flink/connector/kinesis/sink/KinesisDataStreamsSinkWriter.java
@@ -18,10 +18,10 @@
package org.apache.flink.connector.kinesis.sink;
import org.apache.flink.api.connector.sink.Sink;
+import org.apache.flink.connector.aws.util.AWSAsyncSinkUtil;
import org.apache.flink.connector.aws.util.AWSGeneralUtil;
import org.apache.flink.connector.base.sink.writer.AsyncSinkWriter;
import org.apache.flink.connector.base.sink.writer.ElementConverter;
-import org.apache.flink.connector.kinesis.util.AWSKinesisDataStreamsUtil;
import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
@@ -104,8 +104,12 @@ class KinesisDataStreamsSinkWriter<InputT> extends AsyncSinkWriter<InputT, PutRe
final SdkAsyncHttpClient httpClient =
AWSGeneralUtil.createAsyncHttpClient(kinesisClientProperties);
- return AWSKinesisDataStreamsUtil.createKinesisAsyncClient(
- kinesisClientProperties, httpClient);
+ return AWSAsyncSinkUtil.createAwsAsyncClient(
+ kinesisClientProperties,
+ httpClient,
+ KinesisAsyncClient.builder(),
+ KinesisDataStreamsConfigConstants.BASE_KINESIS_USER_AGENT_PREFIX_FORMAT,
+ KinesisDataStreamsConfigConstants.KINESIS_CLIENT_USER_AGENT_PREFIX);
}
@Override
diff --git a/flink-connectors/flink-connector-aws-kinesis-data-streams/src/test/resources/log4j2-test.properties b/flink-connectors/flink-connector-aws-kinesis-data-streams/src/test/resources/log4j2-test.properties
index c5339e7..c4fa187 100644
--- a/flink-connectors/flink-connector-aws-kinesis-data-streams/src/test/resources/log4j2-test.properties
+++ b/flink-connectors/flink-connector-aws-kinesis-data-streams/src/test/resources/log4j2-test.properties
@@ -18,7 +18,7 @@
# Set root logger level to OFF to not flood build logs
# set manually to INFO for debugging purposes
-rootLogger.level = INFO
+rootLogger.level = OFF
rootLogger.appenderRef.test.ref = TestLogger
appender.testlogger.name = TestLogger
diff --git a/flink-connectors/flink-connector-aws-kinesis-data-streams/pom.xml b/flink-connectors/flink-connector-aws-kinesis-firehose/pom.xml
similarity index 83%
copy from flink-connectors/flink-connector-aws-kinesis-data-streams/pom.xml
copy to flink-connectors/flink-connector-aws-kinesis-firehose/pom.xml
index 235fc3c..d391616 100644
--- a/flink-connectors/flink-connector-aws-kinesis-data-streams/pom.xml
+++ b/flink-connectors/flink-connector-aws-kinesis-firehose/pom.xml
@@ -30,8 +30,8 @@ under the License.
<relativePath>..</relativePath>
</parent>
- <artifactId>flink-connector-aws-kinesis-data-streams</artifactId>
- <name>Flink : Connectors : AWS Kinesis Data Streams</name>
+ <artifactId>flink-connector-aws-kinesis-firehose</artifactId>
+ <name>Flink : Connectors : AWS Kinesis Data Firehose</name>
<properties>
<aws.sdk.version>2.17.52</aws.sdk.version>
</properties>
@@ -66,7 +66,7 @@ under the License.
<dependency>
<groupId>software.amazon.awssdk</groupId>
- <artifactId>kinesis</artifactId>
+ <artifactId>firehose</artifactId>
<version>${aws.sdk.version}</version>
</dependency>
@@ -76,19 +76,6 @@ under the License.
<version>${aws.sdk.version}</version>
</dependency>
- <dependency>
- <groupId>software.amazon.awssdk</groupId>
- <artifactId>sts</artifactId>
- <version>${aws.sdk.version}</version>
- </dependency>
-
- <!--Table API dependencies-->
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-table-common</artifactId>
- <version>${project.version}</version>
- </dependency>
-
<!-- Test dependencies -->
<dependency>
<groupId>org.apache.flink</groupId>
@@ -105,28 +92,34 @@ under the License.
<scope>test</scope>
</dependency>
- <!-- Kinesis table factory testing -->
<dependency>
<groupId>org.apache.flink</groupId>
- <artifactId>flink-table-common</artifactId>
+ <artifactId>flink-connector-base</artifactId>
<version>${project.version}</version>
<type>test-jar</type>
<scope>test</scope>
</dependency>
+
<dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-table-planner_${scala.binary.version}</artifactId>
- <version>${project.version}</version>
- <type>test-jar</type>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>testcontainers</artifactId>
<scope>test</scope>
</dependency>
<dependency>
- <groupId>org.testcontainers</groupId>
- <artifactId>testcontainers</artifactId>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>s3</artifactId>
+ <version>${aws.sdk.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>iam</artifactId>
+ <version>${aws.sdk.version}</version>
<scope>test</scope>
</dependency>
- </dependencies>
+ </dependencies>
<build>
<plugins>
diff --git a/flink-connectors/flink-connector-aws-kinesis-data-streams/src/main/java/org/apache/flink/connector/kinesis/config/AWSKinesisDataStreamsConfigConstants.java b/flink-connectors/flink-connector-aws-kinesis-firehose/src/main/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseConfigConstants.java
similarity index 62%
rename from flink-connectors/flink-connector-aws-kinesis-data-streams/src/main/java/org/apache/flink/connector/kinesis/config/AWSKinesisDataStreamsConfigConstants.java
rename to flink-connectors/flink-connector-aws-kinesis-firehose/src/main/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseConfigConstants.java
index b318292..527f74a 100644
--- a/flink-connectors/flink-connector-aws-kinesis-data-streams/src/main/java/org/apache/flink/connector/kinesis/config/AWSKinesisDataStreamsConfigConstants.java
+++ b/flink-connectors/flink-connector-aws-kinesis-firehose/src/main/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseConfigConstants.java
@@ -15,19 +15,18 @@
* limitations under the License.
*/
-package org.apache.flink.connector.kinesis.config;
+package org.apache.flink.connector.firehose.sink;
import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.connector.kinesis.util.AWSKinesisDataStreamsUtil;
-/** Defaults for {@link AWSKinesisDataStreamsUtil}. */
+/** Defaults for {@link KinesisFirehoseSinkWriter}. */
@PublicEvolving
-public class AWSKinesisDataStreamsConfigConstants {
+public class KinesisFirehoseConfigConstants {
- public static final String BASE_KINESIS_USER_AGENT_PREFIX_FORMAT =
- "Apache Flink %s (%s) Kinesis Connector";
+ public static final String BASE_FIREHOSE_USER_AGENT_PREFIX_FORMAT =
+ "Apache Flink %s (%s) Firehose Connector";
- /** Identifier for user agent prefix. */
- public static final String KINESIS_CLIENT_USER_AGENT_PREFIX =
- "aws.kinesis.client.user-agent-prefix";
+ /** Firehose identifier for user agent prefix. */
+ public static final String FIREHOSE_CLIENT_USER_AGENT_PREFIX =
+ "aws.firehose.client.user-agent-prefix";
}
diff --git a/flink-connectors/flink-connector-aws-kinesis-firehose/src/main/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseException.java b/flink-connectors/flink-connector-aws-kinesis-firehose/src/main/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseException.java
new file mode 100644
index 0000000..ff6918c
--- /dev/null
+++ b/flink-connectors/flink-connector-aws-kinesis-firehose/src/main/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseException.java
@@ -0,0 +1,54 @@
+/*
+ * 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.firehose.sink;
+
+import org.apache.flink.annotation.PublicEvolving;
+
+/**
+ * A {@link RuntimeException} wrapper indicating the exception was thrown from the Kinesis Data
+ * Firehose Sink.
+ */
+@PublicEvolving
+class KinesisFirehoseException extends RuntimeException {
+
+ public KinesisFirehoseException(final String message) {
+ super(message);
+ }
+
+ public KinesisFirehoseException(final String message, final Throwable cause) {
+ super(message, cause);
+ }
+
+ /**
+ * When the flag {@code failOnError} is set in {@link KinesisFirehoseSinkWriter}, this exception
+ * is raised as soon as any exception occurs when writing to KDF.
+ */
+ static class KinesisFirehoseFailFastException extends KinesisFirehoseException {
+
+ public KinesisFirehoseFailFastException() {
+ super(
+ "Encountered an exception while persisting records, not retrying due to {failOnError} being set.");
+ }
+
+ public KinesisFirehoseFailFastException(final Throwable cause) {
+ super(
+ "Encountered an exception while persisting records, not retrying due to {failOnError} being set.",
+ cause);
+ }
+ }
+}
diff --git a/flink-connectors/flink-connector-aws-kinesis-firehose/src/main/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSink.java b/flink-connectors/flink-connector-aws-kinesis-firehose/src/main/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSink.java
new file mode 100644
index 0000000..50f04d4
--- /dev/null
+++ b/flink-connectors/flink-connector-aws-kinesis-firehose/src/main/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSink.java
@@ -0,0 +1,115 @@
+/*
+ * 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.firehose.sink;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.connector.sink.SinkWriter;
+import org.apache.flink.connector.base.sink.AsyncSinkBase;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.util.Preconditions;
+
+import software.amazon.awssdk.services.firehose.model.Record;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Optional;
+import java.util.Properties;
+
+/**
+ * A Kinesis Data Firehose (KDF) Sink that performs async requests against a destination delivery
+ * stream using the buffering protocol specified in {@link AsyncSinkBase}.
+ *
+ * <p>The sink internally uses a {@link
+ * software.amazon.awssdk.services.firehose.FirehoseAsyncClient} to communicate with the AWS
+ * endpoint.
+ *
+ * <p>Please see the writer implementation in {@link KinesisFirehoseSinkWriter}
+ *
+ * @param <InputT> Type of the elements handled by this sink
+ */
+@PublicEvolving
+public class KinesisFirehoseSink<InputT> extends AsyncSinkBase<InputT, Record> {
+
+ private final boolean failOnError;
+ private final String deliveryStreamName;
+ private final Properties firehoseClientProperties;
+
+ KinesisFirehoseSink(
+ ElementConverter<InputT, Record> elementConverter,
+ Integer maxBatchSize,
+ Integer maxInFlightRequests,
+ Integer maxBufferedRequests,
+ Long maxBatchSizeInBytes,
+ Long maxTimeInBufferMS,
+ Long maxRecordSizeInBytes,
+ boolean failOnError,
+ String deliveryStreamName,
+ Properties firehoseClientProperties) {
+ super(
+ elementConverter,
+ maxBatchSize,
+ maxInFlightRequests,
+ maxBufferedRequests,
+ maxBatchSizeInBytes,
+ maxTimeInBufferMS,
+ maxRecordSizeInBytes);
+ this.deliveryStreamName =
+ Preconditions.checkNotNull(
+ deliveryStreamName,
+ "The delivery stream name must not be null when initializing the KDF Sink.");
+ Preconditions.checkArgument(
+ !this.deliveryStreamName.isEmpty(),
+ "The delivery stream name must be set when initializing the KDF Sink.");
+ this.failOnError = failOnError;
+ this.firehoseClientProperties = firehoseClientProperties;
+ }
+
+ /**
+ * Create a {@link KinesisFirehoseSinkBuilder} to allow the fluent construction of a new {@code
+ * KinesisFirehoseSink}.
+ *
+ * @param <InputT> type of incoming records
+ * @return {@link KinesisFirehoseSinkBuilder}
+ */
+ public static <InputT> KinesisFirehoseSinkBuilder<InputT> builder() {
+ return new KinesisFirehoseSinkBuilder<>();
+ }
+
+ @Override
+ public SinkWriter<InputT, Void, Collection<Record>> createWriter(
+ InitContext context, List<Collection<Record>> states) {
+ return new KinesisFirehoseSinkWriter<>(
+ getElementConverter(),
+ context,
+ getMaxBatchSize(),
+ getMaxInFlightRequests(),
+ getMaxBufferedRequests(),
+ getMaxBatchSizeInBytes(),
+ getMaxTimeInBufferMS(),
+ getMaxRecordSizeInBytes(),
+ failOnError,
+ deliveryStreamName,
+ firehoseClientProperties);
+ }
+
+ @Override
+ public Optional<SimpleVersionedSerializer<Collection<Record>>> getWriterStateSerializer() {
+ return Optional.empty();
+ }
+}
diff --git a/flink-connectors/flink-connector-aws-kinesis-firehose/src/main/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkBuilder.java b/flink-connectors/flink-connector-aws-kinesis-firehose/src/main/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkBuilder.java
new file mode 100644
index 0000000..ee22e1f
--- /dev/null
+++ b/flink-connectors/flink-connector-aws-kinesis-firehose/src/main/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkBuilder.java
@@ -0,0 +1,149 @@
+/*
+ * 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.firehose.sink;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.connector.base.sink.AsyncSinkBaseBuilder;
+
+import software.amazon.awssdk.http.Protocol;
+import software.amazon.awssdk.services.firehose.model.Record;
+
+import java.util.Optional;
+import java.util.Properties;
+
+import static org.apache.flink.connector.aws.config.AWSConfigConstants.HTTP_PROTOCOL_VERSION;
+import static software.amazon.awssdk.http.Protocol.HTTP1_1;
+
+/**
+ * Builder to construct {@link KinesisFirehoseSink}.
+ *
+ * <p>The following example shows the minimum setup to create a {@link KinesisFirehoseSink} that
+ * writes String values to a Kinesis Data Firehose delivery stream named delivery-stream-name.
+ *
+ * <pre>{@code
+ * private static final KinesisFirehoseSinkElementConverter<String> elementConverter =
+ * KinesisFirehoseSinkElementConverter.<String>builder()
+ * .setSerializationSchema(new SimpleStringSchema())
+ * .build();
+ *
+ * Properties sinkProperties = new Properties();
+ * sinkProperties.put(AWSConfigConstants.AWS_REGION, "eu-west-1");
+ *
+ * KinesisFirehoseSink<String> kdfSink =
+ * KinesisFirehoseSink.<String>builder()
+ * .setElementConverter(elementConverter)
+ * .setDeliveryStreamName("delivery-stream-name")
+ * .setMaxBatchSize(20)
+ * .setFirehoseClientProperties(sinkProperties)
+ * .build();
+ * }</pre>
+ *
+ * <p>If the following parameters are not set in this builder, the following defaults will be used:
+ *
+ * <ul>
+ * <li>{@code maxBatchSize} will be 500
+ * <li>{@code maxInFlightRequests} will be 50
+ * <li>{@code maxBufferedRequests} will be 10000
+ * <li>{@code maxBatchSizeInBytes} will be 4 MB i.e. {@code 4 * 1024 * 1024}
+ * <li>{@code maxTimeInBufferMS} will be 5000ms
+ * <li>{@code maxRecordSizeInBytes} will be 1000 KB i.e. {@code 1000 * 1024}
+ * <li>{@code failOnError} will be false
+ * </ul>
+ *
+ * @param <InputT> type of elements that should be persisted in the destination
+ */
+@PublicEvolving
+public class KinesisFirehoseSinkBuilder<InputT>
+ extends AsyncSinkBaseBuilder<InputT, Record, KinesisFirehoseSinkBuilder<InputT>> {
+
+ private static final int DEFAULT_MAX_BATCH_SIZE = 500;
+ private static final int DEFAULT_MAX_IN_FLIGHT_REQUESTS = 50;
+ private static final int DEFAULT_MAX_BUFFERED_REQUESTS = 10000;
+ private static final long DEFAULT_MAX_BATCH_SIZE_IN_B = 4 * 1024 * 1024;
+ private static final long DEFAULT_MAX_TIME_IN_BUFFER_MS = 5000;
+ private static final long DEFAULT_MAX_RECORD_SIZE_IN_B = 1000 * 1024;
+ private static final boolean DEFAULT_FAIL_ON_ERROR = false;
+ private static final Protocol DEFAULT_HTTP_PROTOCOL = HTTP1_1;
+
+ private Boolean failOnError;
+ private String deliveryStreamName;
+ private Properties firehoseClientProperties;
+
+ KinesisFirehoseSinkBuilder() {}
+
+ /**
+ * Sets the name of the KDF delivery stream that the sink will connect to. There is no default
+ * for this parameter, therefore, this must be provided at sink creation time otherwise the
+ * build will fail.
+ *
+ * @param deliveryStreamName the name of the delivery stream
+ * @return {@link KinesisFirehoseSinkBuilder} itself
+ */
+ public KinesisFirehoseSinkBuilder<InputT> setDeliveryStreamName(String deliveryStreamName) {
+ this.deliveryStreamName = deliveryStreamName;
+ return this;
+ }
+
+ /**
+ * If writing to Kinesis Data Firehose results in a partial or full failure being returned, the
+ * job will fail immediately with a {@link KinesisFirehoseException} if failOnError is set.
+ *
+ * @param failOnError whether to fail on error
+ * @return {@link KinesisFirehoseSinkBuilder} itself
+ */
+ public KinesisFirehoseSinkBuilder<InputT> setFailOnError(boolean failOnError) {
+ this.failOnError = failOnError;
+ return this;
+ }
+
+ /**
+ * A set of properties used by the sink to create the firehose client. This may be used to set
+ * the aws region, credentials etc. See the docs for usage and syntax.
+ *
+ * @param firehoseClientProperties Firehose client properties
+ * @return {@link KinesisFirehoseSinkBuilder} itself
+ */
+ public KinesisFirehoseSinkBuilder<InputT> setFirehoseClientProperties(
+ Properties firehoseClientProperties) {
+ this.firehoseClientProperties = firehoseClientProperties;
+ return this;
+ }
+
+ private Properties getClientPropertiesWithDefaultHttpProtocol() {
+ Properties clientProperties =
+ Optional.ofNullable(firehoseClientProperties).orElse(new Properties());
+ clientProperties.putIfAbsent(HTTP_PROTOCOL_VERSION, DEFAULT_HTTP_PROTOCOL);
+ return clientProperties;
+ }
+
+ @Override
+ public KinesisFirehoseSink<InputT> build() {
+ return new KinesisFirehoseSink<>(
+ getElementConverter(),
+ Optional.ofNullable(getMaxBatchSize()).orElse(DEFAULT_MAX_BATCH_SIZE),
+ Optional.ofNullable(getMaxInFlightRequests())
+ .orElse(DEFAULT_MAX_IN_FLIGHT_REQUESTS),
+ Optional.ofNullable(getMaxBufferedRequests()).orElse(DEFAULT_MAX_BUFFERED_REQUESTS),
+ Optional.ofNullable(getMaxBatchSizeInBytes()).orElse(DEFAULT_MAX_BATCH_SIZE_IN_B),
+ Optional.ofNullable(getMaxTimeInBufferMS()).orElse(DEFAULT_MAX_TIME_IN_BUFFER_MS),
+ Optional.ofNullable(getMaxRecordSizeInBytes()).orElse(DEFAULT_MAX_RECORD_SIZE_IN_B),
+ Optional.ofNullable(failOnError).orElse(DEFAULT_FAIL_ON_ERROR),
+ deliveryStreamName,
+ getClientPropertiesWithDefaultHttpProtocol());
+ }
+}
diff --git a/flink-connectors/flink-connector-aws-kinesis-firehose/src/main/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkElementConverter.java b/flink-connectors/flink-connector-aws-kinesis-firehose/src/main/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkElementConverter.java
new file mode 100644
index 0000000..45b4186
--- /dev/null
+++ b/flink-connectors/flink-connector-aws-kinesis-firehose/src/main/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkElementConverter.java
@@ -0,0 +1,76 @@
+/*
+ * 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.firehose.sink;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.api.connector.sink.SinkWriter;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+import org.apache.flink.util.Preconditions;
+
+import software.amazon.awssdk.core.SdkBytes;
+import software.amazon.awssdk.services.firehose.model.Record;
+
+/**
+ * An implementation of the {@link ElementConverter} that uses the AWS Kinesis SDK v2. The user only
+ * needs to provide a {@link SerializationSchema} of the {@code InputT} to transform it into a
+ * {@link Record} that may be persisted.
+ */
+@PublicEvolving
+public class KinesisFirehoseSinkElementConverter<InputT>
+ implements ElementConverter<InputT, Record> {
+
+ /** A serialization schema to specify how the input element should be serialized. */
+ private final SerializationSchema<InputT> serializationSchema;
+
+ private KinesisFirehoseSinkElementConverter(SerializationSchema<InputT> serializationSchema) {
+ this.serializationSchema = serializationSchema;
+ }
+
+ @Override
+ public Record apply(InputT element, SinkWriter.Context context) {
+ return Record.builder()
+ .data(SdkBytes.fromByteArray(serializationSchema.serialize(element)))
+ .build();
+ }
+
+ public static <InputT> Builder<InputT> builder() {
+ return new Builder<>();
+ }
+
+ /** A builder for the KinesisFirehoseSinkElementConverter. */
+ @PublicEvolving
+ public static class Builder<InputT> {
+
+ private SerializationSchema<InputT> serializationSchema;
+
+ public Builder<InputT> setSerializationSchema(
+ SerializationSchema<InputT> serializationSchema) {
+ this.serializationSchema = serializationSchema;
+ return this;
+ }
+
+ public KinesisFirehoseSinkElementConverter<InputT> build() {
+ Preconditions.checkNotNull(
+ serializationSchema,
+ "No SerializationSchema was supplied to the "
+ + "KinesisFirehoseSinkElementConverter builder.");
+ return new KinesisFirehoseSinkElementConverter<>(serializationSchema);
+ }
+ }
+}
diff --git a/flink-connectors/flink-connector-aws-kinesis-data-streams/src/main/java/org/apache/flink/connector/kinesis/sink/KinesisDataStreamsSinkWriter.java b/flink-connectors/flink-connector-aws-kinesis-firehose/src/main/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkWriter.java
similarity index 54%
copy from flink-connectors/flink-connector-aws-kinesis-data-streams/src/main/java/org/apache/flink/connector/kinesis/sink/KinesisDataStreamsSinkWriter.java
copy to flink-connectors/flink-connector-aws-kinesis-firehose/src/main/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkWriter.java
index b720e5e..4002ec3 100644
--- a/flink-connectors/flink-connector-aws-kinesis-data-streams/src/main/java/org/apache/flink/connector/kinesis/sink/KinesisDataStreamsSinkWriter.java
+++ b/flink-connectors/flink-connector-aws-kinesis-firehose/src/main/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkWriter.java
@@ -15,27 +15,29 @@
* limitations under the License.
*/
-package org.apache.flink.connector.kinesis.sink;
+package org.apache.flink.connector.firehose.sink;
+import org.apache.flink.annotation.Internal;
import org.apache.flink.api.connector.sink.Sink;
+import org.apache.flink.connector.aws.util.AWSAsyncSinkUtil;
import org.apache.flink.connector.aws.util.AWSGeneralUtil;
import org.apache.flink.connector.base.sink.writer.AsyncSinkWriter;
import org.apache.flink.connector.base.sink.writer.ElementConverter;
-import org.apache.flink.connector.kinesis.util.AWSKinesisDataStreamsUtil;
import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import software.amazon.awssdk.http.async.SdkAsyncHttpClient;
-import software.amazon.awssdk.services.kinesis.KinesisAsyncClient;
-import software.amazon.awssdk.services.kinesis.model.PutRecordsRequest;
-import software.amazon.awssdk.services.kinesis.model.PutRecordsRequestEntry;
-import software.amazon.awssdk.services.kinesis.model.PutRecordsResponse;
-import software.amazon.awssdk.services.kinesis.model.PutRecordsResultEntry;
-import software.amazon.awssdk.services.kinesis.model.ResourceNotFoundException;
+import software.amazon.awssdk.services.firehose.FirehoseAsyncClient;
+import software.amazon.awssdk.services.firehose.model.PutRecordBatchRequest;
+import software.amazon.awssdk.services.firehose.model.PutRecordBatchResponse;
+import software.amazon.awssdk.services.firehose.model.PutRecordBatchResponseEntry;
+import software.amazon.awssdk.services.firehose.model.Record;
+import software.amazon.awssdk.services.firehose.model.ResourceNotFoundException;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Properties;
@@ -44,35 +46,36 @@ import java.util.concurrent.CompletionException;
import java.util.function.Consumer;
/**
- * Sink writer created by {@link KinesisDataStreamsSink} to write to Kinesis Data Streams. More
+ * Sink writer created by {@link KinesisFirehoseSink} to write to Kinesis Data Firehose. More
* details on the operation of this sink writer may be found in the doc for {@link
- * KinesisDataStreamsSink}. More details on the internals of this sink writer may be found in {@link
+ * KinesisFirehoseSink}. More details on the internals of this sink writer may be found in {@link
* AsyncSinkWriter}.
*
- * <p>The {@link KinesisAsyncClient} used here may be configured in the standard way for the AWS SDK
- * 2.x. e.g. the provision of {@code AWS_REGION}, {@code AWS_ACCESS_KEY_ID} and {@code
+ * <p>The {@link FirehoseAsyncClient} used here may be configured in the standard way for the AWS
+ * SDK 2.x. e.g. the provision of {@code AWS_REGION}, {@code AWS_ACCESS_KEY_ID} and {@code
* AWS_SECRET_ACCESS_KEY} through environment variables etc.
*/
-class KinesisDataStreamsSinkWriter<InputT> extends AsyncSinkWriter<InputT, PutRecordsRequestEntry> {
- private static final Logger LOG = LoggerFactory.getLogger(KinesisDataStreamsSinkWriter.class);
+@Internal
+class KinesisFirehoseSinkWriter<InputT> extends AsyncSinkWriter<InputT, Record> {
+ private static final Logger LOG = LoggerFactory.getLogger(KinesisFirehoseSinkWriter.class);
/* A counter for the total number of records that have encountered an error during put */
private final Counter numRecordsOutErrorsCounter;
- /* Name of the stream in Kinesis Data Streams */
- private final String streamName;
+ /* Name of the delivery stream in Kinesis Data Firehose */
+ private final String deliveryStreamName;
/* The sink writer metric group */
private final SinkWriterMetricGroup metrics;
- /* The asynchronous Kinesis client - construction is by kinesisClientProperties */
- private final KinesisAsyncClient client;
+ /* The asynchronous Firehose client - construction is by firehoseClientProperties */
+ private final FirehoseAsyncClient client;
/* Flag to whether fatally fail any time we encounter an exception when persisting records */
private final boolean failOnError;
- KinesisDataStreamsSinkWriter(
- ElementConverter<InputT, PutRecordsRequestEntry> elementConverter,
+ KinesisFirehoseSinkWriter(
+ ElementConverter<InputT, Record> elementConverter,
Sink.InitContext context,
int maxBatchSize,
int maxInFlightRequests,
@@ -81,8 +84,8 @@ class KinesisDataStreamsSinkWriter<InputT> extends AsyncSinkWriter<InputT, PutRe
long maxTimeInBufferMS,
long maxRecordSizeInBytes,
boolean failOnError,
- String streamName,
- Properties kinesisClientProperties) {
+ String deliveryStreamName,
+ Properties firehoseClientProperties) {
super(
elementConverter,
context,
@@ -93,38 +96,43 @@ class KinesisDataStreamsSinkWriter<InputT> extends AsyncSinkWriter<InputT, PutRe
maxTimeInBufferMS,
maxRecordSizeInBytes);
this.failOnError = failOnError;
- this.streamName = streamName;
+ this.deliveryStreamName = deliveryStreamName;
this.metrics = context.metricGroup();
this.numRecordsOutErrorsCounter = metrics.getNumRecordsOutErrorsCounter();
- this.client = buildClient(kinesisClientProperties);
+ this.client = buildClient(firehoseClientProperties);
}
- private KinesisAsyncClient buildClient(Properties kinesisClientProperties) {
-
+ private FirehoseAsyncClient buildClient(Properties firehoseClientProperties) {
final SdkAsyncHttpClient httpClient =
- AWSGeneralUtil.createAsyncHttpClient(kinesisClientProperties);
-
- return AWSKinesisDataStreamsUtil.createKinesisAsyncClient(
- kinesisClientProperties, httpClient);
+ AWSGeneralUtil.createAsyncHttpClient(firehoseClientProperties);
+
+ return AWSAsyncSinkUtil.createAwsAsyncClient(
+ firehoseClientProperties,
+ httpClient,
+ FirehoseAsyncClient.builder(),
+ KinesisFirehoseConfigConstants.BASE_FIREHOSE_USER_AGENT_PREFIX_FORMAT,
+ KinesisFirehoseConfigConstants.FIREHOSE_CLIENT_USER_AGENT_PREFIX);
}
@Override
protected void submitRequestEntries(
- List<PutRecordsRequestEntry> requestEntries,
- Consumer<List<PutRecordsRequestEntry>> requestResult) {
+ List<Record> requestEntries, Consumer<Collection<Record>> requestResult) {
- PutRecordsRequest batchRequest =
- PutRecordsRequest.builder().records(requestEntries).streamName(streamName).build();
+ PutRecordBatchRequest batchRequest =
+ PutRecordBatchRequest.builder()
+ .records(requestEntries)
+ .deliveryStreamName(deliveryStreamName)
+ .build();
- LOG.trace("Request to submit {} entries to KDS using KDS Sink.", requestEntries.size());
+ LOG.trace("Request to submit {} entries to KDF using KDF Sink.", requestEntries.size());
- CompletableFuture<PutRecordsResponse> future = client.putRecords(batchRequest);
+ CompletableFuture<PutRecordBatchResponse> future = client.putRecordBatch(batchRequest);
future.whenComplete(
(response, err) -> {
if (err != null) {
handleFullyFailedRequest(err, requestEntries, requestResult);
- } else if (response.failedRecordCount() > 0) {
+ } else if (response.failedPutCount() > 0) {
handlePartiallyFailedRequest(response, requestEntries, requestResult);
} else {
requestResult.accept(Collections.emptyList());
@@ -133,15 +141,19 @@ class KinesisDataStreamsSinkWriter<InputT> extends AsyncSinkWriter<InputT, PutRe
}
@Override
- protected long getSizeInBytes(PutRecordsRequestEntry requestEntry) {
+ protected long getSizeInBytes(Record requestEntry) {
return requestEntry.data().asByteArrayUnsafe().length;
}
private void handleFullyFailedRequest(
Throwable err,
- List<PutRecordsRequestEntry> requestEntries,
- Consumer<List<PutRecordsRequestEntry>> requestResult) {
- LOG.warn("KDS Sink failed to persist {} entries to KDS", requestEntries.size(), err);
+ List<Record> requestEntries,
+ Consumer<Collection<Record>> requestResult) {
+ LOG.warn(
+ "KDF Sink failed to persist {} entries to KDF first request was {}",
+ requestEntries.size(),
+ requestEntries.get(0).toString(),
+ err);
numRecordsOutErrorsCounter.inc(requestEntries.size());
if (isRetryable(err)) {
@@ -150,20 +162,22 @@ class KinesisDataStreamsSinkWriter<InputT> extends AsyncSinkWriter<InputT, PutRe
}
private void handlePartiallyFailedRequest(
- PutRecordsResponse response,
- List<PutRecordsRequestEntry> requestEntries,
- Consumer<List<PutRecordsRequestEntry>> requestResult) {
- LOG.warn("KDS Sink failed to persist {} entries to KDS", response.failedRecordCount());
- numRecordsOutErrorsCounter.inc(response.failedRecordCount());
+ PutRecordBatchResponse response,
+ List<Record> requestEntries,
+ Consumer<Collection<Record>> requestResult) {
+ LOG.warn(
+ "KDF Sink failed to persist {} entries to KDF first request was {}",
+ requestEntries.size(),
+ requestEntries.get(0).toString());
+ numRecordsOutErrorsCounter.inc(response.failedPutCount());
if (failOnError) {
getFatalExceptionCons()
- .accept(new KinesisDataStreamsException.KinesisDataStreamsFailFastException());
+ .accept(new KinesisFirehoseException.KinesisFirehoseFailFastException());
return;
}
- List<PutRecordsRequestEntry> failedRequestEntries =
- new ArrayList<>(response.failedRecordCount());
- List<PutRecordsResultEntry> records = response.records();
+ List<Record> failedRequestEntries = new ArrayList<>(response.failedPutCount());
+ List<PutRecordBatchResponseEntry> records = response.requestResponses();
for (int i = 0; i < records.size(); i++) {
if (records.get(i).errorCode() != null) {
@@ -179,15 +193,13 @@ class KinesisDataStreamsSinkWriter<InputT> extends AsyncSinkWriter<InputT, PutRe
&& err.getCause() instanceof ResourceNotFoundException) {
getFatalExceptionCons()
.accept(
- new KinesisDataStreamsException(
+ new KinesisFirehoseException(
"Encountered non-recoverable exception", err));
return false;
}
if (failOnError) {
getFatalExceptionCons()
- .accept(
- new KinesisDataStreamsException.KinesisDataStreamsFailFastException(
- err));
+ .accept(new KinesisFirehoseException.KinesisFirehoseFailFastException(err));
return false;
}
diff --git a/flink-connectors/flink-connector-aws-kinesis-firehose/src/main/resources/log4j2.properties b/flink-connectors/flink-connector-aws-kinesis-firehose/src/main/resources/log4j2.properties
new file mode 100644
index 0000000..c64a340
--- /dev/null
+++ b/flink-connectors/flink-connector-aws-kinesis-firehose/src/main/resources/log4j2.properties
@@ -0,0 +1,25 @@
+################################################################################
+# 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.
+################################################################################
+
+rootLogger.level = OFF
+rootLogger.appenderRef.console.ref = ConsoleAppender
+
+appender.console.name = ConsoleAppender
+appender.console.type = CONSOLE
+appender.console.layout.type = PatternLayout
+appender.console.layout.pattern = %d{HH:mm:ss,SSS} %-5p %-60c %x - %m%n
diff --git a/flink-connectors/flink-connector-aws-kinesis-firehose/src/test/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkBuilderTest.java b/flink-connectors/flink-connector-aws-kinesis-firehose/src/test/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkBuilderTest.java
new file mode 100644
index 0000000..0dc85da
--- /dev/null
+++ b/flink-connectors/flink-connector-aws-kinesis-firehose/src/test/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkBuilderTest.java
@@ -0,0 +1,70 @@
+/*
+ * 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.firehose.sink;
+
+import org.apache.flink.api.common.serialization.SimpleStringSchema;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+
+import org.assertj.core.api.Assertions;
+import org.junit.Test;
+import software.amazon.awssdk.services.firehose.model.Record;
+
+/** Covers construction, defaults and sanity checking of {@link KinesisFirehoseSinkBuilder}. */
+public class KinesisFirehoseSinkBuilderTest {
+ private static final ElementConverter<String, Record> ELEMENT_CONVERTER_PLACEHOLDER =
+ KinesisFirehoseSinkElementConverter.<String>builder()
+ .setSerializationSchema(new SimpleStringSchema())
+ .build();
+
+ @Test
+ public void elementConverterOfSinkMustBeSetWhenBuilt() {
+ Assertions.assertThatExceptionOfType(NullPointerException.class)
+ .isThrownBy(
+ () ->
+ KinesisFirehoseSink.builder()
+ .setDeliveryStreamName("deliveryStream")
+ .build())
+ .withMessageContaining(
+ "ElementConverter must be not null when initializing the AsyncSinkBase.");
+ }
+
+ @Test
+ public void streamNameOfSinkMustBeSetWhenBuilt() {
+ Assertions.assertThatExceptionOfType(NullPointerException.class)
+ .isThrownBy(
+ () ->
+ KinesisFirehoseSink.<String>builder()
+ .setElementConverter(ELEMENT_CONVERTER_PLACEHOLDER)
+ .build())
+ .withMessageContaining(
+ "The delivery stream name must not be null when initializing the KDF Sink.");
+ }
+
+ @Test
+ public void streamNameOfSinkMustBeSetToNonEmptyWhenBuilt() {
+ Assertions.assertThatExceptionOfType(IllegalArgumentException.class)
+ .isThrownBy(
+ () ->
+ KinesisFirehoseSink.<String>builder()
+ .setDeliveryStreamName("")
+ .setElementConverter(ELEMENT_CONVERTER_PLACEHOLDER)
+ .build())
+ .withMessageContaining(
+ "The delivery stream name must be set when initializing the KDF Sink.");
+ }
+}
diff --git a/flink-connectors/flink-connector-aws-kinesis-firehose/src/test/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkElementConverterTest.java b/flink-connectors/flink-connector-aws-kinesis-firehose/src/test/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkElementConverterTest.java
new file mode 100644
index 0000000..ed0b1c7
--- /dev/null
+++ b/flink-connectors/flink-connector-aws-kinesis-firehose/src/test/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkElementConverterTest.java
@@ -0,0 +1,54 @@
+/*
+ * 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.firehose.sink;
+
+import org.apache.flink.api.common.serialization.SimpleStringSchema;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+
+import org.assertj.core.api.Assertions;
+import org.junit.Test;
+import software.amazon.awssdk.core.SdkBytes;
+import software.amazon.awssdk.services.firehose.model.Record;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Covers construction and sanity checking of {@link KinesisFirehoseSinkElementConverter}. */
+public class KinesisFirehoseSinkElementConverterTest {
+
+ @Test
+ public void elementConverterWillComplainASerializationSchemaIsNotSetIfBuildIsCalledWithoutIt() {
+ Assertions.assertThatExceptionOfType(NullPointerException.class)
+ .isThrownBy(() -> KinesisFirehoseSinkElementConverter.<String>builder().build())
+ .withMessageContaining(
+ "No SerializationSchema was supplied to the KinesisFirehoseSinkElementConverter builder.");
+ }
+
+ @Test
+ public void elementConverterUsesProvidedSchemaToSerializeRecord() {
+ ElementConverter<String, Record> elementConverter =
+ KinesisFirehoseSinkElementConverter.<String>builder()
+ .setSerializationSchema(new SimpleStringSchema())
+ .build();
+
+ String testString = "{many hands make light work;";
+
+ Record serializedRecord = elementConverter.apply(testString, null);
+ byte[] serializedString = (new SimpleStringSchema()).serialize(testString);
+ assertThat(serializedRecord.data()).isEqualTo(SdkBytes.fromByteArray(serializedString));
+ }
+}
diff --git a/flink-connectors/flink-connector-aws-kinesis-firehose/src/test/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkITCase.java b/flink-connectors/flink-connector-aws-kinesis-firehose/src/test/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkITCase.java
new file mode 100644
index 0000000..08cb49b
--- /dev/null
+++ b/flink-connectors/flink-connector-aws-kinesis-firehose/src/test/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkITCase.java
@@ -0,0 +1,124 @@
+/*
+ * 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.firehose.sink;
+
+import org.apache.flink.api.common.serialization.SimpleStringSchema;
+import org.apache.flink.connector.aws.testutils.LocalstackContainer;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.util.DockerImageVersions;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.utility.DockerImageName;
+import software.amazon.awssdk.core.SdkSystemSetting;
+import software.amazon.awssdk.services.firehose.FirehoseAsyncClient;
+import software.amazon.awssdk.services.firehose.model.Record;
+import software.amazon.awssdk.services.iam.IamAsyncClient;
+import software.amazon.awssdk.services.s3.S3AsyncClient;
+import software.amazon.awssdk.services.s3.model.S3Object;
+import software.amazon.awssdk.utils.ImmutableMap;
+
+import java.util.List;
+
+import static org.apache.flink.connector.aws.testutils.AWSServicesTestUtils.createBucket;
+import static org.apache.flink.connector.aws.testutils.AWSServicesTestUtils.createIAMRole;
+import static org.apache.flink.connector.aws.testutils.AWSServicesTestUtils.getConfig;
+import static org.apache.flink.connector.aws.testutils.AWSServicesTestUtils.getIamClient;
+import static org.apache.flink.connector.aws.testutils.AWSServicesTestUtils.getS3Client;
+import static org.apache.flink.connector.aws.testutils.AWSServicesTestUtils.listBucketObjects;
+import static org.apache.flink.connector.firehose.sink.testutils.KinesisFirehoseTestUtils.createDeliveryStream;
+import static org.apache.flink.connector.firehose.sink.testutils.KinesisFirehoseTestUtils.getFirehoseClient;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Integration test suite for the {@code KinesisFirehoseSink} using a localstack container. */
+public class KinesisFirehoseSinkITCase {
+
+ private static final ElementConverter<String, Record> elementConverter =
+ KinesisFirehoseSinkElementConverter.<String>builder()
+ .setSerializationSchema(new SimpleStringSchema())
+ .build();
+
+ private static final Logger LOG = LoggerFactory.getLogger(KinesisFirehoseSinkITCase.class);
+ private S3AsyncClient s3AsyncClient;
+ private FirehoseAsyncClient firehoseAsyncClient;
+ private IamAsyncClient iamAsyncClient;
+
+ private static final String ROLE_NAME = "super-role";
+ private static final String ROLE_ARN = "arn:aws:iam::000000000000:role/" + ROLE_NAME;
+ private static final String BUCKET_NAME = "s3-firehose";
+ private static final String STREAM_NAME = "s3-stream";
+ private static final int NUMBER_OF_ELEMENTS = 92;
+
+ @ClassRule
+ public static LocalstackContainer mockFirehoseContainer =
+ new LocalstackContainer(DockerImageName.parse(DockerImageVersions.LOCALSTACK));
+
+ @Before
+ public void setup() throws Exception {
+ System.setProperty(SdkSystemSetting.CBOR_ENABLED.property(), "false");
+ s3AsyncClient = getS3Client(mockFirehoseContainer.getEndpoint());
+ firehoseAsyncClient = getFirehoseClient(mockFirehoseContainer.getEndpoint());
+ iamAsyncClient = getIamClient(mockFirehoseContainer.getEndpoint());
+ }
+
+ @After
+ public void teardown() {
+ System.clearProperty(SdkSystemSetting.CBOR_ENABLED.property());
+ }
+
+ @Test
+ public void test() throws Exception {
+ LOG.info("1 - Creating the bucket for Firehose to deliver into...");
+ createBucket(s3AsyncClient, BUCKET_NAME);
+ LOG.info("2 - Creating the IAM Role for Firehose to write into the s3 bucket...");
+ createIAMRole(iamAsyncClient, ROLE_NAME);
+ LOG.info("3 - Creating the Firehose delivery stream...");
+ createDeliveryStream(STREAM_NAME, BUCKET_NAME, ROLE_ARN, firehoseAsyncClient);
+
+ ObjectMapper mapper = new ObjectMapper();
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<String> generator =
+ env.fromSequence(1, NUMBER_OF_ELEMENTS)
+ .map(Object::toString)
+ .returns(String.class)
+ .map(data -> mapper.writeValueAsString(ImmutableMap.of("data", data)));
+
+ KinesisFirehoseSink<String> kdsSink =
+ KinesisFirehoseSink.<String>builder()
+ .setElementConverter(elementConverter)
+ .setDeliveryStreamName(STREAM_NAME)
+ .setMaxBatchSize(1)
+ .setFirehoseClientProperties(getConfig(mockFirehoseContainer.getEndpoint()))
+ .build();
+
+ generator.sinkTo(kdsSink);
+ env.execute("Integration Test");
+
+ List<S3Object> objects = listBucketObjects(s3AsyncClient, BUCKET_NAME);
+ assertThat(objects.size()).isEqualTo(NUMBER_OF_ELEMENTS);
+ }
+}
diff --git a/flink-connectors/flink-connector-aws-kinesis-firehose/src/test/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkTest.java b/flink-connectors/flink-connector-aws-kinesis-firehose/src/test/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkTest.java
new file mode 100644
index 0000000..164ec39
--- /dev/null
+++ b/flink-connectors/flink-connector-aws-kinesis-firehose/src/test/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkTest.java
@@ -0,0 +1,76 @@
+/*
+ * 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.firehose.sink;
+
+import org.apache.flink.api.common.serialization.SimpleStringSchema;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+
+import org.assertj.core.api.Assertions;
+import org.junit.Test;
+import software.amazon.awssdk.services.firehose.model.Record;
+
+import java.util.Properties;
+
+/** Covers construction, defaults and sanity checking of {@link KinesisFirehoseSink}. */
+public class KinesisFirehoseSinkTest {
+
+ private static final ElementConverter<String, Record> elementConverter =
+ KinesisFirehoseSinkElementConverter.<String>builder()
+ .setSerializationSchema(new SimpleStringSchema())
+ .build();
+
+ @Test
+ public void deliveryStreamNameMustNotBeNull() {
+ Assertions.assertThatExceptionOfType(NullPointerException.class)
+ .isThrownBy(
+ () ->
+ new KinesisFirehoseSink<>(
+ elementConverter,
+ 500,
+ 16,
+ 10000,
+ 4 * 1024 * 1024L,
+ 5000L,
+ 1000 * 1024L,
+ false,
+ null,
+ new Properties()))
+ .withMessageContaining(
+ "The delivery stream name must not be null when initializing the KDF Sink.");
+ }
+
+ @Test
+ public void deliveryStreamNameMustNotBeEmpty() {
+ Assertions.assertThatExceptionOfType(IllegalArgumentException.class)
+ .isThrownBy(
+ () ->
+ new KinesisFirehoseSink<>(
+ elementConverter,
+ 500,
+ 16,
+ 10000,
+ 4 * 1024 * 1024L,
+ 5000L,
+ 1000 * 1024L,
+ false,
+ "",
+ new Properties()))
+ .withMessageContaining(
+ "The delivery stream name must be set when initializing the KDF Sink.");
+ }
+}
diff --git a/flink-connectors/flink-connector-aws-kinesis-firehose/src/test/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkWriterTest.java b/flink-connectors/flink-connector-aws-kinesis-firehose/src/test/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkWriterTest.java
new file mode 100644
index 0000000..d840033
--- /dev/null
+++ b/flink-connectors/flink-connector-aws-kinesis-firehose/src/test/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkWriterTest.java
@@ -0,0 +1,107 @@
+/*
+ * 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.firehose.sink;
+
+import org.apache.flink.api.common.serialization.SimpleStringSchema;
+import org.apache.flink.api.connector.sink.SinkWriter;
+import org.apache.flink.connector.aws.config.AWSConfigConstants;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+import org.apache.flink.connector.base.sink.writer.TestSinkInitContext;
+
+import org.junit.Before;
+import org.junit.Test;
+import software.amazon.awssdk.core.SdkBytes;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.firehose.model.Record;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Properties;
+
+import static org.apache.flink.connector.aws.config.AWSConfigConstants.AWS_ENDPOINT;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Covers construction, defaults and sanity checking of {@link KinesisFirehoseSinkWriter}. */
+public class KinesisFirehoseSinkWriterTest {
+
+ private KinesisFirehoseSinkWriter<String> sinkWriter;
+
+ private static final ElementConverter<String, Record> ELEMENT_CONVERTER_PLACEHOLDER =
+ KinesisFirehoseSinkElementConverter.<String>builder()
+ .setSerializationSchema(new SimpleStringSchema())
+ .build();
+
+ @Before
+ public void setup() {
+ TestSinkInitContext sinkInitContext = new TestSinkInitContext();
+ Properties sinkProperties = new Properties();
+ sinkProperties.put(AWSConfigConstants.AWS_REGION, "eu-west-1");
+ sinkWriter =
+ new KinesisFirehoseSinkWriter<>(
+ ELEMENT_CONVERTER_PLACEHOLDER,
+ sinkInitContext,
+ 50,
+ 16,
+ 10000,
+ 4 * 1024 * 1024,
+ 5000,
+ 1000 * 1024,
+ true,
+ "streamName",
+ sinkProperties);
+ }
+
+ @Test
+ public void getSizeInBytesReturnsSizeOfBlobBeforeBase64Encoding() {
+ String testString = "{many hands make light work;";
+ Record record = Record.builder().data(SdkBytes.fromUtf8String(testString)).build();
+ assertThat(sinkWriter.getSizeInBytes(record))
+ .isEqualTo(testString.getBytes(StandardCharsets.US_ASCII).length);
+ }
+
+ @Test
+ public void getNumRecordsOutErrorsCounterRecordsCorrectNumberOfFailures()
+ throws IOException, InterruptedException {
+ Properties prop = new Properties();
+ prop.setProperty(AWSConfigConstants.AWS_REGION, Region.EU_WEST_1.toString());
+ prop.setProperty(AWS_ENDPOINT, "https://fake_aws_endpoint");
+ TestSinkInitContext ctx = new TestSinkInitContext();
+ KinesisFirehoseSink<String> kinesisFirehoseSink =
+ new KinesisFirehoseSink<>(
+ ELEMENT_CONVERTER_PLACEHOLDER,
+ 6,
+ 16,
+ 10000,
+ 4 * 1024 * 1024L,
+ 5000L,
+ 1000 * 1024L,
+ true,
+ "test-stream",
+ prop);
+ SinkWriter<String, Void, Collection<Record>> writer =
+ kinesisFirehoseSink.createWriter(ctx, new ArrayList<>());
+
+ for (int i = 0; i < 12; i++) {
+ writer.write("data_bytes", null);
+ }
+
+ assertThat(ctx.metricGroup().getNumRecordsOutErrorsCounter().getCount()).isEqualTo(12);
+ }
+}
diff --git a/flink-connectors/flink-connector-aws-kinesis-firehose/src/test/java/org/apache/flink/connector/firehose/sink/examples/SinkIntoFirehose.java b/flink-connectors/flink-connector-aws-kinesis-firehose/src/test/java/org/apache/flink/connector/firehose/sink/examples/SinkIntoFirehose.java
new file mode 100644
index 0000000..3e9d0ee
--- /dev/null
+++ b/flink-connectors/flink-connector-aws-kinesis-firehose/src/test/java/org/apache/flink/connector/firehose/sink/examples/SinkIntoFirehose.java
@@ -0,0 +1,74 @@
+/*
+ * 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.firehose.sink.examples;
+
+import org.apache.flink.api.common.serialization.SimpleStringSchema;
+import org.apache.flink.connector.aws.config.AWSConfigConstants;
+import org.apache.flink.connector.firehose.sink.KinesisFirehoseSink;
+import org.apache.flink.connector.firehose.sink.KinesisFirehoseSinkElementConverter;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+
+import software.amazon.awssdk.services.firehose.FirehoseAsyncClient;
+import software.amazon.awssdk.utils.ImmutableMap;
+
+import java.util.Properties;
+
+/**
+ * An example application demonstrating how to use the {@link KinesisFirehoseSink} to sink into KDF.
+ *
+ * <p>The {@link FirehoseAsyncClient} used here may be configured in the standard way for the AWS
+ * SDK 2.x. e.g. the provision of {@code AWS_ACCESS_KEY_ID} and {@code AWS_SECRET_ACCESS_KEY}
+ * through environment variables etc.
+ */
+public class SinkIntoFirehose {
+
+ private static final KinesisFirehoseSinkElementConverter<String> elementConverter =
+ KinesisFirehoseSinkElementConverter.<String>builder()
+ .setSerializationSchema(new SimpleStringSchema())
+ .build();
+
+ public static void main(String[] args) throws Exception {
+ ObjectMapper mapper = new ObjectMapper();
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.enableCheckpointing(10_000);
+
+ DataStream<String> generator =
+ env.fromSequence(1, 10_000_000L)
+ .map(Object::toString)
+ .returns(String.class)
+ .map(data -> mapper.writeValueAsString(ImmutableMap.of("data", data)));
+
+ Properties sinkProperties = new Properties();
+ sinkProperties.put(AWSConfigConstants.AWS_REGION, "eu-west-1");
+
+ KinesisFirehoseSink<String> kdfSink =
+ KinesisFirehoseSink.<String>builder()
+ .setElementConverter(elementConverter)
+ .setDeliveryStreamName("delivery-stream")
+ .setMaxBatchSize(20)
+ .setFirehoseClientProperties(sinkProperties)
+ .build();
+
+ generator.sinkTo(kdfSink);
+
+ env.execute("KDF Async Sink Example Program");
+ }
+}
diff --git a/flink-connectors/flink-connector-aws-kinesis-firehose/src/test/java/org/apache/flink/connector/firehose/sink/testutils/KinesisFirehoseTestUtils.java b/flink-connectors/flink-connector-aws-kinesis-firehose/src/test/java/org/apache/flink/connector/firehose/sink/testutils/KinesisFirehoseTestUtils.java
new file mode 100644
index 0000000..1de389d
--- /dev/null
+++ b/flink-connectors/flink-connector-aws-kinesis-firehose/src/test/java/org/apache/flink/connector/firehose/sink/testutils/KinesisFirehoseTestUtils.java
@@ -0,0 +1,73 @@
+/*
+ * 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.firehose.sink.testutils;
+
+import org.apache.flink.connector.aws.util.AWSAsyncSinkUtil;
+import org.apache.flink.connector.firehose.sink.KinesisFirehoseConfigConstants;
+
+import software.amazon.awssdk.services.firehose.FirehoseAsyncClient;
+import software.amazon.awssdk.services.firehose.model.CreateDeliveryStreamRequest;
+import software.amazon.awssdk.services.firehose.model.CreateDeliveryStreamResponse;
+import software.amazon.awssdk.services.firehose.model.DeliveryStreamType;
+import software.amazon.awssdk.services.firehose.model.ExtendedS3DestinationConfiguration;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+
+import static org.apache.flink.connector.aws.testutils.AWSServicesTestUtils.getConfig;
+import static org.apache.flink.connector.aws.testutils.AWSServicesTestUtils.getHttpClient;
+
+/**
+ * A set of static methods that can be used to call common AWS services on the Localstack container.
+ */
+public class KinesisFirehoseTestUtils {
+
+ public static FirehoseAsyncClient getFirehoseClient(String endpoint) throws URISyntaxException {
+ return AWSAsyncSinkUtil.createAwsAsyncClient(
+ getConfig(endpoint),
+ getHttpClient(endpoint),
+ FirehoseAsyncClient.builder().endpointOverride(new URI(endpoint)),
+ KinesisFirehoseConfigConstants.BASE_FIREHOSE_USER_AGENT_PREFIX_FORMAT,
+ KinesisFirehoseConfigConstants.FIREHOSE_CLIENT_USER_AGENT_PREFIX);
+ }
+
+ public static void createDeliveryStream(
+ String deliveryStreamName,
+ String bucketName,
+ String roleARN,
+ FirehoseAsyncClient firehoseAsyncClient)
+ throws ExecutionException, InterruptedException {
+ ExtendedS3DestinationConfiguration s3Config =
+ ExtendedS3DestinationConfiguration.builder()
+ .bucketARN(bucketName)
+ .roleARN(roleARN)
+ .build();
+ CreateDeliveryStreamRequest request =
+ CreateDeliveryStreamRequest.builder()
+ .deliveryStreamName(deliveryStreamName)
+ .extendedS3DestinationConfiguration(s3Config)
+ .deliveryStreamType(DeliveryStreamType.DIRECT_PUT)
+ .build();
+
+ CompletableFuture<CreateDeliveryStreamResponse> deliveryStream =
+ firehoseAsyncClient.createDeliveryStream(request);
+ deliveryStream.get();
+ }
+}
diff --git a/flink-connectors/flink-connector-aws-kinesis-data-streams/src/test/resources/log4j2-test.properties b/flink-connectors/flink-connector-aws-kinesis-firehose/src/test/resources/log4j2-test.properties
similarity index 97%
copy from flink-connectors/flink-connector-aws-kinesis-data-streams/src/test/resources/log4j2-test.properties
copy to flink-connectors/flink-connector-aws-kinesis-firehose/src/test/resources/log4j2-test.properties
index c5339e7..c4fa187 100644
--- a/flink-connectors/flink-connector-aws-kinesis-data-streams/src/test/resources/log4j2-test.properties
+++ b/flink-connectors/flink-connector-aws-kinesis-firehose/src/test/resources/log4j2-test.properties
@@ -18,7 +18,7 @@
# Set root logger level to OFF to not flood build logs
# set manually to INFO for debugging purposes
-rootLogger.level = INFO
+rootLogger.level = OFF
rootLogger.appenderRef.test.ref = TestLogger
appender.testlogger.name = TestLogger
diff --git a/flink-connectors/flink-connector-base/pom.xml b/flink-connectors/flink-connector-base/pom.xml
index 751aecd..58fd369 100644
--- a/flink-connectors/flink-connector-base/pom.xml
+++ b/flink-connectors/flink-connector-base/pom.xml
@@ -72,4 +72,20 @@
<scope>test</scope>
</dependency>
</dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-jar-plugin</artifactId>
+ <executions>
+ <execution>
+ <goals>
+ <goal>test-jar</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
</project>
diff --git a/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriterTest.java b/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriterTest.java
index b508159..00a08b1 100644
--- a/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriterTest.java
+++ b/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriterTest.java
@@ -17,22 +17,8 @@
package org.apache.flink.connector.base.sink.writer;
-import org.apache.flink.api.common.operators.MailboxExecutor;
import org.apache.flink.api.connector.sink.Sink;
-import org.apache.flink.metrics.Counter;
-import org.apache.flink.metrics.Gauge;
-import org.apache.flink.metrics.groups.OperatorIOMetricGroup;
-import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
-import org.apache.flink.metrics.testutils.MetricListener;
-import org.apache.flink.runtime.metrics.groups.InternalSinkWriterMetricGroup;
-import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
-import org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor;
import org.apache.flink.streaming.runtime.tasks.TestProcessingTimeService;
-import org.apache.flink.streaming.runtime.tasks.mailbox.MailboxExecutorImpl;
-import org.apache.flink.streaming.runtime.tasks.mailbox.TaskMailboxImpl;
-import org.apache.flink.util.UserCodeClassLoader;
-import org.apache.flink.util.function.RunnableWithException;
-import org.apache.flink.util.function.ThrowingRunnable;
import org.junit.Before;
import org.junit.Test;
@@ -42,10 +28,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
-import java.util.Optional;
-import java.util.OptionalLong;
import java.util.Set;
-import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -66,12 +49,12 @@ import static org.junit.jupiter.api.Assertions.fail;
public class AsyncSinkWriterTest {
private final List<Integer> res = new ArrayList<>();
- private SinkInitContext sinkInitContext;
+ private TestSinkInitContext sinkInitContext;
@Before
public void before() {
res.clear();
- sinkInitContext = new SinkInitContext();
+ sinkInitContext = new TestSinkInitContext();
}
private void performNormalWriteOfEightyRecordsToMock()
@@ -1008,105 +991,6 @@ public class AsyncSinkWriterTest {
}
}
- private static class SinkInitContext implements Sink.InitContext {
-
- private static final TestProcessingTimeService processingTimeService;
- private final MetricListener metricListener = new MetricListener();
- private final OperatorIOMetricGroup operatorIOMetricGroup =
- UnregisteredMetricGroups.createUnregisteredOperatorMetricGroup().getIOMetricGroup();
- private final SinkWriterMetricGroup metricGroup =
- InternalSinkWriterMetricGroup.mock(
- metricListener.getMetricGroup(), operatorIOMetricGroup);
-
- static {
- processingTimeService = new TestProcessingTimeService();
- }
-
- @Override
- public UserCodeClassLoader getUserCodeClassLoader() {
- return null;
- }
-
- @Override
- public MailboxExecutor getMailboxExecutor() {
- StreamTaskActionExecutor streamTaskActionExecutor =
- new StreamTaskActionExecutor() {
- @Override
- public void run(RunnableWithException e) throws Exception {
- e.run();
- }
-
- @Override
- public <E extends Throwable> void runThrowing(
- ThrowingRunnable<E> throwingRunnable) throws E {
- throwingRunnable.run();
- }
-
- @Override
- public <R> R call(Callable<R> callable) throws Exception {
- return callable.call();
- }
- };
- return new MailboxExecutorImpl(
- new TaskMailboxImpl(Thread.currentThread()),
- Integer.MAX_VALUE,
- streamTaskActionExecutor);
- }
-
- @Override
- public Sink.ProcessingTimeService getProcessingTimeService() {
- return new Sink.ProcessingTimeService() {
- @Override
- public long getCurrentProcessingTime() {
- return processingTimeService.getCurrentProcessingTime();
- }
-
- @Override
- public void registerProcessingTimer(
- long time, ProcessingTimeCallback processingTimerCallback) {
- processingTimeService.registerTimer(
- time, processingTimerCallback::onProcessingTime);
- }
- };
- }
-
- @Override
- public int getSubtaskId() {
- return 0;
- }
-
- @Override
- public int getNumberOfParallelSubtasks() {
- return 0;
- }
-
- @Override
- public SinkWriterMetricGroup metricGroup() {
- return metricGroup;
- }
-
- @Override
- public OptionalLong getRestoredCheckpointId() {
- return OptionalLong.empty();
- }
-
- public TestProcessingTimeService getTestProcessingTimeService() {
- return processingTimeService;
- }
-
- private Optional<Gauge<Long>> getCurrentSendTimeGauge() {
- return metricListener.getGauge("currentSendTime");
- }
-
- private Counter getNumRecordsOutCounter() {
- return metricGroup.getIOMetricGroup().getNumRecordsOutCounter();
- }
-
- private Counter getNumBytesOutCounter() {
- return metricGroup.getIOMetricGroup().getNumBytesOutCounter();
- }
- }
-
/**
* This SinkWriter releases the lock on existing threads blocked by {@code delayedStartLatch}
* and blocks itself until {@code blockedThreadLatch} is unblocked.
diff --git a/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/sink/writer/TestSinkInitContext.java b/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/sink/writer/TestSinkInitContext.java
new file mode 100644
index 0000000..fba2711
--- /dev/null
+++ b/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/sink/writer/TestSinkInitContext.java
@@ -0,0 +1,139 @@
+/*
+ * 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.base.sink.writer;
+
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.api.connector.sink.Sink;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Gauge;
+import org.apache.flink.metrics.groups.OperatorIOMetricGroup;
+import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
+import org.apache.flink.metrics.testutils.MetricListener;
+import org.apache.flink.runtime.metrics.groups.InternalSinkWriterMetricGroup;
+import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor;
+import org.apache.flink.streaming.runtime.tasks.TestProcessingTimeService;
+import org.apache.flink.streaming.runtime.tasks.mailbox.MailboxExecutorImpl;
+import org.apache.flink.streaming.runtime.tasks.mailbox.TaskMailboxImpl;
+import org.apache.flink.util.UserCodeClassLoader;
+import org.apache.flink.util.function.RunnableWithException;
+import org.apache.flink.util.function.ThrowingRunnable;
+
+import java.util.Optional;
+import java.util.OptionalLong;
+import java.util.concurrent.Callable;
+
+/** A mock implementation of a {@code Sink.InitContext} to be used in sink unit tests. */
+public class TestSinkInitContext implements Sink.InitContext {
+
+ private static final TestProcessingTimeService processingTimeService;
+ private final MetricListener metricListener = new MetricListener();
+ private final OperatorIOMetricGroup operatorIOMetricGroup =
+ UnregisteredMetricGroups.createUnregisteredOperatorMetricGroup().getIOMetricGroup();
+ private final SinkWriterMetricGroup metricGroup =
+ InternalSinkWriterMetricGroup.mock(
+ metricListener.getMetricGroup(), operatorIOMetricGroup);
+
+ static {
+ processingTimeService = new TestProcessingTimeService();
+ }
+
+ @Override
+ public UserCodeClassLoader getUserCodeClassLoader() {
+ return null;
+ }
+
+ @Override
+ public MailboxExecutor getMailboxExecutor() {
+ StreamTaskActionExecutor streamTaskActionExecutor =
+ new StreamTaskActionExecutor() {
+ @Override
+ public void run(RunnableWithException e) throws Exception {
+ e.run();
+ }
+
+ @Override
+ public <E extends Throwable> void runThrowing(
+ ThrowingRunnable<E> throwingRunnable) throws E {
+ throwingRunnable.run();
+ }
+
+ @Override
+ public <R> R call(Callable<R> callable) throws Exception {
+ return callable.call();
+ }
+ };
+ return new MailboxExecutorImpl(
+ new TaskMailboxImpl(Thread.currentThread()),
+ Integer.MAX_VALUE,
+ streamTaskActionExecutor);
+ }
+
+ @Override
+ public Sink.ProcessingTimeService getProcessingTimeService() {
+ return new Sink.ProcessingTimeService() {
+ @Override
+ public long getCurrentProcessingTime() {
+ return processingTimeService.getCurrentProcessingTime();
+ }
+
+ @Override
+ public void registerProcessingTimer(
+ long time, ProcessingTimeCallback processingTimerCallback) {
+ processingTimeService.registerTimer(
+ time, processingTimerCallback::onProcessingTime);
+ }
+ };
+ }
+
+ @Override
+ public int getSubtaskId() {
+ return 0;
+ }
+
+ @Override
+ public int getNumberOfParallelSubtasks() {
+ return 0;
+ }
+
+ @Override
+ public SinkWriterMetricGroup metricGroup() {
+ return metricGroup;
+ }
+
+ @Override
+ public OptionalLong getRestoredCheckpointId() {
+ return OptionalLong.empty();
+ }
+
+ public TestProcessingTimeService getTestProcessingTimeService() {
+ return processingTimeService;
+ }
+
+ public Optional<Gauge<Long>> getCurrentSendTimeGauge() {
+ return metricListener.getGauge("currentSendTime");
+ }
+
+ public Counter getNumRecordsOutCounter() {
+ return metricGroup.getIOMetricGroup().getNumRecordsOutCounter();
+ }
+
+ public Counter getNumBytesOutCounter() {
+ return metricGroup.getIOMetricGroup().getNumBytesOutCounter();
+ }
+}
diff --git a/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxyV2.java b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxyV2.java
index e2e6a79..0677765 100644
--- a/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxyV2.java
+++ b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxyV2.java
@@ -18,8 +18,8 @@
package org.apache.flink.streaming.connectors.kinesis.proxy;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.connector.kinesis.util.AWSKinesisDataStreamsUtil;
import org.apache.flink.streaming.connectors.kinesis.internals.publisher.fanout.FanOutRecordPublisherConfiguration;
+import org.apache.flink.streaming.connectors.kinesis.util.AwsV2Util;
import org.apache.flink.util.Preconditions;
import org.slf4j.Logger;
@@ -190,7 +190,7 @@ public class KinesisProxyV2 implements KinesisProxyV2Interface {
try {
response = responseSupplier.get();
} catch (Exception ex) {
- if (AWSKinesisDataStreamsUtil.isRecoverableException(ex)) {
+ if (AwsV2Util.isRecoverableException(ex)) {
long backoffMillis =
backoff.calculateFullJitterBackoff(
jitterBase, jitterMax, jitterExponent, ++attempt);
diff --git a/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxyV2Factory.java b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxyV2Factory.java
index 03767cd..62f25db 100644
--- a/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxyV2Factory.java
+++ b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxyV2Factory.java
@@ -18,9 +18,9 @@
package org.apache.flink.streaming.connectors.kinesis.proxy;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.aws.util.AWSAsyncSinkUtil;
import org.apache.flink.connector.aws.util.AWSGeneralUtil;
-import org.apache.flink.connector.kinesis.config.AWSKinesisDataStreamsConfigConstants;
-import org.apache.flink.connector.kinesis.util.AWSKinesisDataStreamsUtil;
+import org.apache.flink.connector.kinesis.sink.KinesisDataStreamsConfigConstants;
import org.apache.flink.streaming.connectors.kinesis.internals.publisher.fanout.FanOutRecordPublisherConfiguration;
import org.apache.flink.streaming.connectors.kinesis.util.AwsV2Util;
import org.apache.flink.util.Preconditions;
@@ -63,13 +63,17 @@ public class KinesisProxyV2Factory {
Properties legacyConfigProps = new Properties(configProps);
legacyConfigProps.setProperty(
- AWSKinesisDataStreamsConfigConstants.KINESIS_CLIENT_USER_AGENT_PREFIX,
- AWSKinesisDataStreamsUtil.formatFlinkUserAgentPrefix(
- AWSKinesisDataStreamsConfigConstants
- .BASE_KINESIS_USER_AGENT_PREFIX_FORMAT));
+ KinesisDataStreamsConfigConstants.KINESIS_CLIENT_USER_AGENT_PREFIX,
+ AWSAsyncSinkUtil.formatFlinkUserAgentPrefix(
+ KinesisDataStreamsConfigConstants.BASE_KINESIS_USER_AGENT_PREFIX_FORMAT));
final KinesisAsyncClient client =
- AWSKinesisDataStreamsUtil.createKinesisAsyncClient(legacyConfigProps, httpClient);
+ AWSAsyncSinkUtil.createAwsAsyncClient(
+ legacyConfigProps,
+ httpClient,
+ KinesisAsyncClient.builder(),
+ KinesisDataStreamsConfigConstants.BASE_KINESIS_USER_AGENT_PREFIX_FORMAT,
+ KinesisDataStreamsConfigConstants.KINESIS_CLIENT_USER_AGENT_PREFIX);
return new KinesisProxyV2(client, httpClient, configuration, BACKOFF);
}
diff --git a/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/util/AWSUtil.java b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/util/AWSUtil.java
index 5c949fe..684234c 100644
--- a/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/util/AWSUtil.java
+++ b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/util/AWSUtil.java
@@ -19,8 +19,8 @@ package org.apache.flink.streaming.connectors.kinesis.util;
import org.apache.flink.annotation.Internal;
import org.apache.flink.connector.aws.config.AWSConfigConstants.CredentialProvider;
-import org.apache.flink.connector.kinesis.config.AWSKinesisDataStreamsConfigConstants;
-import org.apache.flink.connector.kinesis.util.AWSKinesisDataStreamsUtil;
+import org.apache.flink.connector.aws.util.AWSAsyncSinkUtil;
+import org.apache.flink.connector.kinesis.sink.KinesisDataStreamsConfigConstants;
import org.apache.flink.streaming.connectors.kinesis.config.AWSConfigConstants;
import org.apache.flink.streaming.connectors.kinesis.model.SentinelSequenceNumber;
import org.apache.flink.streaming.connectors.kinesis.model.SequenceNumber;
@@ -83,9 +83,8 @@ public class AWSUtil {
Properties configProps, ClientConfiguration awsClientConfig) {
// set a Flink-specific user agent
awsClientConfig.setUserAgentPrefix(
- AWSKinesisDataStreamsUtil.formatFlinkUserAgentPrefix(
- AWSKinesisDataStreamsConfigConstants
- .BASE_KINESIS_USER_AGENT_PREFIX_FORMAT));
+ AWSAsyncSinkUtil.formatFlinkUserAgentPrefix(
+ KinesisDataStreamsConfigConstants.BASE_KINESIS_USER_AGENT_PREFIX_FORMAT));
// utilize automatic refreshment of credentials by directly passing the
// AWSCredentialsProvider
@@ -133,7 +132,7 @@ public class AWSUtil {
private static AWSCredentialsProvider getCredentialsProvider(
final Properties configProps, final String configPrefix) {
CredentialProvider credentialProviderType =
- AWSKinesisDataStreamsUtil.getCredentialProviderType(configProps, configPrefix);
+ AWSAsyncSinkUtil.getCredentialProviderType(configProps, configPrefix);
switch (credentialProviderType) {
case ENV_VAR:
diff --git a/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/util/AwsV2Util.java b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/util/AwsV2Util.java
index 609e879..aa62a65 100644
--- a/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/util/AwsV2Util.java
+++ b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/util/AwsV2Util.java
@@ -20,6 +20,8 @@ package org.apache.flink.streaming.connectors.kinesis.util;
import org.apache.flink.annotation.Internal;
import software.amazon.awssdk.http.SdkHttpConfigurationOption;
+import software.amazon.awssdk.services.kinesis.model.LimitExceededException;
+import software.amazon.awssdk.services.kinesis.model.ProvisionedThroughputExceededException;
import software.amazon.awssdk.utils.AttributeMap;
import java.time.Duration;
@@ -70,4 +72,10 @@ public class AwsV2Util {
public static boolean isNoneEfoRegistrationType(final Properties properties) {
return NONE.name().equals(properties.get(EFO_REGISTRATION_TYPE));
}
+
+ public static boolean isRecoverableException(Exception e) {
+ Throwable cause = e.getCause();
+ return cause instanceof LimitExceededException
+ || cause instanceof ProvisionedThroughputExceededException;
+ }
}
diff --git a/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/util/AwsV2UtilTest.java b/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/util/AwsV2UtilTest.java
index 47cf23f..ea9f03d 100644
--- a/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/util/AwsV2UtilTest.java
+++ b/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/util/AwsV2UtilTest.java
@@ -19,10 +19,12 @@ package org.apache.flink.streaming.connectors.kinesis.util;
import org.junit.Test;
import software.amazon.awssdk.http.SdkHttpConfigurationOption;
+import software.amazon.awssdk.services.kinesis.model.LimitExceededException;
import software.amazon.awssdk.utils.AttributeMap;
import java.time.Duration;
import java.util.Properties;
+import java.util.concurrent.ExecutionException;
import static org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants.EFORegistrationType.EAGER;
import static org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants.EFORegistrationType.LAZY;
@@ -131,4 +133,36 @@ public class AwsV2UtilTest {
prop.setProperty(EFO_REGISTRATION_TYPE, NONE.name());
assertTrue(AwsV2Util.isNoneEfoRegistrationType(prop));
}
+
+ @Test
+ public void testIsRecoverableExceptionForRecoverable() {
+ Exception recoverable = LimitExceededException.builder().build();
+ assertTrue(AwsV2Util.isRecoverableException(new ExecutionException(recoverable)));
+ }
+
+ @Test
+ public void testIsRecoverableExceptionForNonRecoverable() {
+ Exception nonRecoverable = new IllegalArgumentException("abc");
+ assertFalse(AwsV2Util.isRecoverableException(new ExecutionException(nonRecoverable)));
+ }
+
+ @Test
+ public void testIsRecoverableExceptionForRuntimeExceptionWrappingRecoverable() {
+ Exception recoverable = LimitExceededException.builder().build();
+ Exception runtime = new RuntimeException("abc", recoverable);
+ assertTrue(AwsV2Util.isRecoverableException(runtime));
+ }
+
+ @Test
+ public void testIsRecoverableExceptionForRuntimeExceptionWrappingNonRecoverable() {
+ Exception nonRecoverable = new IllegalArgumentException("abc");
+ Exception runtime = new RuntimeException("abc", nonRecoverable);
+ assertFalse(AwsV2Util.isRecoverableException(runtime));
+ }
+
+ @Test
+ public void testIsRecoverableExceptionForNullCause() {
+ Exception nonRecoverable = new IllegalArgumentException("abc");
+ assertFalse(AwsV2Util.isRecoverableException(nonRecoverable));
+ }
}
diff --git a/flink-connectors/pom.xml b/flink-connectors/pom.xml
index 7495412..a4c97b7 100644
--- a/flink-connectors/pom.xml
+++ b/flink-connectors/pom.xml
@@ -55,6 +55,7 @@ under the License.
<module>flink-connector-aws-base</module>
<module>flink-connector-kinesis</module>
<module>flink-connector-aws-kinesis-data-streams</module>
+ <module>flink-connector-aws-kinesis-firehose</module>
<module>flink-connector-base</module>
<module>flink-file-sink-common</module>
<module>flink-connector-files</module>
diff --git a/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/util/DockerImageVersions.java b/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/util/DockerImageVersions.java
index afc6d9b..99c5293 100644
--- a/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/util/DockerImageVersions.java
+++ b/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/util/DockerImageVersions.java
@@ -37,6 +37,8 @@ public class DockerImageVersions {
public static final String KINESALITE = "instructure/kinesalite:latest";
+ public static final String LOCALSTACK = "localstack/localstack:latest";
+
public static final String PULSAR = "apachepulsar/pulsar:2.8.0";
public static final String CASSANDRA_3 = "cassandra:3.0";
diff --git a/tools/ci/stage.sh b/tools/ci/stage.sh
index 7793c0b..d70c9ff 100755
--- a/tools/ci/stage.sh
+++ b/tools/ci/stage.sh
@@ -109,6 +109,7 @@ flink-connectors/flink-connector-rabbitmq,\
flink-connectors/flink-connector-twitter,\
flink-connectors/flink-connector-kinesis,\
flink-connectors/flink-connector-aws-kinesis-data-streams,\
+flink-connectors/flink-connector-aws-kinesis-firehose,\
flink-metrics/flink-metrics-dropwizard,\
flink-metrics/flink-metrics-graphite,\
flink-metrics/flink-metrics-jmx,\