david-streamlio closed pull request #2539: Elastic connector URL: https://github.com/apache/incubator-pulsar/pull/2539
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..830f9e4933 --- /dev/null +++ b/pulsar-io/elastic-search/src/test/resources/sinkConfig.yaml @@ -0,0 +1,24 @@ +# +# 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 6d3fdc4064..e92f807a0e 100644 --- a/tests/integration/pom.xml +++ b/tests/integration/pom.xml @@ -109,6 +109,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..a03caf4611 --- /dev/null +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/ElasticSearchContainer.java @@ -0,0 +1,29 @@ +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"; + + protected 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 17634a1ab8..791c498c09 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 @@ -42,7 +42,15 @@ import org.apache.pulsar.tests.integration.docker.ContainerExecResult; import org.apache.pulsar.tests.integration.functions.utils.CommandGenerator; import org.apache.pulsar.tests.integration.functions.utils.CommandGenerator.Runtime; -import org.apache.pulsar.tests.integration.io.*; +import org.apache.pulsar.tests.integration.io.CassandraSinkTester; +import org.apache.pulsar.tests.integration.io.ElasticSearchSinkTester; +import org.apache.pulsar.tests.integration.io.HdfsSinkTester; +import org.apache.pulsar.tests.integration.io.JdbcSinkTester; +import org.apache.pulsar.tests.integration.io.JdbcSinkTester.Foo; +import org.apache.pulsar.tests.integration.io.KafkaSinkTester; +import org.apache.pulsar.tests.integration.io.KafkaSourceTester; +import org.apache.pulsar.tests.integration.io.SinkTester; +import org.apache.pulsar.tests.integration.io.SourceTester; import org.apache.pulsar.tests.integration.topologies.FunctionRuntimeType; import org.apache.pulsar.tests.integration.topologies.PulsarCluster; import org.testng.Assert; @@ -60,17 +68,17 @@ @Test public void testKafkaSink() throws Exception { - testSink(new KafkaSinkTester(), true); + testSink(new KafkaSinkTester()); } @Test public void testCassandraSink() throws Exception { - testSink(new CassandraSinkTester(), true); + testSink(new CassandraSinkTester()); } - - @Test - public void testCassandraArchiveSink() throws Exception { - testSink(new CassandraSinkArchiveTester(), false); + + @Test(enabled = false) + public void testHdfsSink() throws Exception { + testSink(new HdfsSinkTester()); } @Test(enabled = false) @@ -80,10 +88,15 @@ public void testHdfsSink() throws Exception { @Test public void testJdbcSink() throws Exception { - testSink(new JdbcSinkTester(), true); + testSink(new JdbcSinkTester()); + } + + @Test + public void testElasticSearchSink() throws Exception { + testSink(new ElasticSearchSinkTester()); } - private void testSink(SinkTester tester, boolean builtin) throws Exception { + private void testSink(SinkTester tester) throws Exception { tester.findSinkServiceContainer(pulsarCluster.getExternalServices()); final String tenant = TopicName.PUBLIC_TENANT; @@ -91,7 +104,7 @@ private void testSink(SinkTester tester, boolean builtin) throws Exception { final String inputTopicName = "test-sink-connector-" + tester.getSinkType() + "-" + functionRuntimeType + "-input-topic-" + randomName(8); final String sinkName = "test-sink-connector-" - + tester.getSinkType().name().toLowerCase() + "-" + functionRuntimeType + "-name-" + randomName(8); + + tester.getSinkType() + "-" + functionRuntimeType + "-name-" + randomName(8); final int numMessages = 20; // prepare the testing environment for sink @@ -101,7 +114,7 @@ private void testSink(SinkTester tester, boolean builtin) throws Exception { submitSinkConnector(tester, tenant, namespace, sinkName, inputTopicName); // get sink info - getSinkInfoSuccess(tester, tenant, namespace, sinkName, builtin); + getSinkInfoSuccess(tester, tenant, namespace, sinkName); // get sink status getSinkStatus(tenant, namespace, sinkName); @@ -109,7 +122,7 @@ private void testSink(SinkTester tester, boolean builtin) throws Exception { // produce messages Map<String, String> kvs; if (tester instanceof JdbcSinkTester) { - kvs = produceSchemaMessagesToInputTopic(inputTopicName, numMessages, AvroSchema.of(JdbcSinkTester.Foo.class)); + kvs = produceSchemaMessagesToInputTopic(inputTopicName, numMessages, AvroSchema.of(Foo.class)); } else { kvs = produceMessagesToInputTopic(inputTopicName, numMessages); } @@ -136,31 +149,16 @@ protected void submitSinkConnector(SinkTester tester, String namespace, String sinkName, String inputTopicName) throws Exception { - String[] commands; - if (tester.getSinkType() != SinkTester.SinkType.UNDEFINED) { - commands = new String[] { - PulsarCluster.ADMIN_SCRIPT, - "sink", "create", - "--tenant", tenant, - "--namespace", namespace, - "--name", sinkName, - "--sink-type", tester.sinkType().name().toLowerCase(), - "--sinkConfig", new Gson().toJson(tester.sinkConfig()), - "--inputs", inputTopicName - }; - } else { - commands = new String[] { - PulsarCluster.ADMIN_SCRIPT, - "sink", "create", - "--tenant", tenant, - "--namespace", namespace, - "--name", sinkName, - "--archive", tester.getSinkArchive(), - "--classname", tester.getSinkClassName(), - "--sinkConfig", new Gson().toJson(tester.sinkConfig()), - "--inputs", inputTopicName - }; - } + String[] commands = { + PulsarCluster.ADMIN_SCRIPT, + "sink", "create", + "--tenant", tenant, + "--namespace", namespace, + "--name", sinkName, + "--sink-type", tester.sinkType().name(), + "--sinkConfig", new Gson().toJson(tester.sinkConfig()), + "--inputs", inputTopicName + }; log.info("Run command : {}", StringUtils.join(commands, ' ')); ContainerExecResult result = pulsarCluster.getAnyWorker().execCmd(commands); assertTrue( @@ -171,8 +169,7 @@ protected void submitSinkConnector(SinkTester tester, protected void getSinkInfoSuccess(SinkTester tester, String tenant, String namespace, - String sinkName, - boolean builtin) throws Exception { + String sinkName) throws Exception { String[] commands = { PulsarCluster.ADMIN_SCRIPT, "functions", @@ -183,17 +180,10 @@ protected void getSinkInfoSuccess(SinkTester tester, }; ContainerExecResult result = pulsarCluster.getAnyWorker().execCmd(commands); log.info("Get sink info : {}", result.getStdout()); - if (builtin) { - assertTrue( - result.getStdout().contains("\"builtin\": \"" + tester.getSinkType().name().toLowerCase() + "\""), - result.getStdout() - ); - } else { - assertTrue( - result.getStdout().contains("\"className\": \"" + tester.getSinkClassName() + "\""), - result.getStdout() - ); - } + assertTrue( + result.getStdout().contains("\"builtin\": \"" + tester.getSinkType() + "\""), + result.getStdout() + ); } protected void getSinkStatus(String tenant, String namespace, String sinkName) throws Exception { @@ -258,7 +248,7 @@ protected void getSinkStatus(String tenant, String namespace, String sinkName) t for (int i = 0; i < numMessages; i++) { String key = "key-" + i; - JdbcSinkTester.Foo obj = new JdbcSinkTester.Foo(); + Foo obj = new Foo(); obj.setField1("field1_" + i); obj.setField2("field2_" + i); obj.setField3(i); 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..435638eff4 --- /dev/null +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/ElasticSearchSinkTester.java @@ -0,0 +1,58 @@ +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 static final String NAME = "ElasticSearch"; + 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(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; ---------------------------------------------------------------- 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: us...@infra.apache.org With regards, Apache Git Services
