[ISSUE #5077] HTTP Sink Connector supports result callback

eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/common/SynchronizedCircularFifoQueue.java

@@ -120,9 +120,10 @@ public synchronized int size() {
      */
     public synchronized List<E> fetchRange(int start, int end, boolean removed) {
 
-        if (start < 0 || end > this.size() || start > end) {
+        if (start < 0 || start > end) {
             throw new IllegalArgumentException("Invalid range");
         }
+        end = Math.min(end, this.size());
 
         Iterator<E> iterator = this.iterator();
         List<E> items = new ArrayList<>(end - start);

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

@@ -20,10 +20,10 @@
 import org.apache.eventmesh.common.config.connector.Config;
 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.handle.CommonHttpSinkHandler;
-import org.apache.eventmesh.connector.http.sink.handle.HttpSinkHandler;
-import org.apache.eventmesh.connector.http.sink.handle.RetryHttpSinkHandler;
-import org.apache.eventmesh.connector.http.sink.handle.WebhookHttpSinkHandler;
+import org.apache.eventmesh.connector.http.sink.handler.HttpSinkHandler;
+import org.apache.eventmesh.connector.http.sink.handler.impl.CommonHttpSinkHandler;
+import org.apache.eventmesh.connector.http.sink.handler.impl.HttpSinkHandlerRetryWrapper;
+import org.apache.eventmesh.connector.http.sink.handler.impl.WebhookHttpSinkHandler;
 import org.apache.eventmesh.openconnect.api.ConnectorCreateService;
 import org.apache.eventmesh.openconnect.api.connector.ConnectorContext;
 import org.apache.eventmesh.openconnect.api.connector.SinkConnectorContext;
@@ -86,7 +86,7 @@ private void doInit() {
             this.sinkHandler = nonRetryHandler;
         } else if (maxRetries > 0) {
             // Wrap the sink handler with a retry handler
-            this.sinkHandler = new RetryHttpSinkHandler(this.httpSinkConfig.connectorConfig, nonRetryHandler);
+            this.sinkHandler = new HttpSinkHandlerRetryWrapper(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/config/HttpRetryConfig.java

@@ -24,8 +24,8 @@ public class HttpRetryConfig {
     // maximum number of retries, default 2, minimum 0
     private int maxRetries = 2;
 
-    // retry interval, default 2000ms
-    private int interval = 2000;
+    // retry interval, default 1000ms
+    private int interval = 1000;
 
     // Default value is false, indicating that only requests with network-level errors will be retried.
     // If set to true, all failed requests will be retried, including network-level errors and non-2xx responses.

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

@@ -20,31 +20,60 @@
 import org.apache.eventmesh.common.remote.offset.http.HttpRecordOffset;
 import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
 
+import java.io.Serializable;
 import java.time.LocalDateTime;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.UUID;
 
 import lombok.Builder;
-import lombok.Data;
+import lombok.Getter;
 
 /**
  * a special ConnectRecord for HttpSinkConnector
  */
-@Data
+@Getter
 @Builder
-public class HttpConnectRecord {
+public class HttpConnectRecord implements Serializable {
 
-    private String type;
+    private static final long serialVersionUID = 5271462532332251473L;
+
+    /**
+     * The unique identifier for the HttpConnectRecord
+     */
+    private final String httpRecordId = UUID.randomUUID().toString();
 
-    private String time;
+    /**
+     * The time when the HttpConnectRecord was created
+     */
+    private LocalDateTime createTime;
 
-    private String uuid;
+    /**
+     * The type of the HttpConnectRecord
+     */
+    private String type;
 
+    /**
+     * The event id of the HttpConnectRecord
+     */
     private String eventId;
 
+    /**
+     * The ConnectRecord to be sent
+     */
     private ConnectRecord data;
 
+    @Override
+    public String toString() {
+        return "HttpConnectRecord{"
+            + "createTime=" + createTime
+            + ", httpRecordId='" + httpRecordId
+            + ", type='" + type
+            + ", eventId='" + eventId
+            + ", data=" + data
+            + '}';
+    }
+
     /**
      * Convert ConnectRecord to HttpConnectRecord
      *
@@ -62,11 +91,8 @@ public static HttpConnectRecord convertConnectRecord(ConnectRecord record, Strin
         }
         return HttpConnectRecord.builder()
             .type(type)
-            .time(LocalDateTime.now().toString())
-            .uuid(UUID.randomUUID().toString())
             .eventId(type + "-" + offset)
             .data(record)
             .build();
     }
-
 }

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

@@ -17,6 +17,7 @@
 
 package org.apache.eventmesh.connector.http.sink.data;
 
+import java.io.Serializable;
 import java.time.LocalDateTime;
 
 import lombok.Builder;
@@ -27,7 +28,10 @@
  */
 @Data
 @Builder
-public class HttpExportMetadata {
+public class HttpExportMetadata implements Serializable {
+
+    private static final long serialVersionUID = 1121010466793041920L;
+
     private String url;
 
     private int code;
@@ -36,7 +40,9 @@ public class HttpExportMetadata {
 
     private LocalDateTime receivedTime;
 
-    private String uuid;
+    private String httpRecordId;
+
+    private String recordId;
 
     private String retriedBy;
 

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

@@ -17,6 +17,8 @@
 
 package org.apache.eventmesh.connector.http.sink.data;
 
+import java.io.Serializable;
+
 import lombok.AllArgsConstructor;
 import lombok.Data;
 
@@ -25,7 +27,9 @@
  */
 @Data
 @AllArgsConstructor
-public class HttpExportRecord {
+public class HttpExportRecord implements Serializable {
+
+    private static final long serialVersionUID = 6010283911452947157L;
 
     private HttpExportMetadata metadata;
 

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

@@ -17,6 +17,7 @@
 
 package org.apache.eventmesh.connector.http.sink.data;
 
+import java.io.Serializable;
 import java.util.List;
 
 import lombok.AllArgsConstructor;
@@ -27,7 +28,9 @@
  */
 @Data
 @AllArgsConstructor
-public class HttpExportRecordPage {
+public class HttpExportRecordPage implements Serializable {
+
+    private static final long serialVersionUID = 1143791658357035990L;
 
     private int pageNum;
 

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

@@ -0,0 +1,80 @@
+/*
+ * 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.data;
+
+import lombok.Data;
+
+/**
+ * Single HTTP retry event
+ */
+@Data
+public class HttpRetryEvent {
+
+    public static final String PREFIX = "http-retry-event-";
+
+    private String parentId;
+
+    private int maxRetries;
+
+    private int currentRetries;
+
+    private Throwable lastException;
+
+    /**
+     * Increase the current retries by 1
+     */
+    public void increaseCurrentRetries() {
+        this.currentRetries++;
+    }
+
+    /**
+     * Check if the current retries is greater than or equal to the max retries
+     * @return true if the current retries is greater than or equal to the max retries
+     */
+    public boolean isMaxRetriesReached() {
+        return this.currentRetries >= this.maxRetries;
+    }
+
+    /**
+     * Get the limited exception message with the default limit of 256
+     * @return the limited exception message
+     */
+    public String getLimitedExceptionMessage() {
+        return getLimitedExceptionMessage(256);
+    }
+
+    /**
+     * Get the limited exception message with the specified limit
+     * @param maxLimit the maximum limit of the exception message
+     * @return the limited exception message
+     */
+    public String getLimitedExceptionMessage(int maxLimit) {
+        if (lastException == null) {
+            return "";
+        }
+        String message = lastException.getMessage();
+        if (message == null) {
+            return "";
+        }
+        if (message.length() > maxLimit) {
+            return message.substring(0, maxLimit);
+        }
+        return message;
+    }
+
+}

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

@@ -0,0 +1,64 @@
+/*
+ * 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.data;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+
+/**
+ * Multi HTTP request context
+ */
+public class MultiHttpRequestContext {
+
+    public static final String NAME = "multi-http-request-context";
+
+    /**
+     * The remaining requests to be processed.
+     */
+    private final AtomicInteger remainingRequests;
+
+    /**
+     * The last failed event.
+     * If there are no retries or retries are not enabled, it will be null.
+     * If retries occur but still fail, it will be logged, and only the last one will be retained.
+     */
+    private HttpRetryEvent lastFailedEvent;
+
+    public MultiHttpRequestContext(int remainingEvents) {
+        this.remainingRequests = new AtomicInteger(remainingEvents);
+    }
+
+    /**
+     * Decrement the remaining requests by 1.
+     */
+    public void decrementRemainingRequests() {
+        remainingRequests.decrementAndGet();
+    }
+
+    public int getRemainingRequests() {
+        return remainingRequests.get();
+    }
+
+    public HttpRetryEvent getLastFailedEvent() {
+        return lastFailedEvent;
+    }
+
+    public void setLastFailedEvent(HttpRetryEvent lastFailedEvent) {
+        this.lastFailedEvent = lastFailedEvent;
+    }
+}