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/05/27 15:24:10 UTC
(pulsar) branch branch-3.0 updated: [fix][io][branch-3.0] Backport Kinesis Sink custom native executable support #23762 (#24349)
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
The following commit(s) were added to refs/heads/branch-3.0 by this push:
new b41ca2d8786 [fix][io][branch-3.0] Backport Kinesis Sink custom native executable support #23762 (#24349)
b41ca2d8786 is described below
commit b41ca2d87869de9c0a61080d76c24543345df19e
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Tue May 27 18:24:03 2025 +0300
[fix][io][branch-3.0] Backport Kinesis Sink custom native executable support #23762 (#24349)
---
pulsar-io/kinesis/pom.xml | 2 +-
.../org/apache/pulsar/io/kinesis/KinesisSink.java | 8 ++++-
.../pulsar/io/kinesis/KinesisSinkConfig.java | 24 ++++++++++++++
.../apache/pulsar/io/kinesis/KinesisSinkTest.java | 37 ++++++++++++----------
.../integration/io/sinks/KinesisSinkTester.java | 13 +++++---
5 files changed, 60 insertions(+), 24 deletions(-)
diff --git a/pulsar-io/kinesis/pom.xml b/pulsar-io/kinesis/pom.xml
index 95ec817ec60..b2066c65551 100644
--- a/pulsar-io/kinesis/pom.xml
+++ b/pulsar-io/kinesis/pom.xml
@@ -32,7 +32,7 @@
<properties>
<amazon-kinesis-client.version>2.2.8</amazon-kinesis-client.version>
- <amazon-kinesis-producer.version>0.14.0</amazon-kinesis-producer.version>
+ <amazon-kinesis-producer.version>0.15.12</amazon-kinesis-producer.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>
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 34a360c723a..da6dc80c232 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
@@ -90,7 +90,6 @@ import org.slf4j.LoggerFactory;
configClass = KinesisSinkConfig.class
)
public class KinesisSink extends AbstractAwsConnector implements Sink<GenericObject> {
-
private static final Logger LOG = LoggerFactory.getLogger(KinesisSink.class);
private KinesisProducer kinesisProducer;
@@ -169,6 +168,12 @@ public class KinesisSink extends AbstractAwsConnector implements Sink<GenericObj
if (kinesisSinkConfig.getAwsEndpointPort() != null) {
kinesisConfig.setKinesisPort(kinesisSinkConfig.getAwsEndpointPort());
}
+ if (kinesisSinkConfig.getAwsStsEndpoint() != null) {
+ kinesisConfig.setStsEndpoint(kinesisSinkConfig.getAwsStsEndpoint());
+ }
+ if (kinesisSinkConfig.getAwsStsPort() != null) {
+ kinesisConfig.setStsPort(kinesisSinkConfig.getAwsStsPort());
+ }
kinesisConfig.setRegion(kinesisSinkConfig.getAwsRegion());
kinesisConfig.setThreadingModel(ThreadingModel.POOLED);
kinesisConfig.setThreadPoolSize(4);
@@ -182,6 +187,7 @@ public class KinesisSink extends AbstractAwsConnector implements Sink<GenericObj
kinesisSinkConfig.getAwsCredentialPluginParam())
.getCredentialProvider();
kinesisConfig.setCredentialsProvider(credentialsProvider);
+ kinesisConfig.setNativeExecutable(StringUtils.trimToEmpty(kinesisSinkConfig.getNativeExecutable()));
kinesisConfig.setAggregationEnabled(kinesisSinkConfig.isAggregationEnabled());
this.streamName = kinesisSinkConfig.getAwsKinesisStreamName();
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 eb08496dbcd..4cb1871906d 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
@@ -108,6 +108,16 @@ public class KinesisSinkConfig extends BaseKinesisConfig implements Serializable
return mapper.readValue(new File(yamlFile), KinesisSinkConfig.class);
}
+ @FieldDoc(
+ required = false,
+ defaultValue = "",
+ help = "Path to the native Amazon Kinesis Producer Library (KPL) binary.\n"
+ + "Only use this setting if you want to use a custom build of the native code.\n"
+ + "This setting can also be set with the environment variable `PULSAR_IO_KINESIS_KPL_PATH`.\n"
+ + "If not set, the Kinesis sink will use the built-in native executable."
+ )
+ private String nativeExecutable = System.getenv("PULSAR_IO_KINESIS_KPL_PATH");
+
public enum MessageFormat {
/**
* Kinesis sink directly publishes pulsar-payload as a message into the kinesis-stream.
@@ -151,4 +161,18 @@ public class KinesisSinkConfig extends BaseKinesisConfig implements Serializable
help = "Enable aggregation. With aggregation, multiple user records could be packed into a single\n"
+ " KinesisRecord. If disabled, each user record is sent in its own KinesisRecord.")
private boolean aggregationEnabled = true;
+
+ @FieldDoc(
+ required = false,
+ defaultValue = "",
+ help = "Custom AWS STS endpoint"
+ )
+ private String awsStsEndpoint = "";
+
+ @FieldDoc(
+ required = false,
+ defaultValue = "",
+ help = "Custom AWS STS port to connect to"
+ )
+ private Integer awsStsPort;
}
diff --git a/pulsar-io/kinesis/src/test/java/org/apache/pulsar/io/kinesis/KinesisSinkTest.java b/pulsar-io/kinesis/src/test/java/org/apache/pulsar/io/kinesis/KinesisSinkTest.java
index 8a5d5be0ca4..c682776a14d 100644
--- a/pulsar-io/kinesis/src/test/java/org/apache/pulsar/io/kinesis/KinesisSinkTest.java
+++ b/pulsar-io/kinesis/src/test/java/org/apache/pulsar/io/kinesis/KinesisSinkTest.java
@@ -18,6 +18,16 @@
*/
package org.apache.pulsar.io.kinesis;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import lombok.SneakyThrows;
import org.apache.pulsar.client.api.Message;
@@ -42,24 +52,14 @@ import software.amazon.awssdk.services.kinesis.model.GetShardIteratorRequest;
import software.amazon.awssdk.services.kinesis.model.ListShardsRequest;
import software.amazon.awssdk.services.kinesis.model.ShardIteratorType;
-import java.net.URI;
-import java.nio.charset.StandardCharsets;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Optional;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-import static org.testng.Assert.assertEquals;
-import static org.testng.Assert.assertTrue;
-
public class KinesisSinkTest {
public static final String STREAM_NAME = "my-stream-1";
- public static LocalStackContainer LOCALSTACK_CONTAINER = new LocalStackContainer(DockerImageName.parse("localstack/localstack:1.0.4"))
- .withServices(LocalStackContainer.Service.KINESIS);
+ public static LocalStackContainer LOCALSTACK_CONTAINER =
+ new LocalStackContainer(DockerImageName.parse("localstack/localstack:4.0.3"))
+ .withServices(LocalStackContainer.Service.KINESIS, LocalStackContainer.Service.STS)
+ .withEnv("KINESIS_PROVIDER", "kinesalite");
@BeforeClass(alwaysRun = true)
public void beforeClass() throws Exception {
@@ -123,10 +123,13 @@ public class KinesisSinkTest {
}
private Map<String, Object> createConfig() {
- final URI endpointOverride = LOCALSTACK_CONTAINER.getEndpointOverride(LocalStackContainer.Service.KINESIS);
+ final URI kinesisEndpointOverride = LOCALSTACK_CONTAINER.getEndpointOverride(LocalStackContainer.Service.KINESIS);
Map<String, Object> map = new HashMap<>();
- map.put("awsEndpoint", endpointOverride.getHost());
- map.put("awsEndpointPort", endpointOverride.getPort());
+ map.put("awsEndpoint", kinesisEndpointOverride.getHost());
+ map.put("awsEndpointPort", kinesisEndpointOverride.getPort());
+ final URI stsEndpointOverride = LOCALSTACK_CONTAINER.getEndpointOverride(LocalStackContainer.Service.STS);
+ map.put("awsStsEndpoint", stsEndpointOverride.getHost());
+ map.put("awsStsPort", stsEndpointOverride.getPort());
map.put("skipCertificateValidation", true);
map.put("awsKinesisStreamName", STREAM_NAME);
map.put("awsRegion", "us-east-1");
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/KinesisSinkTester.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/KinesisSinkTester.java
index 6ca9c7b1083..83cb0088cd7 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/KinesisSinkTester.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/KinesisSinkTester.java
@@ -79,6 +79,11 @@ public class KinesisSinkTester extends SinkTester<LocalStackContainer> {
sinkConfig.put("awsKinesisStreamName", STREAM_NAME);
sinkConfig.put("awsRegion", "us-east-1");
sinkConfig.put("awsCredentialPluginParam", "{\"accessKey\":\"access\",\"secretKey\":\"secret\"}");
+ sinkConfig.put("awsEndpoint", NAME);
+ sinkConfig.put("awsEndpointPort", LOCALSTACK_SERVICE_PORT);
+ sinkConfig.put("awsStsEndpoint", NAME);
+ sinkConfig.put("awsStsPort", LOCALSTACK_SERVICE_PORT);
+ sinkConfig.put("skipCertificateValidation", true);
if (withSchema) {
sinkConfig.put("messageFormat", "FULL_MESSAGE_IN_JSON_EXPAND_VALUE");
}
@@ -100,9 +105,6 @@ public class KinesisSinkTester extends SinkTester<LocalStackContainer> {
public void prepareSink() throws Exception {
final LocalStackContainer localStackContainer = getServiceContainer();
final URI endpointOverride = localStackContainer.getEndpointOverride(LocalStackContainer.Service.KINESIS);
- sinkConfig.put("awsEndpoint", NAME);
- sinkConfig.put("awsEndpointPort", LOCALSTACK_SERVICE_PORT);
- sinkConfig.put("skipCertificateValidation", true);
client = KinesisAsyncClient.builder().credentialsProvider(() -> AwsBasicCredentials.create(
"access",
"secret"))
@@ -128,8 +130,9 @@ public class KinesisSinkTester extends SinkTester<LocalStackContainer> {
@Override
protected LocalStackContainer createSinkService(PulsarCluster cluster) {
- return new LocalStackContainer(DockerImageName.parse("localstack/localstack:1.0.4"))
- .withServices(LocalStackContainer.Service.KINESIS);
+ return new LocalStackContainer(DockerImageName.parse("localstack/localstack:4.0.3"))
+ .withServices(LocalStackContainer.Service.KINESIS, LocalStackContainer.Service.STS)
+ .withEnv("KINESIS_PROVIDER", "kinesalite");
}
@Override