feat(connector): sink support for elasticsearch

java/.gitignore

@@ -1,5 +1,10 @@
 # Java
 target/
 test_db/
+bin/
+.settings/
 *.log
-*.class
+*.class
+.project
+.factorypath
+.classpath

java/connector-node/README.md

@@ -21,6 +21,11 @@ If you meet problem, you can try the following to skip the unit test:
 mvn clean package -DskipTests=true
 ```
 
+To disable building the rust library, you can try the following:
+```
+mvn clean package -Dno-build-rust
+```
+
 This will create a `.tar.gz` file with the Connector Node and all its dependencies in the `risingwave/java/connector-node/assembly/target` directory. To run the Connector Node, execute the following command:
 
 ```

java/connector-node/assembly/assembly.xml

@@ -39,6 +39,7 @@
                 <include>*:risingwave-source-cdc</include>
 
                 <!-- Sink connectors -->
+                <include>*:risingwave-sink-es</include>
                 <include>*:risingwave-sink-jdbc</include>
                 <include>*:risingwave-sink-iceberg</include>
                 <include>*:risingwave-sink-deltalake</include>

java/connector-node/assembly/pom.xml

@@ -29,6 +29,10 @@
             <groupId>com.risingwave.java</groupId>
             <artifactId>risingwave-source-cdc</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.risingwave.java</groupId>
+            <artifactId>risingwave-sink-es</artifactId>
+        </dependency>
         <dependency>
             <groupId>com.risingwave.java</groupId>
             <artifactId>risingwave-sink-jdbc</artifactId>

java/connector-node/python-client/integration_tests.py

@@ -156,6 +156,11 @@ def validate_jdbc_sink(input_file):
                                                                                                    rows[i][j]))
                 exit(1)
 
+def test_elasticsearch_sink(input_file):
+    test_sink("elasticsearch",
+              {"url": "http://127.0.0.1:9200",
+               "index": "test"},
+              input_file)
 
 def test_iceberg_sink(input_file):
     test_sink("iceberg",
@@ -196,6 +201,7 @@ def test_deltalake_sink(input_file):
     parser.add_argument('--iceberg_sink', action='store_true', help="run iceberg sink test")
     parser.add_argument('--upsert_iceberg_sink', action='store_true', help="run upsert iceberg sink test")
     parser.add_argument('--deltalake_sink', action='store_true', help="run deltalake sink test")
+    parser.add_argument('--es_sink', action='store_true', help='run elasticsearch sink test')
     parser.add_argument('--input_file', default="./data/sink_input.json", help="input data to run tests")
     args = parser.parse_args()
     if args.file_sink:
@@ -208,3 +214,5 @@ def test_deltalake_sink(input_file):
         test_upsert_iceberg_sink(args.input_file)
     if args.deltalake_sink:
         test_deltalake_sink(args.input_file)
+    if args.es_sink:
+        test_elasticsearch_sink(args.input_file)

java/connector-node/risingwave-connector-service/pom.xml

@@ -94,5 +94,10 @@
             <artifactId>risingwave-sink-deltalake</artifactId>
             <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>com.risingwave.java</groupId>
+            <artifactId>risingwave-sink-es</artifactId>
+            <scope>provided</scope>
+        </dependency>
     </dependencies>
 </project>

java/connector-node/risingwave-connector-service/src/main/java/com/risingwave/connector/SinkUtils.java

@@ -29,6 +29,8 @@ public static SinkFactory getSinkFactory(String sinkType) {
                 return new IcebergSinkFactory();
             case "deltalake":
                 return new DeltaLakeSinkFactory();
+            case "elasticsearch":
+                return new EsSinkFactory();
             default:
                 throw UNIMPLEMENTED
                         .withDescription("unknown sink type: " + sinkType)

java/connector-node/risingwave-connector-test/pom.xml

@@ -135,6 +135,12 @@
             <version>${testcontainers.version}</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>elasticsearch</artifactId>
+            <version>${testcontainers.version}</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-databind</artifactId>
@@ -163,5 +169,10 @@
             <artifactId>risingwave-sink-jdbc</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>com.risingwave.java</groupId>
+            <artifactId>risingwave-sink-es</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 </project>

java/connector-node/risingwave-connector-test/src/test/java/com/risingwave/connector/sink/elasticsearch/EsSinkTest.java

@@ -0,0 +1,104 @@
+// Copyright 2023 RisingWave Labs
+//
+// Licensed 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 com.risingwave.connector.sink.elasticsearch;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+import com.google.common.collect.Iterators;
+import com.google.common.collect.Lists;
+import com.risingwave.connector.EsSink;
+import com.risingwave.connector.EsSinkConfig;
+import com.risingwave.connector.api.TableSchema;
+import com.risingwave.connector.api.sink.ArraySinkRow;
+import com.risingwave.proto.Data;
+import com.risingwave.proto.Data.DataType.TypeName;
+import com.risingwave.proto.Data.Op;
+import java.io.IOException;
+import java.util.Map;
+import org.elasticsearch.action.search.SearchRequest;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.client.RequestOptions;
+import org.elasticsearch.client.RestHighLevelClient;
+import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.search.SearchHit;
+import org.elasticsearch.search.SearchHits;
+import org.elasticsearch.search.builder.SearchSourceBuilder;
+import org.junit.Test;
+import org.testcontainers.elasticsearch.ElasticsearchContainer;
+
+public class EsSinkTest {
+
+    static TableSchema getTestTableSchema() {
+        return new TableSchema(
+                Lists.newArrayList("id", "name"),
+                Lists.newArrayList(
+                        Data.DataType.newBuilder().setTypeName(TypeName.INT32).build(),
+                        Data.DataType.newBuilder().setTypeName(TypeName.VARCHAR).build()),
+                Lists.newArrayList("id", "name"));
+    }
+
+    public void testEsSink(ElasticsearchContainer container) throws IOException {
+        EsSink sink =
+                new EsSink(
+                        new EsSinkConfig(container.getHttpHostAddress(), "test", "$"),
+                        getTestTableSchema());
+        sink.write(
+                Iterators.forArray(
+                        new ArraySinkRow(Op.INSERT, 1, "Alice"),
+                        new ArraySinkRow(Op.INSERT, 2, "Bob")));
+        sink.sync();
+        // container is slow here, but our default flush time is 5s,
+        // so 2s is enough for sync test
+        try {
+            Thread.sleep(3000);
+        } catch (InterruptedException e) {
+            fail(e.getMessage());
+        }
+
+        RestHighLevelClient client = sink.getClient();
+        SearchRequest searchRequest = new SearchRequest("test");
+        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
+        searchSourceBuilder.query(QueryBuilders.matchAllQuery());
+        searchRequest.source(searchSourceBuilder);
+        SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
+
+        SearchHits hits = searchResponse.getHits();
+        assertEquals(2, hits.getHits().length);
+
+        SearchHit hit = hits.getAt(0);
+        Map<String, Object> sourceAsMap = hit.getSourceAsMap();
+        assertEquals(1, sourceAsMap.get("id"));
+        assertEquals("Alice", sourceAsMap.get("name"));
+        assertEquals("1$Alice", hit.getId());
+
+        hit = hits.getAt(1);
+        sourceAsMap = hit.getSourceAsMap();
+        assertEquals(2, sourceAsMap.get("id"));
+        assertEquals("Bob", sourceAsMap.get("name"));
+        assertEquals("2$Bob", hit.getId());
+
+        sink.drop();
+    }
+
+    @Test
+    public void testElasticSearch() throws IOException {
+        ElasticsearchContainer container =
+                new ElasticsearchContainer("docker.elastic.co/elasticsearch/elasticsearch:7.11.0");
+        container.start();
+        testEsSink(container);
+        container.stop();
+    }
+}

java/connector-node/risingwave-sink-es/pom.xml

@@ -0,0 +1,64 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<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">
+    <parent>
+        <artifactId>java-parent</artifactId>
+        <groupId>com.risingwave.java</groupId>
+        <version>1.0-SNAPSHOT</version>
+        <relativePath>../../pom.xml</relativePath>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+
+    <artifactId>risingwave-sink-es</artifactId>
+    <version>1.0-SNAPSHOT</version>
+    <name>risingwave-sink-es</name>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.risingwave.java</groupId>
+            <artifactId>proto</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.risingwave.java</groupId>
+            <artifactId>connector-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.logging.log4j</groupId>
+            <artifactId>log4j-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.logging.log4j</groupId>
+            <artifactId>log4j-slf4j-impl</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.logging.log4j</groupId>
+            <artifactId>log4j-core</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-text</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+            <version>${jackson.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-core</artifactId>
+            <version>${jackson.version}</version>
+        </dependency>
+
+        <!-- ElasticSearch drivers -->
+        <dependency>
+            <groupId>org.elasticsearch</groupId>
+            <artifactId>elasticsearch</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.elasticsearch.client</groupId>
+            <artifactId>elasticsearch-rest-high-level-client</artifactId>
+        </dependency>
+    </dependencies>
+
+</project>

java/connector-node/risingwave-sink-es/src/main/java/com/risingwave/connector/BulkRequestConsumerFactory.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright 2023 RisingWave Labs
+ *
+ * Licensed 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 com.risingwave.connector;
+
+import java.util.function.BiConsumer;
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.bulk.BulkRequest;
+import org.elasticsearch.action.bulk.BulkResponse;
+
+/**
+ * {@link BulkRequestConsumerFactory} is used to bridge incompatible Elasticsearch Java API calls
+ * across different Elasticsearch versions.
+ */
+interface BulkRequestConsumerFactory
+        extends BiConsumer<BulkRequest, ActionListener<BulkResponse>> {}