[ISSUE #4824] Add HTTP Sink Connector

eventmesh-connectors/eventmesh-connector-http/build.gradle

@@ -18,10 +18,13 @@
 dependencies {
     api project(":eventmesh-openconnect:eventmesh-openconnect-java")
     implementation project(":eventmesh-common")
-    implementation 'io.cloudevents:cloudevents-http-vertx:2.3.0'
+
+    implementation 'io.cloudevents:cloudevents-http-vertx:3.0.0'
     implementation 'io.vertx:vertx-web:4.4.6'
+    implementation 'io.vertx:vertx-web-client:4.4.6'
 
     testImplementation "org.apache.httpcomponents:httpclient"
+    testImplementation 'org.mock-server:mockserver-netty:5.15.0'
     compileOnly 'org.projectlombok:lombok'
     annotationProcessor 'org.projectlombok:lombok'
 }

eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/server/HttpConnectServer.java

@@ -18,6 +18,7 @@
 package org.apache.eventmesh.connector.http.server;
 
 import org.apache.eventmesh.connector.http.config.HttpServerConfig;
+import org.apache.eventmesh.connector.http.sink.connector.HttpSinkConnector;
 import org.apache.eventmesh.connector.http.source.connector.HttpSourceConnector;
 import org.apache.eventmesh.openconnect.Application;
 import org.apache.eventmesh.openconnect.util.ConfigUtil;
@@ -33,7 +34,8 @@ public static void main(String[] args) throws Exception {
         }
 
         if (serverConfig.isSinkEnable()) {
-            // TODO support sink connector
+            Application httpSinkApp = new Application();
+            httpSinkApp.run(HttpSinkConnector.class);
         }
     }
 

eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/sink/config/HttpSinkConfig.java

@@ -0,0 +1,31 @@
+/*
+ * 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.eventmesh.connector.http.sink.config;
+
+
+import org.apache.eventmesh.openconnect.api.config.SinkConfig;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+@Data
+@EqualsAndHashCode(callSuper = true)
+public class HttpSinkConfig extends SinkConfig {
+
+    public SinkConnectorConfig connectorConfig;
+}

eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/sink/config/SinkConnectorConfig.java

@@ -0,0 +1,36 @@
+/*
+ * 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.eventmesh.connector.http.sink.config;
+
+import lombok.Data;
+
+@Data
+public class SinkConnectorConfig {
+
+    private String connectorName;
+
+    private String host;
+
+    private int port;
+
+    private String path;
+
+    private boolean ssl;
+
+    private int idleTimeout;
+}

eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/sink/connector/HttpSinkConnector.java

@@ -0,0 +1,134 @@
+/*
+ * 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.eventmesh.connector.http.sink.connector;
+
+import org.apache.eventmesh.connector.http.sink.config.HttpSinkConfig;
+import org.apache.eventmesh.openconnect.api.config.Config;
+import org.apache.eventmesh.openconnect.api.connector.ConnectorContext;
+import org.apache.eventmesh.openconnect.api.connector.SinkConnectorContext;
+import org.apache.eventmesh.openconnect.api.sink.Sink;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
+
+import java.util.List;
+import java.util.Objects;
+
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.Vertx;
+import io.vertx.ext.web.client.WebClient;
+import io.vertx.ext.web.client.WebClientOptions;
+
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+
+
+@Slf4j
+public class HttpSinkConnector implements Sink {
+
+    private HttpSinkConfig httpSinkConfig;
+
+    private WebClient webClient;
+
+    private volatile boolean isRunning = false;
+
+    @Override
+    public Class<? extends Config> configClass() {
+        return HttpSinkConfig.class;
+    }
+
+    @Override
+    public void init(Config config) throws Exception {
+        httpSinkConfig = (HttpSinkConfig) config;
+        doInit();
+    }
+
+    @Override
+    public void init(ConnectorContext connectorContext) throws Exception {
+        SinkConnectorContext sinkConnectorContext = (SinkConnectorContext) connectorContext;
+        this.httpSinkConfig = (HttpSinkConfig) sinkConnectorContext.getSinkConfig();
+        doInit();
+    }
+
+    @SneakyThrows
+    private void doInit() {
+        final Vertx vertx = Vertx.vertx();
+        // TODO Add more configurations
+        WebClientOptions options = new WebClientOptions()
+            .setDefaultHost(this.httpSinkConfig.connectorConfig.getHost())
+            .setDefaultPort(this.httpSinkConfig.connectorConfig.getPort())
+            .setSsl(this.httpSinkConfig.connectorConfig.isSsl())
+            .setIdleTimeout(this.httpSinkConfig.connectorConfig.getIdleTimeout());
+        this.webClient = WebClient.create(vertx, options);
+    }
+
+    @Override
+    public void start() throws Exception {
+        this.isRunning = true;
+    }
+
+    @Override
+    public void commit(ConnectRecord record) {
+
+    }
+
+    @Override
+    public String name() {
+        return this.httpSinkConfig.connectorConfig.getConnectorName();
+    }
+
+    @Override
+    public void stop() throws Exception {
+        this.isRunning = false;
+        this.webClient.close();
+    }
+
+    public boolean isRunning() {
+        return isRunning;
+    }
+
+    @Override
+    public void put(List<ConnectRecord> sinkRecords) {
+        for (ConnectRecord sinkRecord : sinkRecords) {
+            try {
+                if (Objects.isNull(sinkRecord)) {
+                    log.warn("ConnectRecord data is null, ignore.");
+                    continue;
+                }
+                sendMessage(sinkRecord);
+            } catch (Exception e) {
+                log.error("Failed to sink message via HTTP. Exception: ", e);
+            }
+        }
+    }
+
+    private void sendMessage(ConnectRecord record) {
+        this.webClient.post(this.httpSinkConfig.connectorConfig.getPath())
+            .putHeader("Content-Type", "application/json; charset=utf-8")
+            .sendJson(record, ar -> {
+                if (ar.succeeded()) {
+                    log.info("[HttpSinkConnector] Successfully send message via HTTP. Record: timestamp={}, offset={}", record.getTimestamp(),
+                        record.getPosition().getOffset());
+                    if (ar.result().statusCode() != HttpResponseStatus.OK.code()) {
+                        log.error("[HttpSinkConnector] Unexpected response received. StatusCode: {}", ar.result().statusCode());
+                    }
+                } else {
+                    // This branch is only entered if an error occurs at the network layer
+                    log.error("[HttpSinkConnector] Failed to send message via HTTP. Exception: ", ar.cause());
+                }
+            });
+    }
+}

eventmesh-connectors/eventmesh-connector-http/src/main/resources/sink-config.yml

@@ -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.
+#
+
+pubSubConfig:
+  meshAddress: 127.0.0.1:10000
+  subject: TopicTest
+  idc: FT
+  env: PRD
+  group: httpSink
+  appId: 5032
+  userName: httpSinkUser
+  passWord: httpPassWord
+connectorConfig:
+  connectorName: httpSink
+  host: 127.0.0.1
+  port: 8987
+  path: /test
+  ssl: false
+  idleTimeout: 5

eventmesh-connectors/eventmesh-connector-http/src/test/java/org/apache/eventmesh/connector/http/source/connector/HttpSinkConnectorTest.java

@@ -0,0 +1,107 @@
+/*
+ * 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.eventmesh.connector.http.source.connector;
+
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockserver.model.HttpRequest.request;
+
+import org.apache.eventmesh.connector.http.sink.config.HttpSinkConfig;
+import org.apache.eventmesh.connector.http.sink.config.SinkConnectorConfig;
+import org.apache.eventmesh.connector.http.sink.connector.HttpSinkConnector;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordOffset;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordPartition;
+import org.apache.eventmesh.openconnect.util.ConfigUtil;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockserver.client.MockServerClient;
+import org.mockserver.integration.ClientAndServer;
+import org.mockserver.model.HttpRequest;
+import org.mockserver.model.HttpResponse;
+
+import io.vertx.core.http.HttpMethod;
+
+public class HttpSinkConnectorTest {
+
+    private HttpSinkConnector sinkConnector;
+    private SinkConnectorConfig sinkConnectorConfig;
+
+    private ClientAndServer mockServer;
+
+    @BeforeEach
+    public void setUp() throws Exception {
+        sinkConnector = new HttpSinkConnector();
+        HttpSinkConfig sinkConfig = (HttpSinkConfig) ConfigUtil.parse(sinkConnector.configClass());
+        sinkConnectorConfig = sinkConfig.connectorConfig;
+        sinkConnector.init(sinkConfig);
+        sinkConnector.start();
+        mockServer = ClientAndServer.startClientAndServer(sinkConnectorConfig.getPort());
+    }
+
+    @AfterEach
+    public void stopMockServer() throws Exception {
+        sinkConnector.stop();
+        mockServer.close();
+    }
+
+    @Test
+    void testPut() throws InterruptedException {
+        new MockServerClient(sinkConnectorConfig.getHost(), sinkConnectorConfig.getPort())
+            .when(
+                request()
+                    .withMethod("POST")
+                    .withPath(sinkConnectorConfig.getPath())
+            )
+            .respond(
+                HttpResponse.response()
+                    .withStatusCode(200)
+            );
+
+        final int times = 10;
+        List<ConnectRecord> connectRecords = new ArrayList<>();
+        for (int i = 0; i < times; i++) {
+            RecordPartition partition = new RecordPartition();
+            RecordOffset offset = new RecordOffset();
+            long timestamp = System.currentTimeMillis();
+            ConnectRecord connectRecord = new ConnectRecord(partition, offset,
+                timestamp, "test-http " + i);
+            connectRecords.add(connectRecord);
+        }
+
+        sinkConnector.put(connectRecords);
+        // Sleeps for 3 seconds, waiting for the webClient to finish sending all requests
+        Thread.sleep(3000);
+
+        HttpRequest[] allRequests = mockServer.retrieveRecordedRequests(null);
+        // Determine the total number of requests
+        assertEquals(times, allRequests.length);
+
+        for (int i = 0; i < times; i++) {
+            HttpRequest actualRequest = allRequests[0];
+            // Determine the request method
+            assertEquals(HttpMethod.POST.name(), actualRequest.getMethod().getValue());
+        }
+    }
+
+}