[Issue 5690][pulsar-io] Make type name of Elasticsearch sink connector configurable.

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

@@ -52,6 +52,15 @@ public class ElasticSearchConfig implements Serializable {
     )
     private String indexName;
 
+    @FieldDoc(
+        required = false,
+        defaultValue = "_doc",
+        help = "The type name that the connector writes messages to, with the default value set to _doc." +
+                " This value should be set explicitly to a valid type name other than _doc for Elasticsearch version before 6.2," +
+                " and left to the default value otherwise."
+    )
+    private String typeName = "_doc";
+
     @FieldDoc(
         required = false,
         defaultValue = "1",

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

@@ -61,8 +61,6 @@
 )
 public class ElasticSearchSink implements Sink<byte[]> {
 
-    protected static final String DOCUMENT = "_doc";
-
     private URL url;
     private RestHighLevelClient client;
     private CredentialsProvider credentialsProvider;
@@ -84,7 +82,7 @@ public void close() throws Exception {
     public void write(Record<byte[]> record) {
         KeyValue<String, byte[]> keyValue = extractKeyValue(record);
         IndexRequest indexRequest = Requests.indexRequest(elasticSearchConfig.getIndexName());
-        indexRequest.type(DOCUMENT);
+        indexRequest.type(elasticSearchConfig.getTypeName());
         indexRequest.source(keyValue.getValue(), XContentType.JSON);
 
         try {

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

@@ -20,9 +20,11 @@
 
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNull;
 
 import java.io.File;
 import java.io.IOException;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -37,25 +39,41 @@ public final void loadFromYamlFileTest() throws IOException {
         assertNotNull(config);
         assertEquals(config.getElasticSearchUrl(), "http://localhost:90902");
         assertEquals(config.getIndexName(), "myIndex");
+        assertEquals(config.getTypeName(), "doc");
         assertEquals(config.getUsername(), "scooby");
-        assertEquals(config.getPassword(), "doobie");               
+        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("typeName", "doc");
         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.getTypeName(), "doc");
         assertEquals(config.getUsername(), "racerX");
         assertEquals(config.getPassword(), "go-speedie-go");  
     }
+
+    @Test
+    public final void defaultValueTest() throws IOException {
+        ElasticSearchConfig config = ElasticSearchConfig.load(Collections.emptyMap());
+
+        assertNull(config.getElasticSearchUrl());
+        assertNull(config.getIndexName());
+        assertEquals(config.getTypeName(), "_doc");
+        assertNull(config.getUsername());
+        assertNull(config.getPassword());
+        assertEquals(config.getIndexNumberOfReplicas(), 1);
+        assertEquals(config.getIndexNumberOfShards(), 1);
+    }
 
     @Test
     public final void validValidateTest() throws IOException {

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

@@ -20,6 +20,7 @@
 {
    "elasticSearchUrl": "http://localhost:90902",
    "indexName": "myIndex",
+   "typeName": "doc",
    "username": "scooby",
    "password": "doobie"
-}
+}

site2/docs/io-elasticsearch-sink.md

@@ -16,21 +16,26 @@ The configuration of the ElasticSearch sink connector has the following properti
 |------|----------|----------|---------|-------------|
 | `elasticSearchUrl` | String| true |" " (empty string)| The URL of elastic search cluster to which the connector connects. |
 | `indexName` | String| true |" " (empty string)| The index name to which the connector writes messages. |
+| `typeName` | String | false | "_doc" | The type name to which the connector writes messages to. <br><br> The value should be set explicitly to a valid type name other than "_doc" for Elasticsearch version before 6.2, and left to default otherwise. |
 | `indexNumberOfShards` | int| false |1| The number of shards of the index. |
 | `indexNumberOfReplicas` | int| false |1 | The number of replicas of the index. |
 | `username` | String| false |" " (empty string)| The username used by the connector to connect to the elastic search cluster. <br><br>If `username` is set, then `password` should also be provided. |
 | `password` | String| false | " " (empty string)|The password used by the connector to connect to the elastic search cluster. <br><br>If `username` is set, then `password` should also be provided.  |
 
-### Example
+## Example
 
 Before using the ElasticSearch sink connector, you need to create a configuration file through one of the following methods.
 
+### Configuration
+
+#### For Elasticsearch After 6.2
+
 * JSON 
 
     ```json
     {
-        "elasticSearchUrl": "http://localhost:90902",
-        "indexName": "myIndex",
+        "elasticSearchUrl": "http://localhost:9200",
+        "indexName": "my_index",
         "username": "scooby",
         "password": "doobie"
     }
@@ -40,10 +45,95 @@ Before using the ElasticSearch sink connector, you need to create a configuratio
 
     ```yaml
     configs:
-        elasticSearchUrl: "http://localhost:90902"
-        indexName: "myIndex"
+        elasticSearchUrl: "http://localhost:9200"
+        indexName: "my_index"
         username: "scooby"
         password: "doobie"
     ```
 
+#### For Elasticsearch Before 6.2
+
+* JSON 
+
+    ```json
+    {
+        "elasticSearchUrl": "http://localhost:9200",
+        "indexName": "my_index",
+        "typeName": "doc",
+        "username": "scooby",
+        "password": "doobie"
+    }
+    ```
+
+* YAML
+
+    ```yaml
+    configs:
+        elasticSearchUrl: "http://localhost:9200"
+        indexName: "my_index"
+        typeName: "doc"
+        username: "scooby"
+        password: "doobie"
+    ```
+
+### Usage
+
+1. Start a single node Elasticsearch cluster.
+
+    ```bash
+    $ docker run -p 9200:9200 -p 9300:9300 \
+        -e "discovery.type=single-node" \
+        docker.elastic.co/elasticsearch/elasticsearch:7.5.1
+    ```
+
+2. Start a Pulsar service locally in standalone mode.
+    ```bash
+    $ bin/pulsar standalone
+    ```
+    Make sure the nar file is available at `connectors/pulsar-io-elastic-search-{{pulsar:version}}.nar`.
+
+3. Start the Pulsar Elasticsearch connector in local run mode using one of the following methods.
+    * Use the **JSON** configuration as shown previously. 
+        ```bash
+        $ bin/pulsar-admin sinks localrun \
+            --archive connectors/pulsar-io-elastic-search-{{pulsar:version}}.nar \
+            --tenant public \
+            --namespace default \
+            --name elasticsearch-test-sink \
+            --sink-type elastic_search \
+            --sink-config '{"elasticSearchUrl":"http://localhost:9200","indexName": "my_index","username": "scooby","password": "doobie"}' \
+            --inputs elasticsearch_test
+        ```
+    * Use the **YAML** configuration file as shown previously.
+
+        ```bash
+        $ bin/pulsar-admin sinks localrun \
+            --archive connectors/pulsar-io-elastic-search-{{pulsar:version}}.nar \
+            --tenant public \
+            --namespace default \
+            --name elasticsearch-test-sink \
+            --sink-type elastic_search \
+            --sink-config-file elasticsearch-sink.yml \
+            --inputs elasticsearch_test
+        ```
+
+4. Publish records to the topic.
+
+    ```bash
+    $ bin/pulsar-client produce elasticsearch_test --messages "{\"a\":1}"
+    ```
 
+5. Check documents in Elasticsearch.
+
+    * refresh the index
+        ```bash
+            $ curl -s http://localhost:9200/my_index/_refresh
+        ``` 
+    * search documents
+        ```bash
+            $ curl -s http://localhost:9200/my_index/_search
+        ```
+        You can see the record that published earlier has been successfully written into Elasticsearch.
+        ```json
+        {"took":2,"timed_out":false,"_shards":{"total":1,"successful":1,"skipped":0,"failed":0},"hits":{"total":{"value":1,"relation":"eq"},"max_score":1.0,"hits":[{"_index":"my_index","_type":"_doc","_id":"FSxemm8BLjG_iC0EeTYJ","_score":1.0,"_source":{"a":1}}]}}
+        ```