[ISSUE #5081] Enhancement update for connectors & admin-server

eventmesh-admin-server/src/main/java/org/apache/eventmesh/admin/server/web/service/job/JobInfoBizService.java

@@ -37,6 +37,7 @@
 import org.apache.eventmesh.common.remote.datasource.DataSource;
 import org.apache.eventmesh.common.remote.datasource.DataSourceType;
 import org.apache.eventmesh.common.remote.exception.ErrorCode;
+import org.apache.eventmesh.common.remote.job.JobType;
 import org.apache.eventmesh.common.remote.request.CreateOrUpdateDataSourceReq;
 import org.apache.eventmesh.common.utils.JsonUtils;
 
@@ -231,6 +232,8 @@ public JobDetail getJobDetail(String jobID) {
         }
         detail.setState(state);
         detail.setTransportType(TransportType.getTransportType(job.getTransportType()));
+        detail.setJobType(JobType.fromIndex(job.getJobType()));
+        detail.setJobDesc(job.getJobDesc());
         return detail;
     }
 

eventmesh-common/src/main/java/org/apache/eventmesh/common/config/ConfigService.java

@@ -131,7 +131,7 @@ public <T> T getConfig(ConfigInfo configInfo) throws IOException {
         } else {
             filePath = path.startsWith(FILE_PATH_PREFIX) ? path.substring(FILE_PATH_PREFIX.length()) : this.configPath + path;
         }
-
+        filePath = normalizeFilePath(filePath);
         if (filePath.contains(".jar")) {
             try (final InputStream inputStream = getClass().getResourceAsStream(Objects.requireNonNull(resourceUrl))) {
                 if (inputStream == null) {
@@ -152,6 +152,15 @@ public <T> T getConfig(ConfigInfo configInfo) throws IOException {
         return (T) object;
     }
 
+    private String normalizeFilePath(String filePath) {
+        if (System.getProperty("os.name").toLowerCase().contains("win")) {
+            if (filePath.startsWith("/")) {
+                filePath = filePath.substring(1);
+            }
+        }
+        return filePath;
+    }
+
     private void populateConfig(Object object, Class<?> clazz, Config config)
         throws NoSuchFieldException, IOException, IllegalAccessException {
         ConfigInfo configInfo = new ConfigInfo();

eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/sink/config/HttpRetryConfig.java → eventmesh-common/src/main/java/org/apache/eventmesh/common/config/connector/http/HttpRetryConfig.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.eventmesh.connector.http.sink.config;
+package org.apache.eventmesh.common.config.connector.http;
 
 import lombok.Data;
 

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

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.eventmesh.connector.http.sink.config;
+package org.apache.eventmesh.common.config.connector.http;
 
 import org.apache.eventmesh.common.config.connector.SinkConfig;
 

eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/sink/config/HttpWebhookConfig.java → eventmesh-common/src/main/java/org/apache/eventmesh/common/config/connector/http/HttpWebhookConfig.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.eventmesh.connector.http.sink.config;
+package org.apache.eventmesh.common.config.connector.http;
 
 import lombok.Data;
 

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

@@ -15,9 +15,8 @@
  * limitations under the License.
  */
 
-package org.apache.eventmesh.connector.http.sink.config;
+package org.apache.eventmesh.common.config.connector.http;
 
-import io.vertx.core.http.HttpClientOptions;
 
 import lombok.Data;
 
@@ -29,19 +28,19 @@ public class SinkConnectorConfig {
     private String[] urls;
 
     // keepAlive, default true
-    private boolean keepAlive = HttpClientOptions.DEFAULT_KEEP_ALIVE;
+    private boolean keepAlive = true;
 
     // timeunit: ms, default 60000ms
-    private int keepAliveTimeout = HttpClientOptions.DEFAULT_KEEP_ALIVE_TIMEOUT * 1000; // Keep units consistent
+    private int keepAliveTimeout = 60 * 1000; // Keep units consistent
 
     // timeunit: ms, default 5000ms, recommended scope: 5000ms - 10000ms
     private int connectionTimeout = 5000;
 
     // timeunit: ms, default 5000ms
-    private int idleTimeout;
+    private int idleTimeout = 5000;
 
     // maximum number of HTTP/1 connections a client will pool, default 5
-    private int maxConnectionPoolSize = HttpClientOptions.DEFAULT_MAX_POOL_SIZE;
+    private int maxConnectionPoolSize = 5;
 
     // retry config
     private HttpRetryConfig retryConfig = new HttpRetryConfig();

eventmesh-common/src/main/java/org/apache/eventmesh/common/config/connector/http/SourceConnectorConfig.java

@@ -27,7 +27,7 @@ public class SourceConnectorConfig {
 
     private String connectorName;
 
-    private String path;
+    private String path = "/";
 
     private int port;
 
@@ -51,11 +51,11 @@ public class SourceConnectorConfig {
     private int batchSize = 10;
 
     // protocol, default CloudEvent
-    private String protocol = "CloudEvent";
+    private String protocol = "Common";
 
     // extra config, e.g. GitHub secret
     private Map<String, String> extraConfig = new HashMap<>();
 
     // data consistency enabled, default true
-    private boolean dataConsistencyEnabled = true;
+    private boolean dataConsistencyEnabled = false;
 }

eventmesh-common/src/main/java/org/apache/eventmesh/common/config/connector/rdb/canal/CanalSinkConfig.java

@@ -18,7 +18,8 @@
 package org.apache.eventmesh.common.config.connector.rdb.canal;
 
 import org.apache.eventmesh.common.config.connector.SinkConfig;
-import org.apache.eventmesh.common.remote.job.SyncMode;
+
+import java.util.Map;
 
 import lombok.Data;
 import lombok.EqualsAndHashCode;
@@ -27,25 +28,7 @@
 @EqualsAndHashCode(callSuper = true)
 public class CanalSinkConfig extends SinkConfig {
 
-    // batchSize
-    private Integer batchSize = 50;
-
-    // enable batch
-    private Boolean useBatch = true;
-
-    // sink thread size for single channel
-    private Integer poolSize = 5;
-
-    // sync mode: field/row
-    private SyncMode syncMode;
-
-    private boolean isGTIDMode = true;
-
-    private boolean isMariaDB = true;
-
-    // skip sink process exception
-    private Boolean skipException = false;
-
-    public SinkConnectorConfig sinkConnectorConfig;
+    // used to convert canal full/increment/check connector config
+    private Map<String, Object> sinkConfig;
 
 }

eventmesh-common/src/main/java/org/apache/eventmesh/common/config/connector/rdb/canal/CanalSinkFullConfig.java

@@ -26,6 +26,6 @@
 @Data
 @EqualsAndHashCode(callSuper = true)
 public class CanalSinkFullConfig extends SinkConfig {
-    private SinkConnectorConfig sinkConfig;
+    private SinkConnectorConfig sinkConnectorConfig;
     private String zeroDate;
 }

eventmesh-common/src/main/java/org/apache/eventmesh/common/config/connector/rdb/canal/CanalSinkIncrementConfig.java

@@ -0,0 +1,50 @@
+/*
+ * 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.common.config.connector.rdb.canal;
+
+import org.apache.eventmesh.common.remote.job.SyncMode;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+@Data
+@EqualsAndHashCode(callSuper = true)
+public class CanalSinkIncrementConfig extends CanalSinkConfig {
+
+    // batchSize
+    private Integer batchSize = 50;
+
+    // enable batch
+    private Boolean useBatch = true;
+
+    // sink thread size for single channel
+    private Integer poolSize = 5;
+
+    // sync mode: field/row
+    private SyncMode syncMode;
+
+    private boolean isGTIDMode = true;
+
+    private boolean isMariaDB = true;
+
+    // skip sink process exception
+    private Boolean skipException = false;
+
+    public SinkConnectorConfig sinkConnectorConfig;
+
+}

eventmesh-common/src/main/java/org/apache/eventmesh/common/config/connector/rdb/canal/CanalSourceConfig.java

@@ -18,11 +18,8 @@
 package org.apache.eventmesh.common.config.connector.rdb.canal;
 
 import org.apache.eventmesh.common.config.connector.SourceConfig;
-import org.apache.eventmesh.common.remote.job.SyncConsistency;
-import org.apache.eventmesh.common.remote.job.SyncMode;
-import org.apache.eventmesh.common.remote.offset.RecordPosition;
 
-import java.util.List;
+import java.util.Map;
 
 import lombok.Data;
 import lombok.EqualsAndHashCode;
@@ -31,56 +28,7 @@
 @EqualsAndHashCode(callSuper = true)
 public class CanalSourceConfig extends SourceConfig {
 
-    private String destination;
+    // used to convert canal full/increment/check connector config
+    private Map<String, Object> sourceConfig;
 
-    private Long canalInstanceId;
-
-    private String desc;
-
-    private boolean ddlSync = true;
-
-    private boolean filterTableError = false;
-
-    private Long slaveId;
-
-    private Short clientId;
-
-    private String serverUUID;
-
-    private boolean isMariaDB = true;
-
-    private boolean isGTIDMode = true;
-
-    private Integer batchSize = 10000;
-
-    private Long batchTimeout = -1L;
-
-    private String tableFilter;
-
-    private String fieldFilter;
-
-    private List<RecordPosition> recordPositions;
-
-    // ================================= channel parameter
-    // ================================
-
-    // enable remedy
-    private Boolean enableRemedy = false;
-
-    // sync mode: field/row
-    private SyncMode syncMode;
-
-    // sync consistency
-    private SyncConsistency syncConsistency;
-
-    // ================================= system parameter
-    // ================================
-
-    // Column name of the bidirectional synchronization mark
-    private String needSyncMarkTableColumnName = "needSync";
-
-    // Column value of the bidirectional synchronization mark
-    private String needSyncMarkTableColumnValue = "needSync";
-
-    private SourceConnectorConfig sourceConnectorConfig;
 }

eventmesh-common/src/main/java/org/apache/eventmesh/common/config/connector/rdb/canal/CanalSourceFullConfig.java

@@ -28,7 +28,7 @@
 @Data
 @EqualsAndHashCode(callSuper = true)
 public class CanalSourceFullConfig extends SourceConfig {
-    private SourceConnectorConfig connectorConfig;
+    private SourceConnectorConfig sourceConnectorConfig;
     private List<RecordPosition> startPosition;
     private int parallel;
     private int flushSize;

eventmesh-common/src/main/java/org/apache/eventmesh/common/config/connector/rdb/canal/CanalSourceIncrementConfig.java

@@ -0,0 +1,86 @@
+/*
+ * 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.common.config.connector.rdb.canal;
+
+import org.apache.eventmesh.common.remote.job.SyncConsistency;
+import org.apache.eventmesh.common.remote.job.SyncMode;
+import org.apache.eventmesh.common.remote.offset.RecordPosition;
+
+import java.util.List;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+@Data
+@EqualsAndHashCode(callSuper = true)
+public class CanalSourceIncrementConfig extends CanalSourceConfig {
+
+    private String destination;
+
+    private Long canalInstanceId;
+
+    private String desc;
+
+    private boolean ddlSync = true;
+
+    private boolean filterTableError = false;
+
+    private Long slaveId;
+
+    private Short clientId;
+
+    private String serverUUID;
+
+    private boolean isMariaDB = true;
+
+    private boolean isGTIDMode = true;
+
+    private Integer batchSize = 10000;
+
+    private Long batchTimeout = -1L;
+
+    private String tableFilter;
+
+    private String fieldFilter;
+
+    private List<RecordPosition> recordPositions;
+
+    // ================================= channel parameter
+    // ================================
+
+    // enable remedy
+    private Boolean enableRemedy = false;
+
+    // sync mode: field/row
+    private SyncMode syncMode;
+
+    // sync consistency
+    private SyncConsistency syncConsistency;
+
+    // ================================= system parameter
+    // ================================
+
+    // Column name of the bidirectional synchronization mark
+    private String needSyncMarkTableColumnName = "needSync";
+
+    // Column value of the bidirectional synchronization mark
+    private String needSyncMarkTableColumnValue = "needSync";
+
+    private SourceConnectorConfig sourceConnectorConfig;
+
+}