[ISSUE apache#5052] Enhancement for source\sink connector

xwm1992 · Jul 31, 2024 · 0349768 · 0349768
1 parent 53b7b8c
commit 0349768
Show file tree

Hide file tree

Showing 46 changed files with 301 additions and 38 deletions.
diff --git a/...src/main/java/org/apache/eventmesh/connector/canal/sink/connector/CanalSinkConnector.java b/...src/main/java/org/apache/eventmesh/connector/canal/sink/connector/CanalSinkConnector.java
@@ -38,6 +38,8 @@
 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.callback.SendExceptionContext;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.callback.SendResult;
 import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
 
 import org.apache.commons.lang.StringUtils;
@@ -146,6 +148,11 @@ public String name() {
         return this.sinkConfig.getSinkConnectorConfig().getConnectorName();
     }
 
+    @Override
+    public void onException(ConnectRecord record) {
+
+    }
+
     @Override
     public void stop() {
         executor.shutdown();
@@ -159,7 +166,7 @@ public void put(List<ConnectRecord> sinkRecords) {
             List<CanalConnectRecord> canalConnectRecordList = (List<CanalConnectRecord>) connectRecord.getData();
             canalConnectRecordList = filterRecord(canalConnectRecordList);
             if (isDdlDatas(canalConnectRecordList)) {
-                doDdl(context, canalConnectRecordList);
+                doDdl(context, canalConnectRecordList, connectRecord);
             } else if (sinkConfig.isGTIDMode()) {
                 doLoadWithGtid(context, sinkConfig, connectRecord);
             } else {
@@ -197,7 +204,7 @@ private List<CanalConnectRecord> filterRecord(List<CanalConnectRecord> canalConn
             .collect(Collectors.toList());
     }
 
-    private void doDdl(DbLoadContext context, List<CanalConnectRecord> canalConnectRecordList) {
+    private void doDdl(DbLoadContext context, List<CanalConnectRecord> canalConnectRecordList, ConnectRecord connectRecord) {
         for (final CanalConnectRecord record : canalConnectRecordList) {
             try {
                 Boolean result = jdbcTemplate.execute(new StatementCallback<Boolean>() {
@@ -217,9 +224,30 @@ public Boolean doInStatement(Statement stmt) throws SQLException, DataAccessExce
                     context.getFailedRecords().add(record);
                 }
             } catch (Throwable e) {
+                connectRecord.getCallback().onException(buildSendExceptionContext(connectRecord, e));
                 throw new RuntimeException(e);
             }
         }
+        connectRecord.getCallback().onSuccess(convertToSendResult(connectRecord));
+    }
+
+    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;
+    }
+
+    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 void doBefore(List<CanalConnectRecord> canalConnectRecordList, final DbLoadData loadData) {
@@ -291,21 +319,26 @@ private void doLoadWithGtid(DbLoadContext context, CanalSinkConfig sinkConfig, C
             Exception ex = null;
             try {
                 ex = result.get();
+                if (ex == null) {
+                    connectRecord.getCallback().onSuccess(convertToSendResult(connectRecord));
+                }
             } catch (Exception e) {
                 ex = e;
             }
             Boolean skipException = sinkConfig.getSkipException();
             if (skipException != null && skipException) {
                 if (ex != null) {
                     // do skip
-                    log.warn("skip exception for data : {} , caused by {}",
+                    log.warn("skip exception will ack data : {} , caused by {}",
                         filteredRows,
                         ExceptionUtils.getFullStackTrace(ex));
                     GtidBatchManager.removeGtidBatch(gtid);
+                    connectRecord.getCallback().onSuccess(convertToSendResult(connectRecord));
                 }
             } else {
                 if (ex != null) {
                     log.error("sink connector will shutdown by " + ex.getMessage(), ExceptionUtils.getFullStackTrace(ex));
+                    connectRecord.getCallback().onException(buildSendExceptionContext(connectRecord, ex));
                     gtidSingleExecutor.shutdown();
                     System.exit(1);
                 } else {
@@ -314,6 +347,8 @@ private void doLoadWithGtid(DbLoadContext context, CanalSinkConfig sinkConfig, C
             }
         } else {
             log.info("Batch received, waiting for other batches.");
+            // ack this record
+            connectRecord.getCallback().onSuccess(convertToSendResult(connectRecord));
         }
     }
 

diff --git a/...main/java/org/apache/eventmesh/connector/canal/sink/connector/CanalSinkFullConnector.java b/...main/java/org/apache/eventmesh/connector/canal/sink/connector/CanalSinkFullConnector.java
@@ -109,6 +109,11 @@ public String name() {
         return null;
     }
 
+    @Override
+    public void onException(ConnectRecord record) {
+
+    }
+
     @Override
     public void put(List<ConnectRecord> sinkRecords) {
         if (sinkRecords == null || sinkRecords.isEmpty() || sinkRecords.get(0) == null) {

diff --git a/...main/java/org/apache/eventmesh/connector/canal/source/connector/CanalSourceConnector.java b/...main/java/org/apache/eventmesh/connector/canal/source/connector/CanalSourceConnector.java
@@ -267,6 +267,11 @@ public String name() {
         return this.sourceConfig.getSourceConnectorConfig().getConnectorName();
     }
 
+    @Override
+    public void onException(ConnectRecord record) {
+
+    }
+
     @Override
     public void stop() {
         if (!running) {

diff --git a/.../java/org/apache/eventmesh/connector/canal/source/connector/CanalSourceFullConnector.java b/.../java/org/apache/eventmesh/connector/canal/source/connector/CanalSourceFullConnector.java
@@ -159,6 +159,11 @@ public String name() {
         return this.config.getConnectorConfig().getConnectorName();
     }
 
+    @Override
+    public void onException(ConnectRecord record) {
+
+    }
+
     @Override
     public List<ConnectRecord> poll() {
         while (flag.get()) {

diff --git a/.../java/org/apache/eventmesh/connector/chatgpt/source/connector/ChatGPTSourceConnector.java b/.../java/org/apache/eventmesh/connector/chatgpt/source/connector/ChatGPTSourceConnector.java
@@ -224,6 +224,11 @@ public String name() {
         return this.sourceConfig.getConnectorConfig().getConnectorName();
     }
 
+    @Override
+    public void onException(ConnectRecord record) {
+
+    }
+
     @Override
     public void stop() {
         Throwable t = this.server.close().cause();

diff --git a/...in/java/org/apache/eventmesh/connector/dingtalk/sink/connector/DingDingSinkConnector.java b/...in/java/org/apache/eventmesh/connector/dingtalk/sink/connector/DingDingSinkConnector.java
@@ -103,6 +103,11 @@ public String name() {
         return this.sinkConfig.getSinkConnectorConfig().getConnectorName();
     }
 
+    @Override
+    public void onException(ConnectRecord record) {
+
+    }
+
     @Override
     public void stop() {
         isRunning = false;

diff --git a/...e/src/main/java/org/apache/eventmesh/connector/file/sink/connector/FileSinkConnector.java b/...e/src/main/java/org/apache/eventmesh/connector/file/sink/connector/FileSinkConnector.java
@@ -103,6 +103,11 @@ public String name() {
         return this.sinkConfig.getConnectorConfig().getConnectorName();
     }
 
+    @Override
+    public void onException(ConnectRecord record) {
+
+    }
+
     @Override
     public void stop() {
         outputStream.flush();

diff --git a/...c/main/java/org/apache/eventmesh/connector/file/source/connector/FileSourceConnector.java b/...c/main/java/org/apache/eventmesh/connector/file/source/connector/FileSourceConnector.java
@@ -86,6 +86,11 @@ public String name() {
         return this.sourceConfig.getConnectorConfig().getConnectorName();
     }
 
+    @Override
+    public void onException(ConnectRecord record) {
+
+    }
+
     @Override
     public void stop() {
         try {

diff --git a/...nector-http/src/main/java/org/apache/eventmesh/connector/http/sink/HttpSinkConnector.java b/...nector-http/src/main/java/org/apache/eventmesh/connector/http/sink/HttpSinkConnector.java
@@ -107,6 +107,11 @@ public String name() {
         return this.httpSinkConfig.connectorConfig.getConnectorName();
     }
 
+    @Override
+    public void onException(ConnectRecord record) {
+
+    }
+
     @Override
     public void stop() throws Exception {
         this.sinkHandler.stop();

diff --git a/...or-http/src/main/java/org/apache/eventmesh/connector/http/source/HttpSourceConnector.java b/...or-http/src/main/java/org/apache/eventmesh/connector/http/source/HttpSourceConnector.java
@@ -144,6 +144,11 @@ public String name() {
         return this.sourceConfig.getConnectorConfig().getConnectorName();
     }
 
+    @Override
+    public void onException(ConnectRecord record) {
+
+    }
+
     @Override
     public void stop() {
         if (this.server != null) {

diff --git a/...nector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/sink/JdbcSinkConnector.java b/...nector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/sink/JdbcSinkConnector.java
@@ -139,6 +139,11 @@ public String name() {
         return this.sinkConfig.getSinkConnectorConfig().getConnectorName();
     }
 
+    @Override
+    public void onException(ConnectRecord record) {
+
+    }
+
     /**
      * Stops the Connector.
      *

diff --git a/...or-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/JdbcSourceConnector.java b/...or-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/JdbcSourceConnector.java
@@ -192,6 +192,11 @@ public String name() {
         return "JDBC Source Connector";
     }
 
+    @Override
+    public void onException(ConnectRecord record) {
+
+    }
+
     /**
      * Stops the Connector.
      *

diff --git a/...src/main/java/org/apache/eventmesh/connector/kafka/sink/connector/KafkaSinkConnector.java b/...src/main/java/org/apache/eventmesh/connector/kafka/sink/connector/KafkaSinkConnector.java
@@ -94,6 +94,11 @@ public String name() {
         return this.sinkConfig.getConnectorConfig().getConnectorName();
     }
 
+    @Override
+    public void onException(ConnectRecord record) {
+
+    }
+
     @Override
     public void stop() {
         producer.close();

diff --git a/...main/java/org/apache/eventmesh/connector/kafka/source/connector/KafkaSourceConnector.java b/...main/java/org/apache/eventmesh/connector/kafka/source/connector/KafkaSourceConnector.java
@@ -94,6 +94,11 @@ public String name() {
         return this.sourceConfig.getConnectorConfig().getConnectorName();
     }
 
+    @Override
+    public void onException(ConnectRecord record) {
+
+    }
+
     @Override
     public void stop() {
         kafkaConsumer.unsubscribe();

diff --git a/...main/java/org/apache/eventmesh/connector/knative/sink/connector/KnativeSinkConnector.java b/...main/java/org/apache/eventmesh/connector/knative/sink/connector/KnativeSinkConnector.java
@@ -82,6 +82,11 @@ public String name() {
         return this.sinkConfig.getConnectorConfig().getConnectorName();
     }
 
+    @Override
+    public void onException(ConnectRecord record) {
+
+    }
+
     @Override
     public void stop() {
         started.compareAndSet(true, false);

diff --git a/.../java/org/apache/eventmesh/connector/knative/source/connector/KnativeSourceConnector.java b/.../java/org/apache/eventmesh/connector/knative/source/connector/KnativeSourceConnector.java
@@ -65,6 +65,11 @@ public String name() {
         return this.sourceConfig.getConnectorConfig().getConnectorName();
     }
 
+    @Override
+    public void onException(ConnectRecord record) {
+
+    }
+
     @Override
     public void stop() {
         started.compareAndSet(true, false);

diff --git a/...k/src/main/java/org/apache/eventmesh/connector/lark/sink/connector/LarkSinkConnector.java b/...k/src/main/java/org/apache/eventmesh/connector/lark/sink/connector/LarkSinkConnector.java
@@ -110,6 +110,11 @@ public String name() {
         return this.sinkConfig.getSinkConnectorConfig().getConnectorName();
     }
 
+    @Override
+    public void onException(ConnectRecord record) {
+
+    }
+
     @Override
     public void stop() {
         if (!started.compareAndSet(true, false)) {

diff --git a/...main/java/org/apache/eventmesh/connector/mongodb/sink/connector/MongodbSinkConnector.java b/...main/java/org/apache/eventmesh/connector/mongodb/sink/connector/MongodbSinkConnector.java
@@ -87,6 +87,11 @@ public String name() {
         return this.sinkConfig.getConnectorConfig().getConnectorName();
     }
 
+    @Override
+    public void onException(ConnectRecord record) {
+
+    }
+
     @Override
     public void stop() throws Exception {
         this.client.stop();

diff --git a/.../java/org/apache/eventmesh/connector/mongodb/source/connector/MongodbSourceConnector.java b/.../java/org/apache/eventmesh/connector/mongodb/source/connector/MongodbSourceConnector.java
@@ -93,6 +93,11 @@ public String name() {
         return this.sourceConfig.connectorConfig.getConnectorName();
     }
 
+    @Override
+    public void onException(ConnectRecord record) {
+
+    }
+
     @Override
     public void stop() throws Exception {
         this.client.stop();

diff --git a/...org/apache/eventmesh/connector/openfunction/sink/connector/OpenFunctionSinkConnector.java b/...org/apache/eventmesh/connector/openfunction/sink/connector/OpenFunctionSinkConnector.java
@@ -74,6 +74,11 @@ public String name() {
         return this.sinkConfig.getSinkConnectorConfig().getConnectorName();
     }
 
+    @Override
+    public void onException(ConnectRecord record) {
+
+    }
+
     @Override
     public void stop() {
     }

diff --git a/...apache/eventmesh/connector/openfunction/source/connector/OpenFunctionSourceConnector.java b/...apache/eventmesh/connector/openfunction/source/connector/OpenFunctionSourceConnector.java
@@ -76,6 +76,11 @@ public String name() {
         return this.sourceConfig.getSourceConnectorConfig().getConnectorName();
     }
 
+    @Override
+    public void onException(ConnectRecord record) {
+
+    }
+
     @Override
     public void stop() {
 

diff --git a/...main/java/org/apache/eventmesh/connector/pravega/sink/connector/PravegaSinkConnector.java b/...main/java/org/apache/eventmesh/connector/pravega/sink/connector/PravegaSinkConnector.java
@@ -109,6 +109,11 @@ public String name() {
         return this.sinkConfig.getConnectorConfig().getConnectorName();
     }
 
+    @Override
+    public void onException(ConnectRecord record) {
+
+    }
+
     @Override
     public void stop() {
         writerMap.forEach((topic, writer) -> writer.close());

diff --git a/.../java/org/apache/eventmesh/connector/pravega/source/connector/PravegaSourceConnector.java b/.../java/org/apache/eventmesh/connector/pravega/source/connector/PravegaSourceConnector.java
@@ -148,6 +148,11 @@ public String name() {
         return this.sourceConfig.getConnectorConfig().getConnectorName();
     }
 
+    @Override
+    public void onException(ConnectRecord record) {
+
+    }
+
     @Override
     public void stop() {
         sourceHandlerMap.forEach((topic, handler) -> {

diff --git a/...org/apache/eventmesh/connector/prometheus/source/connector/PrometheusSourceConnector.java b/...org/apache/eventmesh/connector/prometheus/source/connector/PrometheusSourceConnector.java
@@ -145,6 +145,11 @@ public String name() {
         return this.sourceConfig.getConnectorConfig().getConnectorName();
     }
 
+    @Override
+    public void onException(ConnectRecord record) {
+
+    }
+
     @Override
     public void stop() {
         log.info("prometheus source connector stop.");

diff --git a/...c/main/java/org/apache/eventmesh/connector/pulsar/sink/connector/PulsarSinkConnector.java b/...c/main/java/org/apache/eventmesh/connector/pulsar/sink/connector/PulsarSinkConnector.java
@@ -85,6 +85,11 @@ public String name() {
         return this.sinkConfig.getConnectorConfig().getConnectorName();
     }
 
+    @Override
+    public void onException(ConnectRecord record) {
+
+    }
+
     @Override
     public void stop() {
         try {