attributes = new ConcurrentHashMap<>();
+ attributes.put(MultiHttpRequestContext.NAME, new MultiHttpRequestContext(urls.size()));
+
+ // send the record to all URLs
+ for (URI url : urls) {
+ // convert ConnectRecord to HttpConnectRecord
+ String type = String.format("%s.%s.%s",
+ this.sinkConnectorConfig.getConnectorName(), url.getScheme(),
+ this.sinkConnectorConfig.getWebhookConfig().isActivate() ? "webhook" : "common");
+ HttpConnectRecord httpConnectRecord = HttpConnectRecord.convertConnectRecord(record, type);
+
+ // add retry event to attributes
+ HttpRetryEvent retryEvent = new HttpRetryEvent();
+ retryEvent.setMaxRetries(sinkConnectorConfig.getRetryConfig().getMaxRetries());
+ attributes.put(HttpRetryEvent.PREFIX + httpConnectRecord.getHttpRecordId(), retryEvent);
+
+ // deliver the record
+ deliver(url, httpConnectRecord, attributes, record);
+ }
+ }
+
+}
diff --git a/eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/sink/handle/HttpSinkHandler.java b/eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/sink/handler/HttpSinkHandler.java
similarity index 82%
rename from eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/sink/handle/HttpSinkHandler.java
rename to eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/sink/handler/HttpSinkHandler.java
index 09fd66a762..d5a27940e5 100644
--- a/eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/sink/handle/HttpSinkHandler.java
+++ b/eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/sink/handler/HttpSinkHandler.java
@@ -15,12 +15,13 @@
* limitations under the License.
*/
-package org.apache.eventmesh.connector.http.sink.handle;
+package org.apache.eventmesh.connector.http.sink.handler;
import org.apache.eventmesh.connector.http.sink.data.HttpConnectRecord;
import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
import java.net.URI;
+import java.util.Map;
import io.vertx.core.Future;
import io.vertx.core.buffer.Buffer;
@@ -32,14 +33,14 @@
*
* Any class that needs to process ConnectRecords via HTTP or HTTPS should implement this interface.
* Implementing classes must provide implementations for the {@link #start()}, {@link #handle(ConnectRecord)},
- * {@link #deliver(URI, HttpConnectRecord)}, and {@link #stop()} methods.
+ * {@link #deliver(URI, HttpConnectRecord, Map, ConnectRecord)}, and {@link #stop()} methods.
*
* Implementing classes should ensure thread safety and handle HTTP/HTTPS communication efficiently.
* The {@link #start()} method initializes any necessary resources for HTTP/HTTPS communication. The {@link #handle(ConnectRecord)} method processes a
- * ConnectRecord by sending it over HTTP or HTTPS. The {@link #deliver(URI, HttpConnectRecord)} method processes HttpConnectRecord on specified URL
- * while returning its own processing logic {@link #stop()} method releases any resources used for HTTP/HTTPS communication.
+ * ConnectRecord by sending it over HTTP or HTTPS. The {@link #deliver(URI, HttpConnectRecord, Map, ConnectRecord)} method processes HttpConnectRecord
+ * on specified URL while returning its own processing logic {@link #stop()} method releases any resources used for HTTP/HTTPS communication.
*
- * It's recommended to handle exceptions gracefully within the {@link #deliver(URI, HttpConnectRecord)} method
+ *
It's recommended to handle exceptions gracefully within the {@link #deliver(URI, HttpConnectRecord, Map, ConnectRecord)} method
* to prevent message loss or processing interruptions.
*/
public interface HttpSinkHandler {
@@ -62,9 +63,10 @@ public interface HttpSinkHandler {
*
* @param url URI to which the HttpConnectRecord should be sent
* @param httpConnectRecord HttpConnectRecord to process
+ * @param attributes additional attributes to be used in processing
* @return processing chain
*/
- Future> deliver(URI url, HttpConnectRecord httpConnectRecord);
+ Future> deliver(URI url, HttpConnectRecord httpConnectRecord, Map attributes, ConnectRecord connectRecord);
/**
* Cleans up and releases resources used by the HTTP/HTTPS handler. This method should be called when the handler is no longer needed.
diff --git a/eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/sink/handler/impl/CommonHttpSinkHandler.java b/eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/sink/handler/impl/CommonHttpSinkHandler.java
new file mode 100644
index 0000000000..e88707482f
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/sink/handler/impl/CommonHttpSinkHandler.java
@@ -0,0 +1,279 @@
+/*
+ * 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.handler.impl;
+
+import org.apache.eventmesh.common.config.connector.http.SinkConnectorConfig;
+import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.connector.http.sink.data.HttpConnectRecord;
+import org.apache.eventmesh.connector.http.sink.data.HttpRetryEvent;
+import org.apache.eventmesh.connector.http.sink.data.MultiHttpRequestContext;
+import org.apache.eventmesh.connector.http.sink.handler.AbstractHttpSinkHandler;
+import org.apache.eventmesh.connector.http.util.HttpUtils;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.callback.SendExceptionContext;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.callback.SendResult;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
+
+import java.net.URI;
+import java.time.ZoneId;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+import io.netty.handler.codec.http.HttpHeaderNames;
+import io.vertx.core.Future;
+import io.vertx.core.MultiMap;
+import io.vertx.core.Vertx;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.http.HttpHeaders;
+import io.vertx.ext.web.client.HttpResponse;
+import io.vertx.ext.web.client.WebClient;
+import io.vertx.ext.web.client.WebClientOptions;
+
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * Common HTTP/HTTPS Sink Handler implementation to handle ConnectRecords by sending them over HTTP or HTTPS to configured URLs.
+ *
+ * This handler initializes a WebClient for making HTTP requests based on the provided SinkConnectorConfig.
+ * It handles processing ConnectRecords by converting them to HttpConnectRecord and sending them asynchronously to each configured URL using the
+ * WebClient.
+ *
+ * The handler uses Vert.x's WebClient to perform HTTP/HTTPS requests. It initializes the WebClient in the {@link #start()}
+ * method and closes it in the {@link #stop()} method to manage resources efficiently.
+ *
+ * Each ConnectRecord is processed and sent to all configured URLs concurrently using asynchronous HTTP requests.
+ */
+@Slf4j
+@Getter
+public class CommonHttpSinkHandler extends AbstractHttpSinkHandler {
+
+ private WebClient webClient;
+
+
+ public CommonHttpSinkHandler(SinkConnectorConfig sinkConnectorConfig) {
+ super(sinkConnectorConfig);
+ }
+
+ /**
+ * Initializes the WebClient for making HTTP requests based on the provided SinkConnectorConfig.
+ */
+ @Override
+ public void start() {
+ // Create WebClient
+ doInitWebClient();
+ }
+
+ /**
+ * Initializes the WebClient with the provided configuration options.
+ */
+ private void doInitWebClient() {
+ SinkConnectorConfig sinkConnectorConfig = getSinkConnectorConfig();
+ final Vertx vertx = Vertx.vertx();
+ WebClientOptions options = new WebClientOptions()
+ .setKeepAlive(sinkConnectorConfig.isKeepAlive())
+ .setKeepAliveTimeout(sinkConnectorConfig.getKeepAliveTimeout() / 1000)
+ .setIdleTimeout(sinkConnectorConfig.getIdleTimeout())
+ .setIdleTimeoutUnit(TimeUnit.MILLISECONDS)
+ .setConnectTimeout(sinkConnectorConfig.getConnectionTimeout())
+ .setMaxPoolSize(sinkConnectorConfig.getMaxConnectionPoolSize());
+ this.webClient = WebClient.create(vertx, options);
+ }
+
+ /**
+ * Processes HttpConnectRecord on specified URL while returning its own processing logic. This method sends the HttpConnectRecord to the specified
+ * URL using the WebClient.
+ *
+ * @param url URI to which the HttpConnectRecord should be sent
+ * @param httpConnectRecord HttpConnectRecord to process
+ * @param attributes additional attributes to be used in processing
+ * @return processing chain
+ */
+ @Override
+ public Future> deliver(URI url, HttpConnectRecord httpConnectRecord, Map attributes,
+ ConnectRecord connectRecord) {
+ // create headers
+ Map extensionMap = new HashMap<>();
+ Set extensionKeySet = httpConnectRecord.getExtensions().keySet();
+ for (String extensionKey : extensionKeySet) {
+ Object v = httpConnectRecord.getExtensions().getObject(extensionKey);
+ extensionMap.put(extensionKey, v);
+ }
+
+ MultiMap headers = HttpHeaders.headers()
+ .set(HttpHeaderNames.CONTENT_TYPE, "application/json; charset=utf-8")
+ .set(HttpHeaderNames.ACCEPT, "application/json; charset=utf-8")
+ .set("extension", JsonUtils.toJSONString(extensionMap));
+ // get timestamp and offset
+ Long timestamp = httpConnectRecord.getCreateTime()
+ .atZone(ZoneId.systemDefault())
+ .toInstant()
+ .toEpochMilli();
+
+ // send the request
+ return this.webClient.post(url.getPath())
+ .host(url.getHost())
+ .port(url.getPort() == -1 ? (Objects.equals(url.getScheme(), "https") ? 443 : 80) : url.getPort())
+ .putHeaders(headers)
+ .ssl(Objects.equals(url.getScheme(), "https"))
+ .sendJson(httpConnectRecord.getData())
+ .onSuccess(res -> {
+ log.info("Request sent successfully. Record: timestamp={}", timestamp);
+
+ Exception e = null;
+
+ // log the response
+ if (HttpUtils.is2xxSuccessful(res.statusCode())) {
+ if (log.isDebugEnabled()) {
+ log.debug("Received successful response: statusCode={}. Record: timestamp={}, responseBody={}",
+ res.statusCode(), timestamp, res.bodyAsString());
+ } else {
+ log.info("Received successful response: statusCode={}. Record: timestamp={}", res.statusCode(), timestamp);
+ }
+ } else {
+ if (log.isDebugEnabled()) {
+ log.warn("Received non-2xx response: statusCode={}. Record: timestamp={}, responseBody={}",
+ res.statusCode(), timestamp, res.bodyAsString());
+ } else {
+ log.warn("Received non-2xx response: statusCode={}. Record: timestamp={}", res.statusCode(), timestamp);
+ }
+
+ e = new RuntimeException("Unexpected HTTP response code: " + res.statusCode());
+ }
+
+ // try callback
+ tryCallback(httpConnectRecord, e, attributes, connectRecord);
+ }).onFailure(err -> {
+ log.error("Request failed to send. Record: timestamp={}", timestamp, err);
+
+ // try callback
+ tryCallback(httpConnectRecord, err, attributes, connectRecord);
+ });
+ }
+
+ /**
+ * Tries to call the callback based on the result of the request.
+ *
+ * @param httpConnectRecord the HttpConnectRecord to use
+ * @param e the exception thrown during the request, may be null
+ * @param attributes additional attributes to be used in processing
+ */
+ private void tryCallback(HttpConnectRecord httpConnectRecord, Throwable e, Map attributes, ConnectRecord record) {
+ // get the retry event
+ HttpRetryEvent retryEvent = getAndUpdateRetryEvent(attributes, httpConnectRecord, e);
+
+ // get the multi http request context
+ MultiHttpRequestContext multiHttpRequestContext = getAndUpdateMultiHttpRequestContext(attributes, retryEvent);
+
+ if (multiHttpRequestContext.getRemainingRequests() == 0) {
+ // do callback
+ if (record.getCallback() == null) {
+ if (log.isDebugEnabled()) {
+ log.warn("ConnectRecord callback is null. Ignoring callback. {}", record);
+ } else {
+ log.warn("ConnectRecord callback is null. Ignoring callback.");
+ }
+ return;
+ }
+
+ HttpRetryEvent lastFailedEvent = multiHttpRequestContext.getLastFailedEvent();
+ if (lastFailedEvent == null) {
+ // success
+ record.getCallback().onSuccess(convertToSendResult(record));
+ } else {
+ // failure
+ record.getCallback().onException(buildSendExceptionContext(record, lastFailedEvent.getLastException()));
+ }
+ }
+ }
+
+ /**
+ * Gets and updates the retry event based on the provided attributes and HttpConnectRecord.
+ *
+ * @param attributes the attributes to use
+ * @param httpConnectRecord the HttpConnectRecord to use
+ * @param e the exception thrown during the request, may be null
+ * @return the updated retry event
+ */
+ private HttpRetryEvent getAndUpdateRetryEvent(Map attributes, HttpConnectRecord httpConnectRecord, Throwable e) {
+ // get the retry event
+ HttpRetryEvent retryEvent = (HttpRetryEvent) attributes.get(HttpRetryEvent.PREFIX + httpConnectRecord.getHttpRecordId());
+ // update the retry event
+ retryEvent.setLastException(e);
+ return retryEvent;
+ }
+
+
+ /**
+ * Gets and updates the multi http request context based on the provided attributes and HttpConnectRecord.
+ *
+ * @param attributes the attributes to use
+ * @param retryEvent the retry event to use
+ * @return the updated multi http request context
+ */
+ private MultiHttpRequestContext getAndUpdateMultiHttpRequestContext(Map attributes, HttpRetryEvent retryEvent) {
+ // get the multi http request context
+ MultiHttpRequestContext multiHttpRequestContext = (MultiHttpRequestContext) attributes.get(MultiHttpRequestContext.NAME);
+
+ if (retryEvent.getLastException() == null || retryEvent.isMaxRetriesReached()) {
+ // decrement the counter
+ multiHttpRequestContext.decrementRemainingRequests();
+
+ // try set failed event
+ if (retryEvent.getLastException() != null) {
+ multiHttpRequestContext.setLastFailedEvent(retryEvent);
+ }
+ }
+
+ return multiHttpRequestContext;
+ }
+
+ private SendResult convertToSendResult(ConnectRecord record) {
+ SendResult result = new SendResult();
+ result.setMessageId(record.getRecordId());
+ if (org.apache.commons.lang3.StringUtils.isNotEmpty(record.getExtension("topic"))) {
+ result.setTopic(record.getExtension("topic"));
+ }
+ return result;
+ }
+
+ private SendExceptionContext buildSendExceptionContext(ConnectRecord record, Throwable e) {
+ SendExceptionContext sendExceptionContext = new SendExceptionContext();
+ sendExceptionContext.setMessageId(record.getRecordId());
+ sendExceptionContext.setCause(e);
+ if (org.apache.commons.lang3.StringUtils.isNotEmpty(record.getExtension("topic"))) {
+ sendExceptionContext.setTopic(record.getExtension("topic"));
+ }
+ return sendExceptionContext;
+ }
+
+
+ /**
+ * Cleans up and releases resources used by the HTTP/HTTPS handler.
+ */
+ @Override
+ public void stop() {
+ if (this.webClient != null) {
+ this.webClient.close();
+ } else {
+ log.warn("WebClient is null, ignore.");
+ }
+ }
+}
\ No newline at end of file
diff --git a/eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/sink/handler/impl/HttpSinkHandlerRetryWrapper.java b/eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/sink/handler/impl/HttpSinkHandlerRetryWrapper.java
new file mode 100644
index 0000000000..820b46296a
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/sink/handler/impl/HttpSinkHandlerRetryWrapper.java
@@ -0,0 +1,122 @@
+/*
+ * 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.handler.impl;
+
+import org.apache.eventmesh.common.config.connector.http.HttpRetryConfig;
+import org.apache.eventmesh.common.config.connector.http.SinkConnectorConfig;
+import org.apache.eventmesh.connector.http.sink.data.HttpConnectRecord;
+import org.apache.eventmesh.connector.http.sink.data.HttpRetryEvent;
+import org.apache.eventmesh.connector.http.sink.handler.AbstractHttpSinkHandler;
+import org.apache.eventmesh.connector.http.sink.handler.HttpSinkHandler;
+import org.apache.eventmesh.connector.http.util.HttpUtils;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
+
+import java.net.ConnectException;
+import java.net.URI;
+import java.time.Duration;
+import java.util.Map;
+
+import io.vertx.core.Future;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.ext.web.client.HttpResponse;
+
+import lombok.extern.slf4j.Slf4j;
+
+import dev.failsafe.Failsafe;
+import dev.failsafe.RetryPolicy;
+
+
+/**
+ * HttpSinkHandlerRetryWrapper is a wrapper class for the HttpSinkHandler that provides retry functionality for failed HTTP requests.
+ */
+@Slf4j
+public class HttpSinkHandlerRetryWrapper extends AbstractHttpSinkHandler {
+
+ private final HttpRetryConfig httpRetryConfig;
+
+ private final HttpSinkHandler sinkHandler;
+
+ public HttpSinkHandlerRetryWrapper(SinkConnectorConfig sinkConnectorConfig, HttpSinkHandler sinkHandler) {
+ super(sinkConnectorConfig);
+ this.sinkHandler = sinkHandler;
+ this.httpRetryConfig = getSinkConnectorConfig().getRetryConfig();
+ }
+
+ /**
+ * Initializes the WebClient for making HTTP requests based on the provided SinkConnectorConfig.
+ */
+ @Override
+ public void start() {
+ sinkHandler.start();
+ }
+
+
+ /**
+ * Processes HttpConnectRecord on specified URL while returning its own processing logic This method provides the retry power to process the
+ * HttpConnectRecord
+ *
+ * @param url URI to which the HttpConnectRecord should be sent
+ * @param httpConnectRecord HttpConnectRecord to process
+ * @param attributes additional attributes to pass to the processing chain
+ * @return processing chain
+ */
+ @Override
+ public Future> deliver(URI url, HttpConnectRecord httpConnectRecord, Map attributes,
+ ConnectRecord connectRecord) {
+
+ // Build the retry policy
+ RetryPolicy> retryPolicy = RetryPolicy.>builder()
+ .handleIf(e -> e instanceof ConnectException)
+ .handleResultIf(response -> httpRetryConfig.isRetryOnNonSuccess() && !HttpUtils.is2xxSuccessful(response.statusCode()))
+ .withMaxRetries(httpRetryConfig.getMaxRetries())
+ .withDelay(Duration.ofMillis(httpRetryConfig.getInterval()))
+ .onRetry(event -> {
+ if (log.isDebugEnabled()) {
+ log.warn("Retrying the request to {} for the {} time. {}", url, event.getAttemptCount(), httpConnectRecord);
+ } else {
+ log.warn("Retrying the request to {} for the {} time.", url, event.getAttemptCount());
+ }
+ // update the retry event
+ HttpRetryEvent retryEvent = (HttpRetryEvent) attributes.get(HttpRetryEvent.PREFIX + httpConnectRecord.getHttpRecordId());
+ retryEvent.increaseCurrentRetries();
+ })
+ .onFailure(event -> {
+ if (log.isDebugEnabled()) {
+ log.error("Failed to send the request to {} after {} attempts. {}", url, event.getAttemptCount(),
+ httpConnectRecord, event.getException());
+ } else {
+ log.error("Failed to send the request to {} after {} attempts.", url, event.getAttemptCount(), event.getException());
+ }
+ }).build();
+
+ // Handle the ConnectRecord with retry policy
+ Failsafe.with(retryPolicy)
+ .getStageAsync(() -> sinkHandler.deliver(url, httpConnectRecord, attributes, connectRecord).toCompletionStage());
+
+ return null;
+ }
+
+
+ /**
+ * Cleans up and releases resources used by the HTTP/HTTPS handler.
+ */
+ @Override
+ public void stop() {
+ sinkHandler.stop();
+ }
+}
diff --git a/eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/sink/handle/WebhookHttpSinkHandler.java b/eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/sink/handler/impl/WebhookHttpSinkHandler.java
similarity index 82%
rename from eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/sink/handle/WebhookHttpSinkHandler.java
rename to eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/sink/handler/impl/WebhookHttpSinkHandler.java
index 4e64126a9d..7edd84a967 100644
--- a/eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/sink/handle/WebhookHttpSinkHandler.java
+++ b/eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/sink/handler/impl/WebhookHttpSinkHandler.java
@@ -15,16 +15,17 @@
* limitations under the License.
*/
-package org.apache.eventmesh.connector.http.sink.handle;
+package org.apache.eventmesh.connector.http.sink.handler.impl;
+import org.apache.eventmesh.common.config.connector.http.HttpWebhookConfig;
+import org.apache.eventmesh.common.config.connector.http.SinkConnectorConfig;
import org.apache.eventmesh.common.exception.EventMeshException;
import org.apache.eventmesh.connector.http.common.SynchronizedCircularFifoQueue;
-import org.apache.eventmesh.connector.http.sink.config.HttpWebhookConfig;
-import org.apache.eventmesh.connector.http.sink.config.SinkConnectorConfig;
import org.apache.eventmesh.connector.http.sink.data.HttpConnectRecord;
import org.apache.eventmesh.connector.http.sink.data.HttpExportMetadata;
import org.apache.eventmesh.connector.http.sink.data.HttpExportRecord;
import org.apache.eventmesh.connector.http.sink.data.HttpExportRecordPage;
+import org.apache.eventmesh.connector.http.sink.data.HttpRetryEvent;
import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
import org.apache.commons.lang3.StringUtils;
@@ -32,6 +33,7 @@
import java.net.URI;
import java.time.LocalDateTime;
import java.util.List;
+import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
@@ -61,8 +63,6 @@
@Slf4j
public class WebhookHttpSinkHandler extends CommonHttpSinkHandler {
- private final SinkConnectorConfig sinkConnectorConfig;
-
// the configuration for webhook
private final HttpWebhookConfig webhookConfig;
@@ -86,7 +86,7 @@ public boolean isExportDestroyed() {
public WebhookHttpSinkHandler(SinkConnectorConfig sinkConnectorConfig) {
super(sinkConnectorConfig);
- this.sinkConnectorConfig = sinkConnectorConfig;
+
this.webhookConfig = sinkConnectorConfig.getWebhookConfig();
int maxQueueSize = this.webhookConfig.getMaxStorageSize();
this.receivedDataQueue = new SynchronizedCircularFifoQueue<>(maxQueueSize);
@@ -94,9 +94,6 @@ public WebhookHttpSinkHandler(SinkConnectorConfig sinkConnectorConfig) {
doInitExportServer();
}
- public SynchronizedCircularFifoQueue getReceivedDataQueue() {
- return receivedDataQueue;
- }
/**
* Initialize the server for exporting the received data
@@ -202,22 +199,6 @@ public void start() {
});
}
- /**
- * Processes a ConnectRecord by sending it over HTTP or HTTPS. This method should be called for each ConnectRecord that needs to be processed.
- *
- * @param record the ConnectRecord to process
- */
- @Override
- public void handle(ConnectRecord record) {
- for (URI url : super.getUrls()) {
- // convert ConnectRecord to HttpConnectRecord
- String type = String.format("%s.%s.%s", this.getConnectorConfig().getConnectorName(), url.getScheme(), "webhook");
- HttpConnectRecord httpConnectRecord = HttpConnectRecord.convertConnectRecord(record, type);
- // handle the HttpConnectRecord
- deliver(url, httpConnectRecord);
- }
- }
-
/**
* Processes HttpConnectRecord on specified URL while returning its own processing logic This method sends the HttpConnectRecord to the specified
@@ -225,30 +206,28 @@ public void handle(ConnectRecord record) {
*
* @param url URI to which the HttpConnectRecord should be sent
* @param httpConnectRecord HttpConnectRecord to process
+ * @param attributes additional attributes to be used in processing
* @return processing chain
*/
@Override
- public Future> deliver(URI url, HttpConnectRecord httpConnectRecord) {
+ public Future> deliver(URI url, HttpConnectRecord httpConnectRecord, Map attributes,
+ ConnectRecord connectRecord) {
// send the request
- Future> responseFuture = super.deliver(url, httpConnectRecord);
+ Future> responseFuture = super.deliver(url, httpConnectRecord, attributes, connectRecord);
// store the received data
return responseFuture.onComplete(arr -> {
- // If open retry, return directly and handled by RetryHttpSinkHandler
- if (sinkConnectorConfig.getRetryConfig().getMaxRetries() > 0) {
- return;
+ // get tryEvent from attributes
+ HttpRetryEvent retryEvent = (HttpRetryEvent) attributes.get(HttpRetryEvent.PREFIX + httpConnectRecord.getHttpRecordId());
+
+ HttpResponse response = null;
+ if (arr.succeeded()) {
+ response = arr.result();
+ } else {
+ retryEvent.setLastException(arr.cause());
}
- // create ExportMetadataBuilder
- HttpResponse response = arr.succeeded() ? arr.result() : null;
-
- HttpExportMetadata httpExportMetadata = HttpExportMetadata.builder()
- .url(url.toString())
- .code(response != null ? response.statusCode() : -1)
- .message(response != null ? response.statusMessage() : arr.cause().getMessage())
- .receivedTime(LocalDateTime.now())
- .retriedBy(null)
- .uuid(httpConnectRecord.getUuid())
- .retryNum(0)
- .build();
+
+ // create ExportMetadata
+ HttpExportMetadata httpExportMetadata = buildHttpExportMetadata(url, response, httpConnectRecord, retryEvent);
// create ExportRecord
HttpExportRecord exportRecord = new HttpExportRecord(httpExportMetadata, arr.succeeded() ? arr.result().bodyAsString() : null);
@@ -257,6 +236,37 @@ public Future> deliver(URI url, HttpConnectRecord httpConne
});
}
+ /**
+ * Builds the HttpExportMetadata object based on the response, HttpConnectRecord, and HttpRetryEvent.
+ *
+ * @param url the URI to which the HttpConnectRecord was sent
+ * @param response the response received from the URI
+ * @param httpConnectRecord the HttpConnectRecord that was sent
+ * @param retryEvent the SingleHttpRetryEvent that was used for retries
+ * @return the HttpExportMetadata object
+ */
+ private HttpExportMetadata buildHttpExportMetadata(URI url, HttpResponse response, HttpConnectRecord httpConnectRecord,
+ HttpRetryEvent retryEvent) {
+
+ String msg = null;
+ // order of precedence: lastException > response > null
+ if (retryEvent.getLastException() != null) {
+ msg = retryEvent.getLimitedExceptionMessage();
+ retryEvent.setLastException(null);
+ } else if (response != null) {
+ msg = response.statusMessage();
+ }
+
+ return HttpExportMetadata.builder()
+ .url(url.toString())
+ .code(response != null ? response.statusCode() : -1)
+ .message(msg)
+ .receivedTime(LocalDateTime.now())
+ .recordId(httpConnectRecord.getHttpRecordId())
+ .retryNum(retryEvent.getCurrentRetries())
+ .build();
+ }
+
/**
* Cleans up and releases resources used by the HTTP/HTTPS handler.
diff --git a/eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/source/HttpSourceConnector.java b/eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/source/HttpSourceConnector.java
index 4155aff910..2b2a01a9dd 100644
--- a/eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/source/HttpSourceConnector.java
+++ b/eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/source/HttpSourceConnector.java
@@ -34,6 +34,7 @@
import java.util.List;
import java.util.concurrent.TimeUnit;
+import io.netty.handler.codec.http.HttpResponseStatus;
import io.vertx.core.Vertx;
import io.vertx.core.http.HttpServer;
import io.vertx.core.http.HttpServerOptions;
@@ -41,6 +42,7 @@
import io.vertx.ext.web.Router;
import io.vertx.ext.web.handler.LoggerHandler;
+import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
@Slf4j
@@ -52,22 +54,18 @@ public class HttpSourceConnector implements Source, ConnectorCreateService configClass() {
@@ -106,7 +104,7 @@ private void doInit() {
final Vertx vertx = Vertx.vertx();
final Router router = Router.router(vertx);
- final Route route = router.route()
+ route = router.route()
.path(this.sourceConfig.connectorConfig.getPath())
.handler(LoggerHandler.create());
@@ -136,7 +134,15 @@ public void start() {
@Override
public void commit(ConnectRecord record) {
-
+ if (this.route != null && sourceConfig.getConnectorConfig().isDataConsistencyEnabled()) {
+ this.route.handler(ctx -> {
+ // Return 200 OK
+ ctx.response()
+ .putHeader("content-type", "application/json")
+ .setStatusCode(HttpResponseStatus.OK.code())
+ .end("{\"status\":\"success\",\"recordId\":\"" + record.getRecordId() + "\"}");
+ });
+ }
}
@Override
@@ -144,6 +150,19 @@ public String name() {
return this.sourceConfig.getConnectorConfig().getConnectorName();
}
+ @Override
+ public void onException(ConnectRecord record) {
+ if (this.route != null) {
+ this.route.failureHandler(ctx -> {
+ log.error("Failed to handle the request, recordId {}. ", record.getRecordId(), ctx.failure());
+ // Return Bad Response
+ ctx.response()
+ .setStatusCode(HttpResponseStatus.INTERNAL_SERVER_ERROR.code())
+ .end("{\"status\":\"failed\",\"recordId\":\"" + record.getRecordId() + "\"}");
+ });
+ }
+ }
+
@Override
public void stop() {
if (this.server != null) {
diff --git a/eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/source/data/WebhookRequest.java b/eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/source/data/WebhookRequest.java
index 2fe7399da2..9e1dcb7b4c 100644
--- a/eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/source/data/WebhookRequest.java
+++ b/eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/source/data/WebhookRequest.java
@@ -20,6 +20,8 @@
import java.io.Serializable;
import java.util.Map;
+import io.vertx.ext.web.RoutingContext;
+
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
@@ -42,4 +44,6 @@ public class WebhookRequest implements Serializable {
private Object payload;
+ private RoutingContext routingContext;
+
}
diff --git a/eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/source/protocol/impl/CommonProtocol.java b/eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/source/protocol/impl/CommonProtocol.java
index 80e4f0a753..0761170ac0 100644
--- a/eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/source/protocol/impl/CommonProtocol.java
+++ b/eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/source/protocol/impl/CommonProtocol.java
@@ -19,21 +19,23 @@
import org.apache.eventmesh.common.Constants;
import org.apache.eventmesh.common.config.connector.http.SourceConnectorConfig;
+import org.apache.eventmesh.common.utils.JsonUtils;
import org.apache.eventmesh.connector.http.common.SynchronizedCircularFifoQueue;
import org.apache.eventmesh.connector.http.source.data.CommonResponse;
import org.apache.eventmesh.connector.http.source.data.WebhookRequest;
import org.apache.eventmesh.connector.http.source.protocol.Protocol;
import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
+import java.util.Base64;
import java.util.Map;
import java.util.stream.Collectors;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.vertx.core.http.HttpMethod;
+import io.vertx.core.json.JsonObject;
import io.vertx.ext.web.Route;
import io.vertx.ext.web.handler.BodyHandler;
-
import lombok.extern.slf4j.Slf4j;
/**
@@ -45,6 +47,8 @@ public class CommonProtocol implements Protocol {
public static final String PROTOCOL_NAME = "Common";
+ private SourceConnectorConfig sourceConnectorConfig;
+
/**
* Initialize the protocol
*
@@ -52,7 +56,7 @@ public class CommonProtocol implements Protocol {
*/
@Override
public void initialize(SourceConnectorConfig sourceConnectorConfig) {
-
+ this.sourceConnectorConfig = sourceConnectorConfig;
}
/**
@@ -67,20 +71,24 @@ public void setHandler(Route route, SynchronizedCircularFifoQueue