You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2018/06/07 20:35:02 UTC
[GitHub] rdhabalia closed pull request #1904: Introduce kinesis sink on
function
rdhabalia closed pull request #1904: Introduce kinesis sink on function
URL: https://github.com/apache/incubator-pulsar/pull/1904
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/FunctionsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/FunctionsBase.java
index a00a2dd408..bacade64b6 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/FunctionsBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/FunctionsBase.java
@@ -80,7 +80,7 @@ public Response registerFunction(final @PathParam("tenant") String tenant,
return functions.registerFunction(
tenant, namespace, functionName, uploadedInputStream, fileDetail, functionPkgUrl, functionDetailsJson);
}
-
+
@PUT
@ApiOperation(value = "Updates a Pulsar Function currently running in cluster mode")
@ApiResponses(value = {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarSinkE2ETest.java b/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarSinkE2ETest.java
new file mode 100644
index 0000000000..5089e95b87
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarSinkE2ETest.java
@@ -0,0 +1,293 @@
+/**
+ * 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.pulsar.io;
+
+import static org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest.retryStrategically;
+import static org.mockito.Mockito.spy;
+
+import java.io.File;
+import java.lang.reflect.Method;
+import java.net.InetAddress;
+import java.net.MalformedURLException;
+import java.net.URI;
+import java.net.URL;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.bookkeeper.test.PortManager;
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.ServiceConfigurationUtils;
+import org.apache.pulsar.broker.loadbalance.impl.SimpleLoadManagerImpl;
+import org.apache.pulsar.client.admin.BrokerStats;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.api.Authentication;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerBuilder;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.common.policies.data.ClusterData;
+import org.apache.pulsar.common.policies.data.SubscriptionStats;
+import org.apache.pulsar.common.policies.data.TenantInfo;
+import org.apache.pulsar.functions.api.utils.DefaultSerDe;
+import org.apache.pulsar.functions.api.utils.IdentityFunction;
+import org.apache.pulsar.functions.proto.Function;
+import org.apache.pulsar.functions.proto.Function.FunctionDetails;
+import org.apache.pulsar.functions.proto.Function.SinkSpec;
+import org.apache.pulsar.functions.proto.Function.SourceSpec;
+import org.apache.pulsar.functions.sink.PulsarSink;
+import org.apache.pulsar.functions.utils.Reflections;
+import org.apache.pulsar.functions.worker.Utils;
+import org.apache.pulsar.functions.worker.WorkerConfig;
+import org.apache.pulsar.functions.worker.WorkerService;
+import org.apache.pulsar.functions.worker.rest.WorkerServer;
+import org.apache.pulsar.zookeeper.LocalBookkeeperEnsemble;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import com.google.gson.Gson;
+
+import io.netty.util.concurrent.DefaultThreadFactory;
+import jersey.repackaged.com.google.common.collect.Lists;
+
+/**
+ * Test Pulsar sink on function
+ *
+ */
+public class PulsarSinkE2ETest {
+ LocalBookkeeperEnsemble bkEnsemble;
+
+ ServiceConfiguration config;
+ WorkerConfig workerConfig;
+ URL url;
+ PulsarService pulsar;
+ PulsarAdmin admin;
+ PulsarClient pulsarClient;
+ BrokerStats brokerStatsClient;
+ WorkerServer functionsWorkerServer;
+ WorkerService functionsWorkerService;
+ final String tenant = "external-repl-prop";
+ String pulsarFunctionsNamespace = tenant + "/use/pulsar-function-admin";
+ String primaryHost;
+ ExecutorService executor;
+ ExecutorService workerExecutor;
+
+ private final int ZOOKEEPER_PORT = PortManager.nextFreePort();
+ private final int brokerWebServicePort = PortManager.nextFreePort();
+ private final int brokerServicePort = PortManager.nextFreePort();
+ private final int workerServicePort = PortManager.nextFreePort();
+ private static final Logger log = LoggerFactory.getLogger(PulsarSinkE2ETest.class);
+
+ @BeforeMethod
+ void setup(Method method) throws Exception {
+
+ log.info("--- Setting up method {} ---", method.getName());
+
+ executor = new ThreadPoolExecutor(5, 20, 30, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
+ workerExecutor = Executors.newScheduledThreadPool(1, new DefaultThreadFactory("pulsar-worker-test"));
+
+ // Start local bookkeeper ensemble
+ bkEnsemble = new LocalBookkeeperEnsemble(3, ZOOKEEPER_PORT, PortManager.nextFreePort());
+ bkEnsemble.start();
+
+ String hostHttpUrl = "http://127.0.0.1" + ":";
+
+ config = spy(new ServiceConfiguration());
+ config.setClusterName("use");
+ config.setWebServicePort(brokerWebServicePort);
+ config.setZookeeperServers("127.0.0.1" + ":" + ZOOKEEPER_PORT);
+ config.setBrokerServicePort(brokerServicePort);
+ config.setLoadManagerClassName(SimpleLoadManagerImpl.class.getName());
+ functionsWorkerService = createPulsarFunctionWorker(config);
+ url = new URL(hostHttpUrl + brokerWebServicePort);
+ boolean isFunctionWebServerRequired = method.getName()
+ .equals("testExternalReplicatorRedirectionToWorkerService");
+ Optional<WorkerService> functionWorkerService = isFunctionWebServerRequired ? Optional.ofNullable(null)
+ : Optional.of(functionsWorkerService);
+ pulsar = new PulsarService(config, functionWorkerService);
+ pulsar.start();
+ admin = new PulsarAdmin(url, (Authentication) null);
+ brokerStatsClient = admin.brokerStats();
+ primaryHost = String.format("http://%s:%d", InetAddress.getLocalHost().getHostName(), brokerWebServicePort);
+
+ // update cluster metadata
+ ClusterData clusterData = new ClusterData(url.toString());
+ admin.clusters().updateCluster(config.getClusterName(), clusterData);
+
+ pulsarClient = PulsarClient.builder().serviceUrl(url.toString()).statsInterval(0, TimeUnit.SECONDS).build();
+
+ TenantInfo propAdmin = new TenantInfo();
+ propAdmin.setAllowedClusters(Sets.newHashSet(Lists.newArrayList("use")));
+ admin.tenants().updateTenant(tenant, propAdmin);
+
+ if (isFunctionWebServerRequired) {
+ URI dlogURI = Utils.initializeDlogNamespace(config.getZookeeperServers(), "/ledgers");
+ functionsWorkerService.start(dlogURI);
+ functionsWorkerServer = new WorkerServer(functionsWorkerService);
+ workerExecutor.submit(functionsWorkerServer);
+ }
+ Thread.sleep(100);
+ }
+
+ @AfterMethod
+ void shutdown() throws Exception {
+ log.info("--- Shutting down ---");
+ if (executor != null) {
+ executor.shutdown();
+ }
+ if (workerExecutor != null) {
+ workerExecutor.shutdown();
+ }
+ admin.close();
+ pulsar.close();
+ functionsWorkerService.stop();
+ bkEnsemble.stop();
+ }
+
+ private WorkerService createPulsarFunctionWorker(ServiceConfiguration config) {
+ workerConfig = new WorkerConfig();
+ workerConfig.setPulsarFunctionsNamespace(pulsarFunctionsNamespace);
+ workerConfig.setSchedulerClassName(
+ org.apache.pulsar.functions.worker.scheduler.RoundRobinScheduler.class.getName());
+ workerConfig.setThreadContainerFactory(new WorkerConfig.ThreadContainerFactory().setThreadGroupName("use"));
+ // worker talks to local broker
+ workerConfig.setPulsarServiceUrl("pulsar://127.0.0.1:" + config.getBrokerServicePort());
+ workerConfig.setPulsarWebServiceUrl("http://127.0.0.1:" + config.getWebServicePort());
+ workerConfig.setFailureCheckFreqMs(100);
+ workerConfig.setNumFunctionPackageReplicas(1);
+ workerConfig.setClusterCoordinationTopicName("coordinate");
+ workerConfig.setFunctionAssignmentTopicName("assignment");
+ workerConfig.setFunctionMetadataTopicName("metadata");
+ workerConfig.setInstanceLivenessCheckFreqMs(100);
+ workerConfig.setWorkerPort(workerServicePort);
+ workerConfig.setPulsarFunctionsCluster(config.getClusterName());
+ String hostname = ServiceConfigurationUtils.getDefaultOrConfiguredAddress(config.getAdvertisedAddress());
+ workerConfig.setWorkerHostname(hostname);
+ workerConfig
+ .setWorkerId("c-" + config.getClusterName() + "-fw-" + hostname + "-" + workerConfig.getWorkerPort());
+ return new WorkerService(workerConfig);
+ }
+
+ /**
+ * Validates pulsar sink e2e functionality on functions.
+ *
+ * @throws Exception
+ */
+ @Test
+ public void testE2EPulsarSink() throws Exception {
+
+ final String namespacePortion = "myReplNs";
+ final String replNamespace = tenant + "/" + namespacePortion;
+ final String sourceTopic = "persistent://" + replNamespace + "/my-topic1";
+ admin.namespaces().createNamespace(replNamespace);
+ Set<String> clusters = Sets.newHashSet(Lists.newArrayList("use"));
+ admin.namespaces().setNamespaceReplicationClusters(replNamespace, clusters);
+
+ // create a producer that creates a topic at broker
+ ProducerBuilder<byte[]> producerBuilder = pulsarClient.newProducer().topic(sourceTopic);
+ Producer<byte[]> producer = producerBuilder.create();
+
+ String jarFilePathUrl = Utils.FILE + ":"
+ + PulsarSink.class.getProtectionDomain().getCodeSource().getLocation().getPath();
+ FunctionDetails functionDetails = createSinkConfig(jarFilePathUrl, tenant, namespacePortion, "PulsarSink-test");
+ admin.functions().createFunctionWithUrl(functionDetails, jarFilePathUrl);
+
+ retryStrategically((test) -> {
+ try {
+ return admin.topics().getStats(sourceTopic).subscriptions.size() == 1;
+ } catch (PulsarAdminException e) {
+ return false;
+ }
+ }, 5, 150);
+ // validate pulsar sink consumer has started on the topic
+ Assert.assertEquals(admin.topics().getStats(sourceTopic).subscriptions.size(), 1);
+
+ for (int i = 0; i < 5; i++) {
+ String message = "my-message-" + i;
+ producer.send(message.getBytes());
+ }
+ retryStrategically((test) -> {
+ try {
+ SubscriptionStats subStats = admin.topics().getStats(sourceTopic).subscriptions.values()
+ .iterator().next();
+ return subStats.unackedMessages == 0;
+ } catch (PulsarAdminException e) {
+ return false;
+ }
+ }, 5, 150);
+ // validate pulsar-sink consumer has consumed all messages and delivered to Pulsar sink but unacked messages
+ // due to publish failure
+ Assert.assertEquals(admin.topics().getStats(sourceTopic).subscriptions.values().iterator()
+ .next().unackedMessages, 0);
+
+ }
+
+ protected FunctionDetails createSinkConfig(String jarFile, String tenant, String namespace, String sinkName) {
+
+ File file = new File(jarFile);
+ try {
+ Reflections.loadJar(file);
+ } catch (MalformedURLException e) {
+ throw new RuntimeException("Failed to load user jar " + file, e);
+ }
+ String sourceTopicPattern = String.format("persistent://%s/%s/my.*", tenant, namespace);
+ Class<?> typeArg = byte[].class;
+
+ FunctionDetails.Builder functionDetailsBuilder = FunctionDetails.newBuilder();
+ functionDetailsBuilder.setTenant(tenant);
+ functionDetailsBuilder.setNamespace(namespace);
+ functionDetailsBuilder.setName(sinkName);
+ functionDetailsBuilder.setRuntime(FunctionDetails.Runtime.JAVA);
+ functionDetailsBuilder.setParallelism(1);
+ functionDetailsBuilder.setClassName(IdentityFunction.class.getName());
+
+ // set source spec
+ // source spec classname should be empty so that the default pulsar source will be used
+ SourceSpec.Builder sourceSpecBuilder = SourceSpec.newBuilder();
+ sourceSpecBuilder.setSubscriptionType(Function.SubscriptionType.SHARED);
+ sourceSpecBuilder.setTypeClassName(byte[].class.getName());
+ sourceSpecBuilder.setTopicsPattern(sourceTopicPattern);
+ sourceSpecBuilder.putTopicsToSerDeClassName(sourceTopicPattern, DefaultSerDe.class.getName());
+ functionDetailsBuilder.setAutoAck(true);
+ functionDetailsBuilder.setSource(sourceSpecBuilder);
+
+ // set up sink spec
+ SinkSpec.Builder sinkSpecBuilder = SinkSpec.newBuilder();
+ //sinkSpecBuilder.setClassName(PulsarSink.class.getName());
+ sinkSpecBuilder.setTopic(String.format("persistent://%s/%s/%s", tenant, namespace, "output"));
+ Map<String, Object> sinkConfigMap = Maps.newHashMap();
+ sinkSpecBuilder.setConfigs(new Gson().toJson(sinkConfigMap));
+ sinkSpecBuilder.setTypeClassName(typeArg.getName());
+ functionDetailsBuilder.setSink(sinkSpecBuilder);
+
+ return functionDetailsBuilder.build();
+ }
+}
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/FunctionsImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/FunctionsImpl.java
index 1a1ab60a8a..68d3046250 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/FunctionsImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/FunctionsImpl.java
@@ -226,4 +226,4 @@ public static void mergeJson(String json, Builder builder) throws IOException {
public static String printJson(MessageOrBuilder msg) throws IOException {
return JsonFormat.printer().print(msg);
}
-}
+}
\ No newline at end of file
diff --git a/pulsar-functions/worker/pom.xml b/pulsar-functions/worker/pom.xml
index 4860ae7826..518b18eacf 100644
--- a/pulsar-functions/worker/pom.xml
+++ b/pulsar-functions/worker/pom.xml
@@ -102,7 +102,7 @@
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
</dependency>
-
+
</dependencies>
</project>
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionActioner.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionActioner.java
index 70a1ce4859..a03828b6f0 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionActioner.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionActioner.java
@@ -232,4 +232,4 @@ private String getDownloadPackagePath(FunctionMetaData functionMetaData, int ins
},
File.separatorChar);
}
-}
+}
\ No newline at end of file
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java
index 161d210573..535a7dd332 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java
@@ -69,7 +69,7 @@
@Getter
@EqualsAndHashCode
@ToString
- static class ThreadContainerFactory {
+ public static class ThreadContainerFactory {
private String threadGroupName;
}
private ThreadContainerFactory threadContainerFactory;
@@ -79,7 +79,7 @@
@Getter
@EqualsAndHashCode
@ToString
- static class ProcessContainerFactory {
+ public static class ProcessContainerFactory {
private String javaInstanceJarLocation;
private String pythonInstanceLocation;
private String logDirectory;
diff --git a/pulsar-io/kinesis/pom.xml b/pulsar-io/kinesis/pom.xml
new file mode 100644
index 0000000000..7f9f488e29
--- /dev/null
+++ b/pulsar-io/kinesis/pom.xml
@@ -0,0 +1,125 @@
+<!--
+
+ 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-io</artifactId>
+ <version>2.1.0-incubating-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>pulsar-io-kinesis</artifactId>
+ <name>Pulsar IO :: Kinesis</name>
+
+ <dependencies>
+
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>pulsar-io-core</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>pulsar-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ <version>${jackson.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>com.fasterxml.jackson.dataformat</groupId>
+ <artifactId>jackson-dataformat-yaml</artifactId>
+ <version>${jackson.version}</version>
+ </dependency>
+
+ <!-- kinesis dependencies -->
+ <dependency>
+ <groupId>com.amazonaws</groupId>
+ <artifactId>aws-java-sdk-core</artifactId>
+ <version>1.11.46</version>
+ </dependency>
+
+ <dependency>
+ <groupId>com.amazonaws</groupId>
+ <artifactId>amazon-kinesis-client</artifactId>
+ <version>1.7.2</version>
+ </dependency>
+ <dependency>
+ <groupId>com.amazonaws</groupId>
+ <artifactId>amazon-kinesis-producer</artifactId>
+ <version>0.12.8</version>
+ </dependency>
+ <dependency>
+ <groupId>com.google.protobuf</groupId>
+ <artifactId>protobuf-java</artifactId>
+ <version>2.6.1</version>
+ </dependency>
+ <!-- /kinesis dependencies -->
+
+ </dependencies>
+
+ <build>
+ <plugins>
+ <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>com.amazonaws*:*</include>
+ <include>com.google.protobuf*:*</include>
+ <include>${project.groupId}:*</include>
+ <include>com.fasterxml.jackson*:*</include>
+ </includes>
+ </artifactSet>
+ <relocations>
+ <relocation>
+ <pattern>com.google.protobuf</pattern>
+ <shadedPattern>org.apache.pulsar.replicator.com.google.protobuf</shadedPattern>
+ </relocation>
+ <relocation>
+ <pattern>com.amazonaws</pattern>
+ <shadedPattern>org.apache.pulsar.com.amazonaws</shadedPattern>
+ </relocation>
+ </relocations>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+
+</project>
diff --git a/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/AwsCredentialProviderPlugin.java b/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/AwsCredentialProviderPlugin.java
new file mode 100644
index 0000000000..8c616cc677
--- /dev/null
+++ b/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/AwsCredentialProviderPlugin.java
@@ -0,0 +1,60 @@
+/**
+ * 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.pulsar.io.kinesis;
+
+/**
+ * Kinesis source/sink calls credential-provider while refreshing aws accessKey and secreKey. So, implementation
+ * AwsCredentialProviderPlugin needs to makes sure to return non-expired keys when it requires.
+ *
+ */
+public interface AwsCredentialProviderPlugin {
+
+ /**
+ * accepts aws-account related param and initialize credential provider.
+ *
+ * @param param
+ */
+ void init(String param);
+
+ /**
+ * Returns the AWS access key ID for this credentials object.
+ *
+ * @return The AWS access key ID for this credentials object.
+ */
+ String getAWSAccessKeyId();
+
+ /**
+ * Returns the AWS secret access key for this credentials object.
+ *
+ * @return The AWS secret access key for this credentials object.
+ */
+ String getAWSSecretKey();
+
+ /**
+ * Forces this credentials provider to refresh its credentials. For many
+ * implementations of credentials provider, this method may simply be a
+ * no-op, such as any credentials provider implementation that vends
+ * static/non-changing credentials. For other implementations that vend
+ * different credentials through out their lifetime, this method should
+ * force the credentials provider to refresh its credentials.
+ */
+ void refresh();
+
+}
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
new file mode 100644
index 0000000000..39af1f300a
--- /dev/null
+++ b/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/KinesisSink.java
@@ -0,0 +1,250 @@
+/**
+ * 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.pulsar.io.kinesis;
+
+import static com.google.common.base.Preconditions.checkArgument;
+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 java.io.IOException;
+import java.lang.reflect.Constructor;
+import java.nio.ByteBuffer;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.pulsar.io.core.RecordContext;
+import org.apache.pulsar.io.core.Sink;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.amazonaws.auth.AWSCredentials;
+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.UserRecordResult;
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.ListenableFuture;
+import org.apache.commons.lang3.builder.ReflectionToStringBuilder;
+import org.apache.commons.lang3.builder.ToStringStyle;
+
+import io.netty.util.Recycler;
+import io.netty.util.Recycler.Handle;
+
+/**
+ * A Kinesis sink
+ */
+public class KinesisSink implements Sink<byte[]> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(KinesisSink.class);
+
+ private KinesisProducer kinesisProducer;
+ private KinesisSinkConfig kinesisSinkConfig;
+ private String streamName;
+
+ public static final String ACCESS_KEY_NAME = "accessKey";
+ public static final String SECRET_KEY_NAME = "secretKey";
+
+ @Override
+ public void write(RecordContext inputRecordContext, byte[] value) throws Exception {
+ final String partitionedKey = inputRecordContext.getPartitionId();
+ ListenableFuture<UserRecordResult> addRecordResult = kinesisProducer.addUserRecord(this.streamName,
+ partitionedKey, ByteBuffer.wrap(value));
+ addCallback(addRecordResult,
+ ProducerSendCallback.create(this.streamName, inputRecordContext, System.nanoTime()), directExecutor());
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (kinesisProducer != null) {
+ kinesisProducer.flush();
+ kinesisProducer.destroy();
+ }
+ LOG.info("Kinesis sink stopped.");
+ }
+
+ @Override
+ public void open(Map<String, Object> config) throws Exception {
+ kinesisSinkConfig = KinesisSinkConfig.load(config);
+
+ checkArgument(isNotBlank(kinesisSinkConfig.getAwsKinesisStreamName()), "empty kinesis-stream name");
+ checkArgument(isNotBlank(kinesisSinkConfig.getAwsEndpoint()), "empty aws-end-point");
+ checkArgument(isNotBlank(kinesisSinkConfig.getAwsRegion()), "empty aws region name");
+ checkArgument(isNotBlank(kinesisSinkConfig.getAwsKinesisStreamName()), "empty kinesis stream name");
+ checkArgument(isNotBlank(kinesisSinkConfig.getAwsCredentialPluginParam()), "empty aws-credential param");
+
+ KinesisProducerConfiguration kinesisConfig = new KinesisProducerConfiguration();
+ kinesisConfig.setKinesisEndpoint(kinesisSinkConfig.getAwsEndpoint());
+ kinesisConfig.setRegion(kinesisSinkConfig.getAwsRegion());
+ kinesisConfig.setThreadingModel(ThreadingModel.POOLED);
+ kinesisConfig.setThreadPoolSize(4);
+ kinesisConfig.setCollectionMaxCount(1);
+ AWSCredentialsProvider credentialsProvider = createCredentialProvider(
+ kinesisSinkConfig.getAwsCredentialPluginName(), kinesisSinkConfig.getAwsCredentialPluginParam());
+ kinesisConfig.setCredentialsProvider(credentialsProvider);
+
+ this.streamName = kinesisSinkConfig.getAwsKinesisStreamName();
+ this.kinesisProducer = new KinesisProducer(kinesisConfig);
+
+ LOG.info("Kinesis sink started. {}", (ReflectionToStringBuilder.toString(kinesisConfig, ToStringStyle.SHORT_PREFIX_STYLE)));
+ }
+
+ private AWSCredentialsProvider createCredentialProvider(String awsCredentialPluginName,
+ String awsCredentialPluginParam) {
+ if (isNotBlank(awsCredentialPluginName)) {
+ return createCredentialProvider(awsCredentialPluginName, awsCredentialPluginParam);
+ } else {
+ return defaultCredentialProvider(awsCredentialPluginParam);
+ }
+ }
+
+ private static final class ProducerSendCallback implements FutureCallback<UserRecordResult> {
+
+ private RecordContext resultContext;
+ private String streamName;
+ private long startTime = 0;
+ private final Handle<ProducerSendCallback> recyclerHandle;
+
+ private ProducerSendCallback(Handle<ProducerSendCallback> recyclerHandle) {
+ this.recyclerHandle = recyclerHandle;
+ }
+
+ static ProducerSendCallback create(String streamName, RecordContext result, long startTime) {
+ ProducerSendCallback sendCallback = RECYCLER.get();
+ sendCallback.resultContext = result;
+ sendCallback.streamName = streamName;
+ sendCallback.startTime = startTime;
+ return sendCallback;
+ }
+
+ private void recycle() {
+ resultContext = null;
+ streamName = null;
+ startTime = 0;
+ recyclerHandle.recycle(this);
+ }
+
+ private static final Recycler<ProducerSendCallback> RECYCLER = new Recycler<ProducerSendCallback>() {
+ @Override
+ protected ProducerSendCallback newObject(Handle<ProducerSendCallback> handle) {
+ return new ProducerSendCallback(handle);
+ }
+ };
+
+ @Override
+ public void onSuccess(UserRecordResult result) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Successfully published message for replicator of {}-{} with latency", this.streamName,
+ result.getShardId(), TimeUnit.NANOSECONDS.toMillis((System.nanoTime() - startTime)));
+ }
+ this.resultContext.ack();
+ recycle();
+ }
+
+ @Override
+ public void onFailure(Throwable exception) {
+ LOG.error("[{}] Failed to published message for replicator of {}-{} ", streamName,
+ resultContext.getPartitionId(), resultContext.getRecordSequence());
+ this.resultContext.fail();
+ recycle();
+ }
+ }
+
+ public static AWSCredentialsProvider createCredentialProviderPlugin(String pluginFQClassName, String param)
+ throws IllegalArgumentException {
+ try {
+ Class<?> clazz = Class.forName(pluginFQClassName);
+ Constructor<?> ctor = clazz.getConstructor();
+ final AwsCredentialProviderPlugin plugin = (AwsCredentialProviderPlugin) ctor.newInstance(new Object[] {});
+ plugin.init(param);
+ return new AWSCredentialsProvider() {
+ @Override
+ public AWSCredentials getCredentials() {
+ return new AWSCredentials() {
+ @Override
+ public String getAWSAccessKeyId() {
+ return plugin.getAWSAccessKeyId();
+ }
+
+ @Override
+ public String getAWSSecretKey() {
+ return plugin.getAWSSecretKey();
+ }
+ };
+ }
+ @Override
+ public void refresh() {
+ plugin.refresh();
+ }
+ };
+ } catch (Exception e) {
+ LOG.error("Failed to initialize AwsCredentialProviderPlugin {}", pluginFQClassName, e);
+ throw new IllegalArgumentException(
+ String.format("invalid authplugin name %s , failed to init %s", pluginFQClassName, e.getMessage()));
+ }
+ }
+
+ private AWSCredentialsProvider defaultCredentialProvider(String awsCredentialPluginParam) {
+ String[] credentials = awsCredentialPluginParam.split(",");
+ String accessKey = null;
+ String secretKey = null;
+ if (credentials.length == 2) {
+ for (String credential : credentials) {
+ String[] keys = credential.split("=");
+ if (keys.length == 2) {
+ if (keys[0].equals(ACCESS_KEY_NAME)) {
+ accessKey = keys[1];
+ } else if (keys[0].equals(SECRET_KEY_NAME)) {
+ secretKey = keys[1];
+ }
+ }
+ }
+ }
+ checkArgument(isNotBlank(accessKey) && isNotBlank(secretKey),
+ String.format("access-key/secret-key not present in param: format: %s=<access-key>,%s=<secret-key>",
+ ACCESS_KEY_NAME, SECRET_KEY_NAME));
+ return defaultCredentialProvider(accessKey, secretKey);
+ }
+
+ private AWSCredentialsProvider defaultCredentialProvider(String accessKey, String secretKey) {
+ return new AWSCredentialsProvider() {
+ @Override
+ public AWSCredentials getCredentials() {
+ return new AWSCredentials() {
+ @Override
+ public String getAWSAccessKeyId() {
+ return accessKey;
+ }
+
+ @Override
+ public String getAWSSecretKey() {
+ return secretKey;
+ }
+ };
+ }
+ @Override
+ public void refresh() {
+ // no-op
+ }
+ };
+ }
+
+}
\ No newline at end of file
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
new file mode 100644
index 0000000000..b7dbad441b
--- /dev/null
+++ b/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/KinesisSinkConfig.java
@@ -0,0 +1,57 @@
+/**
+ * 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.pulsar.io.kinesis;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
+import lombok.*;
+import lombok.experimental.Accessors;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Map;
+
+@Data
+@Setter
+@Getter
+@EqualsAndHashCode
+@ToString
+@Accessors(chain = true)
+public class KinesisSinkConfig implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private String awsEndpoint;
+ private String awsRegion;
+ private String awsKinesisStreamName;
+ private String awsCredentialPluginName;
+ private String awsCredentialPluginParam;
+
+ public static KinesisSinkConfig load(String yamlFile) throws IOException {
+ ObjectMapper mapper = new ObjectMapper(new YAMLFactory());
+ return mapper.readValue(new File(yamlFile), KinesisSinkConfig.class);
+ }
+
+ public static KinesisSinkConfig load(Map<String, Object> map) throws IOException {
+ ObjectMapper mapper = new ObjectMapper();
+ return mapper.readValue(new ObjectMapper().writeValueAsString(map), KinesisSinkConfig.class);
+ }
+}
\ No newline at end of file
diff --git a/pulsar-io/pom.xml b/pulsar-io/pom.xml
index fb63146b9c..20915ae90d 100644
--- a/pulsar-io/pom.xml
+++ b/pulsar-io/pom.xml
@@ -38,6 +38,7 @@
<module>aerospike</module>
<module>kafka</module>
<module>rabbitmq</module>
+ <module>kinesis</module>
</modules>
</project>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services