[ISSUE #4824] Add HTTP Sink Connector

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

@@ -75,14 +75,14 @@ private void doInit() {
         }
 
         int maxRetries = this.httpSinkConfig.connectorConfig.getRetryConfig().getMaxRetries();
-        if (maxRetries < 0) {
-            throw new IllegalArgumentException("Max retries must be greater than or equal to 0.");
-        } else if (maxRetries == 0) {
+        if (maxRetries == 0) {
             // Use the original sink handler
             this.sinkHandler = nonRetryHandler;
-        } else {
+        } else if (maxRetries > 0) {
             // Wrap the sink handler with a retry handler
             this.sinkHandler = new RetryHttpSinkHandler(this.httpSinkConfig.connectorConfig, nonRetryHandler);
+        } else {
+            throw new IllegalArgumentException("Max retries must be greater than or equal to 0.");
         }
     }
 

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

@@ -20,6 +20,7 @@
 import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
 
 import java.time.LocalDateTime;
+import java.util.UUID;
 
 import lombok.Builder;
 import lombok.Data;
@@ -33,7 +34,11 @@ public class HttpConnectRecord {
 
     private String type;
 
-    private String timestamp;
+    private String time;
+
+    private String uuid;
+
+    private String eventId;
 
     private ConnectRecord data;
 
@@ -46,7 +51,9 @@ public class HttpConnectRecord {
     public static HttpConnectRecord convertConnectRecord(ConnectRecord record, String type) {
         return HttpConnectRecord.builder()
             .type(type)
-            .timestamp(LocalDateTime.now().toString())
+            .time(LocalDateTime.now().toString())
+            .uuid(UUID.randomUUID().toString())
+            .eventId(type + "-" + record.getTimestamp())
             .data(record)
             .build();
     }

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

@@ -123,39 +123,43 @@ public void handle(ConnectRecord record) {
      */
     @Override
     public Future<HttpResponse<Buffer>> deliver(URI url, HttpConnectRecord httpConnectRecord) {
-        // Only webhook mode needs to use the firstTryId
-        String firstTryId = UUID.randomUUID().toString();
+        // Only webhook mode needs to use the UUID to identify the request
+        String id = httpConnectRecord.getUuid();
 
         // Build the retry policy
         RetryPolicy<HttpResponse<Buffer>> retryPolicy = retryPolicyBuilder
-            .onSuccess(e -> {
+            .onSuccess(event -> {
                 if (connectorConfig.getWebhookConfig().isActivate()) {
                     // convert the result to an HttpExportRecord
-                    HttpExportRecord exportRecord = covertToExportRecord(e, e.getResult(), e.getException(), url, firstTryId);
+                    HttpExportRecord exportRecord = covertToExportRecord(httpConnectRecord, event, event.getResult(), event.getException(), url, id);
                     // add the data to the queue
                     ((WebhookHttpSinkHandler) sinkHandler).addDataToQueue(exportRecord);
                 }
             })
-            .onRetry(e -> {
+            .onRetry(event -> {
                 if (log.isDebugEnabled()) {
-                    log.warn("Retrying the request to {} for the {} time. HttpConnectRecord= {}", url, e.getAttemptCount(), httpConnectRecord);
+                    log.warn("Retrying the request to {} for the {} time. HttpConnectRecord= {}", url, event.getAttemptCount(), httpConnectRecord);
                 } else {
-                    log.warn("Retrying the request to {} for the {} time.", url, e.getAttemptCount());
+                    log.warn("Retrying the request to {} for the {} time.", url, event.getAttemptCount());
                 }
                 if (connectorConfig.getWebhookConfig().isActivate()) {
-                    HttpExportRecord exportRecord = covertToExportRecord(e, e.getLastResult(), e.getLastException(), url, firstTryId);
+                    HttpExportRecord exportRecord =
+                        covertToExportRecord(httpConnectRecord, event, event.getLastResult(), event.getLastException(), url, id);
                     ((WebhookHttpSinkHandler) sinkHandler).addDataToQueue(exportRecord);
                 }
+                // update the HttpConnectRecord
+                httpConnectRecord.setTime(LocalDateTime.now().toString());
+                httpConnectRecord.setUuid(UUID.randomUUID().toString());
             })
-            .onFailure(e -> {
+            .onFailure(event -> {
                 if (log.isDebugEnabled()) {
-                    log.error("Failed to send the request to {} after {} attempts. HttpConnectRecord= {}", url, e.getAttemptCount(),
-                        httpConnectRecord, e.getException());
+                    log.error("Failed to send the request to {} after {} attempts. HttpConnectRecord= {}", url, event.getAttemptCount(),
+                        httpConnectRecord, event.getException());
                 } else {
-                    log.error("Failed to send the request to {} after {} attempts.", url, e.getAttemptCount(), e.getException());
+                    log.error("Failed to send the request to {} after {} attempts.", url, event.getAttemptCount(), event.getException());
                 }
                 if (connectorConfig.getWebhookConfig().isActivate()) {
-                    HttpExportRecord exportRecord = covertToExportRecord(e, e.getResult(), e.getException(), url, firstTryId);
+                    HttpExportRecord exportRecord = covertToExportRecord(httpConnectRecord, event, event.getResult(), event.getException(), url, id);
                     ((WebhookHttpSinkHandler) sinkHandler).addDataToQueue(exportRecord);
                 }
             }).build();
@@ -170,25 +174,24 @@ public Future<HttpResponse<Buffer>> deliver(URI url, HttpConnectRecord httpConne
     /**
      * Converts the ExecutionCompletedEvent to an HttpExportRecord.
      *
-     * @param event      the ExecutionCompletedEvent to convert
-     * @param response   the response of the request, may be null
-     * @param e          the exception thrown during the request, may be null
-     * @param url        the URL the request was sent to
-     * @param firstTryId the UUID of the first try
+     * @param httpConnectRecord HttpConnectRecord
+     * @param event             ExecutionEvent
+     * @param response          the response of the request, may be null
+     * @param e                 the exception thrown during the request, may be null
+     * @param url               the URL the request was sent to
+     * @param id                UUID
      * @return the converted HttpExportRecord
      */
-    private HttpExportRecord covertToExportRecord(ExecutionEvent event, HttpResponse<Buffer> response, Throwable e, URI url, String firstTryId) {
+    private HttpExportRecord covertToExportRecord(HttpConnectRecord httpConnectRecord, ExecutionEvent event, HttpResponse<Buffer> response,
+        Throwable e, URI url, String id) {
         HttpExportMetadataBuilder builder = HttpExportMetadata.builder()
             .url(url.toString())
             .receivedTime(LocalDateTime.now())
-            .retryNum(event.getAttemptCount() - 1);
+            .retryNum(event.getAttemptCount() - 1)
+            .uuid(httpConnectRecord.getUuid());
 
-        if (event.getAttemptCount() == 1) {
-            builder.retriedBy(null)
-                .uuid(firstTryId);
-        } else {
-            builder.retriedBy(firstTryId)
-                .uuid(UUID.randomUUID().toString());
+        if (event.getAttemptCount() > 1) {
+            builder.retriedBy(id);
         }
 
         if (response != null) {

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

@@ -35,7 +35,6 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Objects;
-import java.util.UUID;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -233,7 +232,7 @@ public Future<HttpResponse<Buffer>> deliver(URI url, HttpConnectRecord httpConne
                 .url(url.toString())
                 .receivedTime(LocalDateTime.now())
                 .retriedBy(null)
-                .uuid(UUID.randomUUID().toString())
+                .uuid(httpConnectRecord.getUuid())
                 .retryNum(0);
 
             if (arr.succeeded()) {
@@ -271,11 +270,7 @@ public void addDataToQueue(HttpExportRecord exportRecord) {
         // Try to put the received data into the queue
         if (receivedDataQueue.offer(exportRecord)) {
             currentQueueSize.incrementAndGet();
-            if (log.isDebugEnabled()) {
-                log.debug("Successfully put the received data into the queue: {}", exportRecord);
-            } else {
-                log.info("Successfully put the received data into the queue");
-            }
+            log.debug("Successfully put the received data into the queue: {}", exportRecord);
         } else {
             log.error("Failed to put the received data into the queue: {}", exportRecord);
         }