You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2025/08/26 09:13:14 UTC
(pulsar) 02/02: [improve][io] Upgrade AWS SDK v1 & v2, Kinesis KPL and KPC versions (#24661)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 667569125fd643412753bc32b2d04d67c5351c6b
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Tue Aug 26 07:44:05 2025 +0300
[improve][io] Upgrade AWS SDK v1 & v2, Kinesis KPL and KPC versions (#24661)
(cherry picked from commit dc2ab26675bdbe11eabdbf86ff01f928bb820ebe)
---
pom.xml | 11 +-
pulsar-io/aws/pom.xml | 1 -
pulsar-io/kinesis-kpl-shaded/pom.xml | 149 +++++++++++++++++++++
pulsar-io/kinesis/pom.xml | 23 ++--
.../org/apache/pulsar/io/kinesis/KinesisSink.java | 15 ++-
.../pulsar/io/kinesis/KinesisSinkConfig.java | 1 +
.../pulsar/io/kinesis/KinesisSinkConfigTests.java | 2 +-
pulsar-io/pom.xml | 2 +
tests/integration/pom.xml | 2 +-
9 files changed, 186 insertions(+), 20 deletions(-)
diff --git a/pom.xml b/pom.xml
index 12698abea0f..1750822692c 100644
--- a/pom.xml
+++ b/pom.xml
@@ -185,7 +185,8 @@ flexible messaging model and an intuitive client API.</description>
<aerospike-client.version>4.4.20</aerospike-client.version>
<kafka-client.version>3.9.1</kafka-client.version>
<rabbitmq-client.version>5.18.0</rabbitmq-client.version>
- <aws-sdk.version>1.12.638</aws-sdk.version>
+ <aws-sdk.version>1.12.788</aws-sdk.version>
+ <aws-sdk2.version>2.32.28</aws-sdk2.version>
<avro.version>1.11.4</avro.version>
<joda.version>2.10.10</joda.version>
<jclouds.version>2.6.0</jclouds.version>
@@ -1292,6 +1293,14 @@ flexible messaging model and an intuitive client API.</description>
<scope>import</scope>
</dependency>
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>bom</artifactId>
+ <version>${aws-sdk2.version}</version>
+ <type>pom</type>
+ <scope>import</scope>
+ </dependency>
+
<dependency>
<groupId>org.apache.distributedlog</groupId>
<artifactId>distributedlog-core</artifactId>
diff --git a/pulsar-io/aws/pom.xml b/pulsar-io/aws/pom.xml
index 422ee793d51..c2228171b7f 100644
--- a/pulsar-io/aws/pom.xml
+++ b/pulsar-io/aws/pom.xml
@@ -51,7 +51,6 @@
<dependency>
<groupId>software.amazon.awssdk</groupId>
<artifactId>sts</artifactId>
- <version>2.10.56</version>
</dependency>
<!-- /aws dependencies -->
diff --git a/pulsar-io/kinesis-kpl-shaded/pom.xml b/pulsar-io/kinesis-kpl-shaded/pom.xml
new file mode 100644
index 00000000000..204f17bc04e
--- /dev/null
+++ b/pulsar-io/kinesis-kpl-shaded/pom.xml
@@ -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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.pulsar</groupId>
+ <artifactId>pulsar</artifactId>
+ <version>3.0.14-SNAPSHOT</version>
+ <relativePath>../../pom.xml</relativePath>
+ </parent>
+
+ <artifactId>pulsar-io-kinesis-kpl-shaded</artifactId>
+ <name>Pulsar IO :: Kinesis KPL Shaded</name>
+
+ <!--
+ There's a need to shade the protobuf-java dependency in the AWS Kinesis Producer Library (KPL)
+ since it conflicts at runtime with Pulsar Functions GRPC protobuf-java version.
+ -->
+
+ <properties>
+ <amazon-kinesis-producer.version>1.0.4</amazon-kinesis-producer.version>
+ <protobuf-java.version>4.29.0</protobuf-java.version>
+ </properties>
+
+ <dependencyManagement>
+ <dependencies>
+ <!-- enforce protobuf-java version that is compatible with KPL 1.0.x -->
+ <!-- see https://github.com/awslabs/amazon-kinesis-producer/blob/v1.0.4/java/amazon-kinesis-producer/pom.xml#L20 -->
+ <dependency>
+ <groupId>com.google.protobuf</groupId>
+ <artifactId>protobuf-bom</artifactId>
+ <version>${protobuf-java.version}</version>
+ <type>pom</type>
+ <scope>import</scope>
+ </dependency>
+ </dependencies>
+ </dependencyManagement>
+ <dependencies>
+ <dependency>
+ <groupId>software.amazon.kinesis</groupId>
+ <artifactId>amazon-kinesis-producer</artifactId>
+ <version>${amazon-kinesis-producer.version}</version>
+ </dependency>
+ </dependencies>
+ <build>
+ <finalName>${project.artifactId}-${project.version}</finalName>
+ <plugins>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>build-helper-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>attach-shade-jar</id>
+ <phase>package</phase>
+ <goals>
+ <goal>attach-artifact</goal>
+ </goals>
+ <configuration>
+ <artifacts>
+ <artifact>
+ <file>${project.build.directory}/${project.build.finalName}.jar</file>
+ <type>jar</type>
+ <classifier>shaded</classifier>
+ </artifact>
+ </artifacts>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <executions>
+ <execution>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <createDependencyReducedPom>true</createDependencyReducedPom>
+ <promoteTransitiveDependencies>true</promoteTransitiveDependencies>
+ <minimizeJar>false</minimizeJar>
+ <artifactSet>
+ <includes>
+ <include>software.amazon.kinesis:amazon-kinesis-producer</include>
+ <include>com.google.protobuf:protobuf-java</include>
+ </includes>
+ <excludes>
+ <!-- This is required for execute shade multiple times without clean -->
+ <exclude>org.apache.pulsar:pulsar-io-kinesis-kpl-shaded</exclude>
+ </excludes>
+ </artifactSet>
+ <relocations>
+ <relocation>
+ <pattern>com.google.protobuf</pattern>
+ <shadedPattern>org.apache.pulsar.io.kinesis.shaded.com.google.protobuf</shadedPattern>
+ </relocation>
+ </relocations>
+ <filters>
+ <filter>
+ <artifact>*:*</artifact>
+ <excludes>
+ <exclude>META-INF/*.SF</exclude>
+ <exclude>META-INF/*.DSA</exclude>
+ <exclude>META-INF/*.RSA</exclude>
+ <exclude>META-INF/maven/${project.groupId}/${project.artifactId}/pom.xml</exclude>
+ </excludes>
+ </filter>
+ </filters>
+ <transformers>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
+ <manifestEntries>
+ <Multi-Release>true</Multi-Release>
+ </manifestEntries>
+ </transformer>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.PluginXmlResourceTransformer"/>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.IncludeResourceTransformer">
+ <resource>META-INF/maven/${project.groupId}/${project.artifactId}/pom.xml</resource>
+ <file>${project.basedir}/dependency-reduced-pom.xml</file>
+ </transformer>
+ </transformers>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>
\ No newline at end of file
diff --git a/pulsar-io/kinesis/pom.xml b/pulsar-io/kinesis/pom.xml
index ffe60c8c0ac..5cef477ae28 100644
--- a/pulsar-io/kinesis/pom.xml
+++ b/pulsar-io/kinesis/pom.xml
@@ -19,7 +19,7 @@
-->
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
- xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.apache.pulsar</groupId>
@@ -31,8 +31,7 @@
<name>Pulsar IO :: Kinesis</name>
<properties>
- <amazon-kinesis-client.version>2.2.8</amazon-kinesis-client.version>
- <amazon-kinesis-producer.version>0.15.12</amazon-kinesis-producer.version>
+ <amazon-kinesis-client.version>3.1.2</amazon-kinesis-client.version>
<json-flattener.version>0.13.0</json-flattener.version>
<flatbuffers-java.version>1.9.0</flatbuffers-java.version>
<jaxb-api.version>2.3.0</jaxb-api.version>
@@ -50,6 +49,12 @@
</dependencyManagement>
<dependencies>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>pulsar-io-kinesis-kpl-shaded</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>pulsar-io-common</artifactId>
@@ -125,12 +130,7 @@
<version>${amazon-kinesis-client.version}</version>
</dependency>
- <dependency>
- <groupId>com.amazonaws</groupId>
- <artifactId>amazon-kinesis-producer</artifactId>
- <version>${amazon-kinesis-producer.version}</version>
- </dependency>
- <!-- /kinesis dependencies -->
+ <!-- /kinesis dependencies -->
<dependency>
<groupId>com.google.flatbuffers</groupId>
@@ -170,6 +170,11 @@
<plugin>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-nar-maven-plugin</artifactId>
+ <configuration>
+ <excludes>
+ <exclude>**/amazon-kinesis-producer-*.jar</exclude>
+ </excludes>
+ </configuration>
</plugin>
</plugins>
</build>
diff --git a/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/KinesisSink.java b/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/KinesisSink.java
index da6dc80c232..2476ff39d4a 100644
--- a/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/KinesisSink.java
+++ b/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/KinesisSink.java
@@ -23,11 +23,6 @@ import static com.google.common.util.concurrent.Futures.addCallback;
import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
import static org.apache.commons.lang3.StringUtils.isNotBlank;
import com.amazonaws.auth.AWSCredentialsProvider;
-import com.amazonaws.services.kinesis.producer.KinesisProducer;
-import com.amazonaws.services.kinesis.producer.KinesisProducerConfiguration;
-import com.amazonaws.services.kinesis.producer.KinesisProducerConfiguration.ThreadingModel;
-import com.amazonaws.services.kinesis.producer.UserRecordFailedException;
-import com.amazonaws.services.kinesis.producer.UserRecordResult;
import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -57,6 +52,12 @@ import org.apache.pulsar.io.core.annotations.IOType;
import org.apache.pulsar.io.kinesis.KinesisSinkConfig.MessageFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
+import software.amazon.kinesis.producer.KinesisProducer;
+import software.amazon.kinesis.producer.KinesisProducerConfiguration;
+import software.amazon.kinesis.producer.KinesisProducerConfiguration.ThreadingModel;
+import software.amazon.kinesis.producer.UserRecordFailedException;
+import software.amazon.kinesis.producer.UserRecordResult;
/**
* A Kinesis sink which can be configured by {@link KinesisSinkConfig}.
@@ -182,10 +183,10 @@ public class KinesisSink extends AbstractAwsConnector implements Sink<GenericObj
&& kinesisSinkConfig.getSkipCertificateValidation()) {
kinesisConfig.setVerifyCertificate(false);
}
- AWSCredentialsProvider credentialsProvider = createCredentialProvider(
+ AwsCredentialsProvider credentialsProvider = createCredentialProvider(
kinesisSinkConfig.getAwsCredentialPluginName(),
kinesisSinkConfig.getAwsCredentialPluginParam())
- .getCredentialProvider();
+ .getV2CredentialsProvider();
kinesisConfig.setCredentialsProvider(credentialsProvider);
kinesisConfig.setNativeExecutable(StringUtils.trimToEmpty(kinesisSinkConfig.getNativeExecutable()));
kinesisConfig.setAggregationEnabled(kinesisSinkConfig.isAggregationEnabled());
diff --git a/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/KinesisSinkConfig.java b/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/KinesisSinkConfig.java
index 4cb1871906d..cc4ef17a872 100644
--- a/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/KinesisSinkConfig.java
+++ b/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/KinesisSinkConfig.java
@@ -26,6 +26,7 @@ import java.io.Serializable;
import lombok.Data;
import lombok.EqualsAndHashCode;
import org.apache.pulsar.io.core.annotations.FieldDoc;
+import software.amazon.kinesis.producer.KinesisProducerConfiguration;
@Data
@EqualsAndHashCode(callSuper = true)
diff --git a/pulsar-io/kinesis/src/test/java/org/apache/pulsar/io/kinesis/KinesisSinkConfigTests.java b/pulsar-io/kinesis/src/test/java/org/apache/pulsar/io/kinesis/KinesisSinkConfigTests.java
index 6f76d9e69a2..4426bea12b7 100644
--- a/pulsar-io/kinesis/src/test/java/org/apache/pulsar/io/kinesis/KinesisSinkConfigTests.java
+++ b/pulsar-io/kinesis/src/test/java/org/apache/pulsar/io/kinesis/KinesisSinkConfigTests.java
@@ -20,7 +20,6 @@ package org.apache.pulsar.io.kinesis;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;
-
import java.io.File;
import java.io.IOException;
import java.util.HashMap;
@@ -31,6 +30,7 @@ import org.apache.pulsar.io.core.SinkContext;
import org.apache.pulsar.io.kinesis.KinesisSinkConfig.MessageFormat;
import org.mockito.Mockito;
import org.testng.annotations.Test;
+import software.amazon.kinesis.producer.KinesisProducerConfiguration;
public class KinesisSinkConfigTests {
diff --git a/pulsar-io/pom.xml b/pulsar-io/pom.xml
index 6cf7f97e761..99cdf9a38f7 100644
--- a/pulsar-io/pom.xml
+++ b/pulsar-io/pom.xml
@@ -55,6 +55,7 @@
<module>kafka</module>
<module>rabbitmq</module>
<module>kinesis</module>
+ <module>kinesis-kpl-shaded</module>
<module>hdfs3</module>
<module>jdbc</module>
<module>data-generator</module>
@@ -92,6 +93,7 @@
<module>http</module>
<module>rabbitmq</module>
<module>kinesis</module>
+ <module>kinesis-kpl-shaded</module>
<module>hdfs3</module>
<module>jdbc</module>
<module>data-generator</module>
diff --git a/tests/integration/pom.xml b/tests/integration/pom.xml
index b616e08af33..d096a62b6bb 100644
--- a/tests/integration/pom.xml
+++ b/tests/integration/pom.xml
@@ -211,7 +211,7 @@
<dependency>
<groupId>software.amazon.kinesis</groupId>
<artifactId>amazon-kinesis-client</artifactId>
- <version>2.2.8</version>
+ <version>3.1.2</version>
<scope>test</scope>
</dependency>
<dependency>