Elastic connector (#2546)

### Motivation

Added a sink connector that writes JSON documents into ElasticSearch

### Modifications

Added new pulsar-io module and associated integration tests

### Result

An ElasticSearch sink connector will be available for use.

pulsar-io/elastic-search/pom.xml

+<!--
+
+    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

pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchAbstractSink.java

+/**
+ * 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;
+    }
+}

pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchConfig.java

+/**
+ * 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

pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchStringSink.java

+/**
+ * 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()));
+    }
+}

pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/package-info.java

+/**
+ * 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

pulsar-io/elastic-search/src/main/resources/META-INF/services/pulsar-io.yaml

+#
+# 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

pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchConfigTests.java

+/**
+ * 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());
+    }
+}

pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSinkTests.java

+/**
+ * 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();
+    }
+}

pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/data/Profile.java

+/**
+ * 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;
+}

pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/data/UserProfile.java

+/**
+ * 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;
+}

pulsar-io/elastic-search/src/test/resources/sinkConfig.yaml

+#
+# 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

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>

pulsar-io/pom.xml

@@ -42,6 +42,7 @@
     <module>hdfs</module>
     <module>jdbc</module>
     <module>data-genenator</module>
+    <module>elastic-search</module>
   </modules>
 
 </project>

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>

tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/ElasticSearchContainer.java

+/**
+ * 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());
+    }
+
+}

tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java

@@ -84,6 +84,11 @@ public abstract class PulsarFunctionsTest extends PulsarFunctionsTestBase {
         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 @@ public abstract class PulsarFunctionsTest extends PulsarFunctionsTestBase {
         }
     }
 
-}
+}
\ No newline at end of file

tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/ElasticSearchSinkTester.java

+/**
+ * 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();
+        }
+    }
+
+}

tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/SinkTester.java

@@ -34,7 +34,8 @@ public abstract class SinkTester {
         CASSANDRA,
         KAFKA,
         JDBC,
-        HDFS
+        HDFS,
+        ELASTIC_SEARCH
     }
 
     protected final SinkType sinkType;

tests/integration/src/test/java/org/apache/pulsar/tests/integration/suites/PulsarTestSuite.java

@@ -20,6 +20,7 @@ package org.apache.pulsar.tests.integration.suites;
 
 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 @@ public class PulsarTestSuite extends PulsarClusterTestBase implements ITest {
             jdbcServiceName,
             new MySQLContainer()
                 .withExposedPorts(3306));
+        
+        externalServices.put(
+                ElasticSearchContainer.NAME, 
+                new ElasticSearchContainer(ElasticSearchContainer.NAME)
+                .withExposedPorts(9200));
 
         builder = builder.externalServices(externalServices);