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/09/10 20:31:51 UTC
[GitHub] sijie closed pull request #2546: Elastic connector
sijie closed pull request #2546: Elastic connector
URL: https://github.com/apache/incubator-pulsar/pull/2546
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-io/elastic-search/pom.xml b/pulsar-io/elastic-search/pom.xml
new file mode 100644
index 0000000000..1339b2f311
--- /dev/null
+++ b/pulsar-io/elastic-search/pom.xml
@@ -0,0 +1,91 @@
+<!--
+
+ 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.2.0-incubating-SNAPSHOT</version>
+ </parent>
+ <artifactId>pulsar-io-elastic-search</artifactId>
+ <name>Pulsar IO :: ElasticSearch</name>
+
+ <repositories>
+ <repository>
+ <id>jcenter</id>
+ <url>https://jcenter.bintray.com/</url>
+ </repository>
+ </repositories>
+
+ <dependencies>
+
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>pulsar-io-core</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>com.fasterxml.jackson.dataformat</groupId>
+ <artifactId>jackson-dataformat-yaml</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-lang3</artifactId>
+ <version>3.4</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.elasticsearch.client</groupId>
+ <artifactId>elasticsearch-rest-high-level-client</artifactId>
+ <version>6.3.2</version>
+ </dependency>
+
+ <dependency>
+ <groupId>net.andreinc.mockneat</groupId>
+ <artifactId>mockneat</artifactId>
+ <version>0.2.2</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>com.google.code.gson</groupId>
+ <artifactId>gson</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-nar-maven-plugin</artifactId>
+ </plugin>
+ </plugins>
+ </build>
+
+</project>
\ No newline at end of file
diff --git a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchAbstractSink.java b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchAbstractSink.java
new file mode 100644
index 0000000000..3760d4072b
--- /dev/null
+++ b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchAbstractSink.java
@@ -0,0 +1,150 @@
+/**
+ * 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.elasticsearch;
+
+import java.io.IOException;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.util.Map;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.http.HttpHost;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.apache.http.client.CredentialsProvider;
+import org.apache.http.impl.client.BasicCredentialsProvider;
+import org.apache.pulsar.functions.api.Record;
+import org.apache.pulsar.io.core.KeyValue;
+import org.apache.pulsar.io.core.Sink;
+import org.apache.pulsar.io.core.SinkContext;
+import org.elasticsearch.action.DocWriteResponse;
+import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
+import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
+import org.elasticsearch.action.admin.indices.get.GetIndexRequest;
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.action.index.IndexResponse;
+import org.elasticsearch.client.Requests;
+import org.elasticsearch.client.RestClient;
+import org.elasticsearch.client.RestClientBuilder;
+import org.elasticsearch.client.RestHighLevelClient;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.XContentType;
+
+/**
+ * The base abstract class for ElasticSearch sinks.
+ * Users need to implement extractKeyValue function to use this sink.
+ * This class assumes that the input will be JSON documents
+ */
+public abstract class ElasticSearchAbstractSink<K, V> implements Sink<byte[]> {
+
+ protected static final String DOCUMENT = "doc";
+
+ private URL url;
+ private RestHighLevelClient client;
+ private CredentialsProvider credentialsProvider;
+ private ElasticSearchConfig elasticSearchConfig;
+
+ @Override
+ public void open(Map<String, Object> config, SinkContext sinkContext) throws Exception {
+ elasticSearchConfig = ElasticSearchConfig.load(config);
+ elasticSearchConfig.validate();
+ createIndexIfNeeded();
+ }
+
+ @Override
+ public void close() throws Exception {
+ client.close();
+ }
+
+ @Override
+ public void write(Record<byte[]> record) {
+ KeyValue<K, V> keyValue = extractKeyValue(record);
+ IndexRequest indexRequest = Requests.indexRequest(elasticSearchConfig.getIndexName());
+ indexRequest.type(DOCUMENT);
+ indexRequest.source(keyValue.getValue(), XContentType.JSON);
+
+ try {
+ IndexResponse indexResponse = getClient().index(indexRequest);
+ if (indexResponse.getResult().equals(DocWriteResponse.Result.CREATED)) {
+ record.ack();
+ } else {
+ record.fail();
+ }
+ } catch (final IOException ex) {
+ record.fail();
+ }
+ }
+
+ public abstract KeyValue<K, V> extractKeyValue(Record<byte[]> record);
+
+ private void createIndexIfNeeded() throws IOException {
+ GetIndexRequest request = new GetIndexRequest();
+ request.indices(elasticSearchConfig.getIndexName());
+ boolean exists = getClient().indices().exists(request);
+
+ if (!exists) {
+ CreateIndexRequest cireq = new CreateIndexRequest(elasticSearchConfig.getIndexName());
+
+ cireq.settings(Settings.builder()
+ .put("index.number_of_shards", elasticSearchConfig.getIndexNumberOfShards())
+ .put("index.number_of_replicas", elasticSearchConfig.getIndexNumberOfReplicas()));
+
+ CreateIndexResponse ciresp = getClient().indices().create(cireq);
+ if (!ciresp.isAcknowledged() || !ciresp.isShardsAcknowledged()) {
+ throw new RuntimeException("Unable to create index.");
+ }
+ }
+ }
+
+ private URL getUrl() throws MalformedURLException {
+ if (url == null) {
+ url = new URL(elasticSearchConfig.getElasticSearchUrl());
+ }
+ return url;
+ }
+
+ private CredentialsProvider getCredentialsProvider() {
+
+ if (StringUtils.isEmpty(elasticSearchConfig.getUsername())
+ || StringUtils.isEmpty(elasticSearchConfig.getPassword())) {
+ return null;
+ }
+
+ credentialsProvider = new BasicCredentialsProvider();
+ credentialsProvider.setCredentials(AuthScope.ANY,
+ new UsernamePasswordCredentials(elasticSearchConfig.getUsername(),
+ elasticSearchConfig.getPassword()));
+ return credentialsProvider;
+ }
+
+ private RestHighLevelClient getClient() throws MalformedURLException {
+ if (client == null) {
+ CredentialsProvider cp = getCredentialsProvider();
+ RestClientBuilder builder = RestClient.builder(new HttpHost(getUrl().getHost(),
+ getUrl().getPort(), getUrl().getProtocol()));
+
+ if (cp != null) {
+ builder.setHttpClientConfigCallback(httpClientBuilder ->
+ httpClientBuilder.setDefaultCredentialsProvider(cp));
+ }
+ client = new RestHighLevelClient(builder);
+ }
+ return client;
+ }
+}
diff --git a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchConfig.java b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchConfig.java
new file mode 100644
index 0000000000..8b54353c86
--- /dev/null
+++ b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchConfig.java
@@ -0,0 +1,88 @@
+/**
+ * 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.elasticsearch;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
+import java.io.File;
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Map;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.Setter;
+import lombok.ToString;
+import lombok.experimental.Accessors;
+import org.apache.commons.lang3.StringUtils;
+
+/**
+ * Configuration class for the ElasticSearch Sink Connector.
+ */
+@Data
+@Setter
+@Getter
+@EqualsAndHashCode
+@ToString
+@Accessors(chain = true)
+public class ElasticSearchConfig implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private String elasticSearchUrl;
+
+ private String indexName;
+
+ private int indexNumberOfShards = 1;
+
+ private int indexNumberOfReplicas = 1;
+
+ private String username;
+
+ private String password;
+
+ public static ElasticSearchConfig load(String yamlFile) throws IOException {
+ ObjectMapper mapper = new ObjectMapper(new YAMLFactory());
+ return mapper.readValue(new File(yamlFile), ElasticSearchConfig.class);
+ }
+
+ public static ElasticSearchConfig load(Map<String, Object> map) throws IOException {
+ ObjectMapper mapper = new ObjectMapper();
+ return mapper.readValue(new ObjectMapper().writeValueAsString(map), ElasticSearchConfig.class);
+ }
+
+ public void validate() {
+ if (StringUtils.isEmpty(elasticSearchUrl) || StringUtils.isEmpty(indexName)) {
+ throw new IllegalArgumentException("Required property not set.");
+ }
+
+ if ((StringUtils.isNotEmpty(username) && StringUtils.isEmpty(password))
+ || (StringUtils.isEmpty(username) && StringUtils.isNotEmpty(password))) {
+ throw new IllegalArgumentException("Values for both Username & password are required.");
+ }
+
+ if (indexNumberOfShards < 1) {
+ throw new IllegalArgumentException("indexNumberOfShards must be a positive integer");
+ }
+
+ if (indexNumberOfReplicas < 1) {
+ throw new IllegalArgumentException("indexNumberOfReplicas must be a positive integer");
+ }
+ }
+}
\ No newline at end of file
diff --git a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchStringSink.java b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchStringSink.java
new file mode 100644
index 0000000000..6cfa03d5c0
--- /dev/null
+++ b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchStringSink.java
@@ -0,0 +1,35 @@
+/**
+ * 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.elasticsearch;
+
+import org.apache.pulsar.functions.api.Record;
+import org.apache.pulsar.io.core.KeyValue;
+
+/**
+ * Concrete ElasticSearch sink.
+ * This class assumes that the input will be JSON documents
+ */
+public class ElasticSearchStringSink extends ElasticSearchAbstractSink<String, String> {
+
+ @Override
+ public KeyValue<String, String> extractKeyValue(Record<byte[]> record) {
+ String key = record.getKey().orElseGet(() -> new String(record.getValue()));
+ return new KeyValue<>(key, new String(record.getValue()));
+ }
+}
diff --git a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/package-info.java b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/package-info.java
new file mode 100644
index 0000000000..c4f5d6dccc
--- /dev/null
+++ b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/package-info.java
@@ -0,0 +1,19 @@
+/**
+ * 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.elasticsearch;
\ No newline at end of file
diff --git a/pulsar-io/elastic-search/src/main/resources/META-INF/services/pulsar-io.yaml b/pulsar-io/elastic-search/src/main/resources/META-INF/services/pulsar-io.yaml
new file mode 100644
index 0000000000..0307516cc8
--- /dev/null
+++ b/pulsar-io/elastic-search/src/main/resources/META-INF/services/pulsar-io.yaml
@@ -0,0 +1,22 @@
+#
+# 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.
+#
+
+name: Elastic Search
+description: Writes data into Elastic Search
+sinkClass: org.apache.pulsar.io.elasticsearch.ElasticSearchStringSink
diff --git a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchConfigTests.java b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchConfigTests.java
new file mode 100644
index 0000000000..65b6c225f6
--- /dev/null
+++ b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchConfigTests.java
@@ -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.
+ */
+package org.apache.pulsar.io.elasticsearch;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.testng.annotations.Test;
+
+public class ElasticSearchConfigTests {
+
+ @Test
+ public final void loadFromYamlFileTest() throws IOException {
+ File yamlFile = getFile("sinkConfig.yaml");
+ ElasticSearchConfig config = ElasticSearchConfig.load(yamlFile.getAbsolutePath());
+ assertNotNull(config);
+ assertEquals(config.getElasticSearchUrl(), "http://localhost:90902");
+ assertEquals(config.getIndexName(), "myIndex");
+ assertEquals(config.getUsername(), "scooby");
+ assertEquals(config.getPassword(), "doobie");
+ }
+
+ @Test
+ public final void loadFromMapTest() throws IOException {
+ Map<String, Object> map = new HashMap<String, Object> ();
+ map.put("elasticSearchUrl", "http://localhost:90902");
+ map.put("indexName", "myIndex");
+ map.put("username", "racerX");
+ map.put("password", "go-speedie-go");
+
+ ElasticSearchConfig config = ElasticSearchConfig.load(map);
+ assertNotNull(config);
+ assertEquals(config.getElasticSearchUrl(), "http://localhost:90902");
+ assertEquals(config.getIndexName(), "myIndex");
+ assertEquals(config.getUsername(), "racerX");
+ assertEquals(config.getPassword(), "go-speedie-go");
+ }
+
+ @Test
+ public final void validValidateTest() throws IOException {
+ Map<String, Object> map = new HashMap<String, Object> ();
+ map.put("elasticSearchUrl", "http://localhost:90902");
+ map.put("indexName", "myIndex");
+ map.put("username", "racerX");
+ map.put("password", "go-speedie-go");
+
+ ElasticSearchConfig config = ElasticSearchConfig.load(map);
+ assertNotNull(config);
+ config.validate();
+ }
+
+ @Test(expectedExceptions = IllegalArgumentException.class,
+ expectedExceptionsMessageRegExp = "Required property not set.")
+ public final void missingRequiredPropertiesTest() throws IOException {
+ Map<String, Object> map = new HashMap<String, Object> ();
+ map.put("elasticSearchUrl", "http://localhost:90902");
+
+ ElasticSearchConfig config = ElasticSearchConfig.load(map);
+ config.validate();
+ }
+
+ @Test(expectedExceptions = IllegalArgumentException.class,
+ expectedExceptionsMessageRegExp = "indexNumberOfShards must be a positive integer")
+ public final void invalidPropertyValueTest() throws IOException {
+ Map<String, Object> map = new HashMap<String, Object> ();
+ map.put("elasticSearchUrl", "http://localhost:90902");
+ map.put("indexName", "myIndex");
+ map.put("username", "racerX");
+ map.put("password", "go-speedie-go");
+ map.put("indexNumberOfShards", "-1");
+
+ ElasticSearchConfig config = ElasticSearchConfig.load(map);
+ config.validate();
+ }
+
+ @Test(expectedExceptions = IllegalArgumentException.class,
+ expectedExceptionsMessageRegExp = "Values for both Username & password are required.")
+ public final void userCredentialsTest() throws IOException {
+ Map<String, Object> map = new HashMap<String, Object> ();
+ map.put("elasticSearchUrl", "http://localhost:90902");
+ map.put("indexName", "myIndex");
+ map.put("username", "racerX");
+
+ ElasticSearchConfig config = ElasticSearchConfig.load(map);
+ config.validate();
+ }
+
+ @Test(expectedExceptions = IllegalArgumentException.class,
+ expectedExceptionsMessageRegExp = "Values for both Username & password are required.")
+ public final void passwordCredentialsTest() throws IOException {
+ Map<String, Object> map = new HashMap<String, Object> ();
+ map.put("elasticSearchUrl", "http://localhost:90902");
+ map.put("indexName", "myIndex");
+ map.put("password", "go-speedie-go");
+
+ ElasticSearchConfig config = ElasticSearchConfig.load(map);
+ config.validate();
+ }
+
+ private File getFile(String name) {
+ ClassLoader classLoader = getClass().getClassLoader();
+ return new File(classLoader.getResource(name).getFile());
+ }
+}
diff --git a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSinkTests.java b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSinkTests.java
new file mode 100644
index 0000000000..ea2b886e39
--- /dev/null
+++ b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSinkTests.java
@@ -0,0 +1,144 @@
+/**
+ * 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.elasticsearch;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+import org.apache.pulsar.functions.api.Record;
+import org.apache.pulsar.io.core.SinkContext;
+import org.apache.pulsar.io.elasticsearch.data.Profile;
+import org.apache.pulsar.io.elasticsearch.data.UserProfile;
+import org.elasticsearch.ElasticsearchStatusException;
+import org.mockito.Mock;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+
+import net.andreinc.mockneat.MockNeat;
+
+public class ElasticSearchSinkTests {
+
+ protected static MockNeat mockNeat;
+ protected static Gson gson;
+
+ @Mock
+ protected Record<byte[]> mockRecord;
+
+ @Mock
+ protected SinkContext mockSinkContext;
+ protected Map<String, Object> map;
+ protected ElasticSearchStringSink sink;
+
+ @BeforeClass
+ public static final void init() {
+ mockNeat = MockNeat.threadLocal();
+ gson = new GsonBuilder()
+ .setPrettyPrinting()
+ .create();
+ }
+
+ @SuppressWarnings("unchecked")
+ @BeforeMethod
+ public final void setUp() throws Exception {
+ map = new HashMap<String, Object> ();
+ map.put("elasticSearchUrl", "http://localhost:9200");
+ sink = new ElasticSearchStringSink();
+
+ mockRecord = mock(Record.class);
+ mockSinkContext = mock(SinkContext.class);
+
+ when(mockRecord.getKey()).thenAnswer(new Answer<Optional<String>>() {
+ long sequenceCounter = 0;
+ public Optional<String> answer(InvocationOnMock invocation) throws Throwable {
+ return Optional.of( "key-" + sequenceCounter++);
+ }});
+
+ when(mockRecord.getValue()).thenAnswer(new Answer<byte[]>() {
+ public byte[] answer(InvocationOnMock invocation) throws Throwable {
+ return getJSON().getBytes();
+ }});
+ }
+
+ @AfterMethod
+ public final void tearDown() throws Exception {
+ sink.close();
+ }
+
+ @Test(enabled = false, expectedExceptions = ElasticsearchStatusException.class)
+ public final void invalidIndexNameTest() throws Exception {
+ map.put("indexName", "myIndex");
+ sink.open(map, mockSinkContext);
+ }
+
+ @Test(enabled = false)
+ public final void createIndexTest() throws Exception {
+ map.put("indexName", "test-index");
+ sink.open(map, mockSinkContext);
+ }
+
+ @Test(enabled = false)
+ public final void singleRecordTest() throws Exception {
+ map.put("indexName", "test-index");
+ sink.open(map, mockSinkContext);
+ send(1);
+ verify(mockRecord, times(1)).ack();
+ }
+
+ @Test(enabled = false)
+ public final void send100Test() throws Exception {
+ map.put("indexName", "test-index");
+ sink.open(map, mockSinkContext);
+ send(100);
+ verify(mockRecord, times(100)).ack();
+ }
+
+ protected final void send(int numRecords) throws Exception {
+ for (int idx = 0; idx < numRecords; idx++) {
+ sink.write(mockRecord);
+ }
+ }
+
+ private static String getJSON() {
+ return mockNeat
+ .reflect(UserProfile.class)
+ .field("name", mockNeat.names().full())
+ .field("userName", mockNeat.users())
+ .field("email", mockNeat.emails())
+ .field("profiles",
+ mockNeat.reflect(Profile.class)
+ .field("profileId", mockNeat.ints().range(100, 1000))
+ .field("profileAdded", mockNeat.localDates().toUtilDate())
+ .list(2))
+ .map(gson::toJson)
+ .val();
+ }
+}
diff --git a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/data/Profile.java b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/data/Profile.java
new file mode 100644
index 0000000000..64797303fc
--- /dev/null
+++ b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/data/Profile.java
@@ -0,0 +1,33 @@
+/**
+ * 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.elasticsearch.data;
+
+import java.util.Date;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+public class Profile {
+ Integer profileId;
+ Date profileAdded;
+}
diff --git a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/data/UserProfile.java b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/data/UserProfile.java
new file mode 100644
index 0000000000..2b75a90e4c
--- /dev/null
+++ b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/data/UserProfile.java
@@ -0,0 +1,35 @@
+/**
+ * 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.elasticsearch.data;
+
+import java.util.List;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+public class UserProfile {
+ String name;
+ String userName;
+ String email;
+ List<Profile> profiles;
+}
diff --git a/pulsar-io/elastic-search/src/test/resources/sinkConfig.yaml b/pulsar-io/elastic-search/src/test/resources/sinkConfig.yaml
new file mode 100644
index 0000000000..1872d81329
--- /dev/null
+++ b/pulsar-io/elastic-search/src/test/resources/sinkConfig.yaml
@@ -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.
+#
+
+{
+ "elasticSearchUrl": "http://localhost:90902",
+ "indexName": "myIndex",
+ "username": "scooby",
+ "password": "doobie"
+}
\ No newline at end of file
diff --git a/pulsar-io/hdfs/pom.xml b/pulsar-io/hdfs/pom.xml
index 0d552077d5..dde6e693db 100644
--- a/pulsar-io/hdfs/pom.xml
+++ b/pulsar-io/hdfs/pom.xml
@@ -26,6 +26,7 @@
<version>2.2.0-incubating-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-hdfs</artifactId>
+ <name>Pulsar IO :: Hdfs</name>
<dependencies>
<dependency>
diff --git a/pulsar-io/pom.xml b/pulsar-io/pom.xml
index 5c033708ba..a212df0ba9 100644
--- a/pulsar-io/pom.xml
+++ b/pulsar-io/pom.xml
@@ -42,6 +42,7 @@
<module>hdfs</module>
<module>jdbc</module>
<module>data-genenator</module>
+ <module>elastic-search</module>
</modules>
</project>
diff --git a/tests/integration/pom.xml b/tests/integration/pom.xml
index 7a8915965b..ec8d7e71a0 100644
--- a/tests/integration/pom.xml
+++ b/tests/integration/pom.xml
@@ -121,6 +121,13 @@
<artifactId>jackson-dataformat-yaml</artifactId>
<scope>test</scope>
</dependency>
+
+ <dependency>
+ <groupId>org.elasticsearch.client</groupId>
+ <artifactId>elasticsearch-rest-high-level-client</artifactId>
+ <version>6.3.2</version>
+ </dependency>
+
</dependencies>
<build>
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/ElasticSearchContainer.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/ElasticSearchContainer.java
new file mode 100644
index 0000000000..a1184899b4
--- /dev/null
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/ElasticSearchContainer.java
@@ -0,0 +1,47 @@
+/**
+ * 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.tests.integration.containers;
+
+import org.testcontainers.containers.wait.strategy.HostPortWaitStrategy;
+
+public class ElasticSearchContainer extends ChaosContainer<ElasticSearchContainer> {
+
+ public static final String NAME = "ElasticSearch";
+ static final Integer[] PORTS = { 9200, 9300 };
+
+ private static final String IMAGE_NAME = "docker.elastic.co/elasticsearch/elasticsearch:6.4.0";
+
+ public ElasticSearchContainer(String clusterName) {
+ super(clusterName, IMAGE_NAME);
+ }
+
+ @Override
+ protected void configure() {
+ super.configure();
+ this.withNetworkAliases(NAME)
+ .withExposedPorts(PORTS)
+ .withEnv("discovery.type", "single-node")
+ .withCreateContainerCmdModifier(createContainerCmd -> {
+ createContainerCmd.withHostName(NAME);
+ createContainerCmd.withName(clusterName + "-" + NAME);
+ })
+ .waitingFor(new HostPortWaitStrategy());
+ }
+
+}
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java
index 0bf9ded6b3..1cb388eb4d 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java
@@ -84,6 +84,11 @@ public void testJdbcSink() throws Exception {
testSink(new JdbcSinkTester(), true);
}
+ @Test(enabled = false)
+ public void testElasticSearchSink() throws Exception {
+ testSink(new ElasticSearchSinkTester(), true);
+ }
+
private void testSink(SinkTester tester, boolean builtin) throws Exception {
tester.findSinkServiceContainer(pulsarCluster.getExternalServices());
@@ -778,4 +783,4 @@ private static void publishAndConsumeAvroMessages(String inputTopic,
}
}
-}
+}
\ No newline at end of file
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/ElasticSearchSinkTester.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/ElasticSearchSinkTester.java
new file mode 100644
index 0000000000..0effc8eadd
--- /dev/null
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/ElasticSearchSinkTester.java
@@ -0,0 +1,75 @@
+/**
+ * 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.tests.integration.io;
+
+import static com.google.common.base.Preconditions.checkState;
+import static org.testng.Assert.assertTrue;
+
+import java.util.Map;
+
+import org.apache.http.Header;
+import org.apache.http.HttpHost;
+import org.apache.pulsar.tests.integration.containers.ElasticSearchContainer;
+import org.elasticsearch.action.search.SearchRequest;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.client.RestClient;
+import org.elasticsearch.client.RestClientBuilder;
+import org.elasticsearch.client.RestHighLevelClient;
+import org.testcontainers.containers.GenericContainer;
+
+public class ElasticSearchSinkTester extends SinkTester {
+
+ private RestHighLevelClient elasticClient;
+
+ public ElasticSearchSinkTester() {
+ super(SinkType.ELASTIC_SEARCH);
+
+ sinkConfig.put("elasticSearchUrl", "http://localhost:9200");
+ sinkConfig.put("indexName", "test-index");
+ }
+
+ @Override
+ public void findSinkServiceContainer(Map<String, GenericContainer<?>> externalServices) {
+ GenericContainer<?> container = externalServices.get(ElasticSearchContainer.NAME);
+ checkState(container instanceof ElasticSearchContainer,
+ "No ElasticSearch service found in the cluster");
+ }
+
+ @Override
+ public void prepareSink() throws Exception {
+ RestClientBuilder builder = RestClient.builder(new HttpHost("localhost", 9200, "http"));
+ elasticClient = new RestHighLevelClient(builder);
+ }
+
+ @Override
+ public void validateSinkResult(Map<String, String> kvs) {
+
+ SearchRequest searchRequest = new SearchRequest("test-index");
+ searchRequest.types("doc");
+
+ try {
+ Header headers = null;
+ SearchResponse searchResult = elasticClient.search(searchRequest, headers);
+ assertTrue(searchResult.getHits().getTotalHits() > 0);
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+
+}
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/SinkTester.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/SinkTester.java
index 7f4b2d9a1b..5e060d5e77 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/SinkTester.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/SinkTester.java
@@ -34,7 +34,8 @@
CASSANDRA,
KAFKA,
JDBC,
- HDFS
+ HDFS,
+ ELASTIC_SEARCH
}
protected final SinkType sinkType;
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/suites/PulsarTestSuite.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/suites/PulsarTestSuite.java
index 438c96e2a1..b4a6b83f58 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/suites/PulsarTestSuite.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/suites/PulsarTestSuite.java
@@ -20,6 +20,7 @@
import java.util.Map;
import org.apache.pulsar.tests.integration.containers.CassandraContainer;
+import org.apache.pulsar.tests.integration.containers.ElasticSearchContainer;
import org.apache.pulsar.tests.integration.containers.HdfsContainer;
import org.apache.pulsar.tests.integration.topologies.PulsarClusterSpec.PulsarClusterSpecBuilder;
import org.apache.pulsar.tests.integration.topologies.PulsarClusterTestBase;
@@ -75,6 +76,11 @@ protected PulsarClusterSpecBuilder beforeSetupCluster(String clusterName, Pulsar
jdbcServiceName,
new MySQLContainer()
.withExposedPorts(3306));
+
+ externalServices.put(
+ ElasticSearchContainer.NAME,
+ new ElasticSearchContainer(ElasticSearchContainer.NAME)
+ .withExposedPorts(9200));
builder = builder.externalServices(externalServices);
----------------------------------------------------------------
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