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 index 9a7b1cbe58ad4..b646236af21ba 100644 --- 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 @@ -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", diff --git a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java index 58eb30aa55c02..d55510d1408b4 100644 --- a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java +++ b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java @@ -61,8 +61,6 @@ ) public class ElasticSearchSink implements Sink { - 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 record) { KeyValue keyValue = extractKeyValue(record); IndexRequest indexRequest = Requests.indexRequest(elasticSearchConfig.getIndexName()); - indexRequest.type(DOCUMENT); + indexRequest.type(elasticSearchConfig.getTypeName()); indexRequest.source(keyValue.getValue(), XContentType.JSON); try { 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 index 63fbe1680c80f..68928ff07bce8 100644 --- 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 @@ -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,8 +39,9 @@ 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 @@ -46,6 +49,7 @@ public final void loadFromMapTest() throws IOException { Map map = new HashMap (); map.put("elasticSearchUrl", "http://localhost:90902"); map.put("indexName", "myIndex"); + map.put("typeName", "doc"); map.put("username", "racerX"); map.put("password", "go-speedie-go"); @@ -53,9 +57,23 @@ public final void loadFromMapTest() throws IOException { 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 { diff --git a/pulsar-io/elastic-search/src/test/resources/sinkConfig.yaml b/pulsar-io/elastic-search/src/test/resources/sinkConfig.yaml index 1872d8132982d..4347a4a09ad72 100644 --- a/pulsar-io/elastic-search/src/test/resources/sinkConfig.yaml +++ b/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" -} \ No newline at end of file +} diff --git a/site2/docs/io-elasticsearch-sink.md b/site2/docs/io-elasticsearch-sink.md index d3312e5d82e47..a7c4d4cf762cc 100644 --- a/site2/docs/io-elasticsearch-sink.md +++ b/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.

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.

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.

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}}]}} + ```