Add task execution engine and progress tracking

Author: lmj

connectors/src/main/java/com/dbsyncer/connectors/client/KafkaConnectClient.java

@@ -307,6 +307,15 @@ public boolean waitForConnectorRunning(String connectorName, long timeoutMs, lon
                 }
                 log.debug("Connector {} not yet running, waiting...", connectorName);
                 Thread.sleep(pollIntervalMs);
+            } catch (ConnectorNotFoundException e) {
+                // Connector may not be immediately available after creation; keep waiting
+                log.debug("Connector {} not found yet, waiting...", connectorName);
+                try {
+                    Thread.sleep(pollIntervalMs);
+                } catch (InterruptedException ie) {
+                    Thread.currentThread().interrupt();
+                    throw new KafkaConnectException("Interrupted while waiting for connector", ie);
+                }
             } catch (InterruptedException e) {
                 Thread.currentThread().interrupt();
                 throw new KafkaConnectException("Interrupted while waiting for connector", e);

connectors/src/main/java/com/dbsyncer/connectors/config/MySqlSourceConnectorConfig.java

@@ -101,6 +101,11 @@ public class MySqlSourceConnectorConfig implements SourceConnectorConfig {
     private String sslTruststore;
     private String sslTruststorePassword;
 
+    // Optional transformation injection
+    @Builder.Default
+    private boolean enableTypeMappingTransform = false;
+    private String typeMappingSourceDb; // e.g. "mysql"
+
     @Override
     public String getConnectorClass() {
         return CONNECTOR_CLASS;
@@ -240,6 +245,14 @@ public Map<String, String> toConfigMap() {
             config.put("database.ssl.truststore.password", sslTruststorePassword);
         }
 
+        // Optional SMT transform to normalize schema/types
+        if (enableTypeMappingTransform && typeMappingSourceDb != null && !typeMappingSourceDb.isBlank()) {
+            config.put("transforms", "applyTypeMapping");
+            config.put("transforms.applyTypeMapping.type",
+                    "com.dbsyncer.transformations.smt.ApplyTypeMapping");
+            config.put("transforms.applyTypeMapping.source.db", typeMappingSourceDb);
+        }
+
         return config;
     }
 

connectors/src/main/java/com/dbsyncer/connectors/config/OracleSourceConnectorConfig.java

@@ -123,6 +123,11 @@ public String getValue() {
     @Builder.Default
     private int maxQueueSize = 8192;
 
+    // Optional transformation injection
+    @Builder.Default
+    private boolean enableTypeMappingTransform = false;
+    private String typeMappingSourceDb; // e.g. "oracle"
+
     @Override
     public String getConnectorClass() {
         return CONNECTOR_CLASS;
@@ -247,6 +252,14 @@ public Map<String, String> toConfigMap() {
         config.put("max.batch.size", String.valueOf(maxBatchSize));
         config.put("max.queue.size", String.valueOf(maxQueueSize));
 
+        // Optional SMT transform to normalize schema/types
+        if (enableTypeMappingTransform && typeMappingSourceDb != null && !typeMappingSourceDb.isBlank()) {
+            config.put("transforms", "applyTypeMapping");
+            config.put("transforms.applyTypeMapping.type",
+                    "com.dbsyncer.transformations.smt.ApplyTypeMapping");
+            config.put("transforms.applyTypeMapping.source.db", typeMappingSourceDb);
+        }
+
         return config;
     }
 

connectors/src/main/java/com/dbsyncer/connectors/config/PostgresSourceConnectorConfig.java

@@ -121,6 +121,11 @@ public String getValue() {
     private String sslKey;
     private String sslPassword;
 
+    // Optional transformation injection
+    @Builder.Default
+    private boolean enableTypeMappingTransform = false;
+    private String typeMappingSourceDb; // e.g. "postgresql"
+
     @Override
     public String getConnectorClass() {
         return CONNECTOR_CLASS;
@@ -259,6 +264,14 @@ public Map<String, String> toConfigMap() {
             config.put("database.sslpassword", sslPassword);
         }
 
+        // Optional SMT transform to normalize schema/types
+        if (enableTypeMappingTransform && typeMappingSourceDb != null && !typeMappingSourceDb.isBlank()) {
+            config.put("transforms", "applyTypeMapping");
+            config.put("transforms.applyTypeMapping.type",
+                    "com.dbsyncer.transformations.smt.ApplyTypeMapping");
+            config.put("transforms.applyTypeMapping.source.db", typeMappingSourceDb);
+        }
+
         return config;
     }
 

docs/ARCHITECTURE.md

@@ -154,8 +154,8 @@ db-syncer-debezium/
 │   └── config/             # Connector 配置生成器
 │
 ├── transformations/        # 数据转换层
-│   ├── smt/                # Single Message Transforms
-│   ├── mapper/             # 类型映射器
+│   ├── smt/                # Single Message Transforms（如 ApplyTypeMapping）
+│   ├── mapper/             # 类型映射器（MySQL/Oracle → PostgreSQL）
 │   └── schema/             # Schema 转换
 │
 └── monitoring/             # 监控模块
@@ -238,6 +238,32 @@ CREATE TABLE schema_history (
 );
 ```
 
+### 2. Transformations（数据转换层）
+
+**职责**:
+- 统一不同源数据库的类型，便于落地到目标（如 PostgreSQL）
+- 标准化 Debezium 事件中的逻辑类型，方便下游 Sink 处理
+
+**关键实现**:
+- `ApplyTypeMapping` SMT: 可选注入到 Source Connector，用于重写记录值的 Schema/Value
+  - 时间逻辑类型映射:
+    - `io.debezium.time.Date` → Kafka Connect `Date`
+    - `io.debezium.time.Time` → Kafka Connect `Time`
+    - `io.debezium.time.Timestamp`/`MicroTimestamp`/`NanoTimestamp` → Kafka Connect `Timestamp`
+  - JSON 逻辑类型映射:
+    - `io.debezium.data.Json` → 普通 `string`
+  - Decimal 逻辑类型:
+    - `io.debezium.data.VariableScaleDecimal` → `string`（默认）或 Kafka Connect `Decimal`（可选，存在 Schema 变更频率高的风险）
+  - 配置项:
+    - `transforms.applyTypeMapping.source.db`（mysql/oracle/postgresql）
+    - `transforms.applyTypeMapping.enable.time.mapping`（默认 true）
+    - `transforms.applyTypeMapping.enable.json.mapping`（默认 true）
+    - `transforms.applyTypeMapping.enable.decimal.mapping`（默认 true）
+    - `transforms.applyTypeMapping.decimal.target`（string/decimal，默认 string）
+
+**类型映射器**:
+- `MySqlToPostgresTypeMapper` 与 `OracleToPostgresTypeMapper` 提供列级类型映射（供 Schema 生成/校验使用）。
+
 ### 2. CLI Tool (命令行工具)
 
 **职责**:

metadata-service/pom.xml

@@ -16,12 +16,29 @@
     <description>Metadata management service with PostgreSQL storage</description>
 
     <dependencies>
+        <!-- Spring Boot Actuator & Metrics -->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-actuator</artifactId>
+        </dependency>
+
+        <!-- Micrometer Prometheus Registry -->
+        <dependency>
+            <groupId>io.micrometer</groupId>
+            <artifactId>micrometer-registry-prometheus</artifactId>
+        </dependency>
+
         <!-- Internal -->
         <dependency>
             <groupId>com.dbsyncer</groupId>
             <artifactId>common</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>com.dbsyncer</groupId>
+            <artifactId>connectors</artifactId>
+            <version>${project.version}</version>
+        </dependency>
 
         <!-- Spring Boot -->
         <dependency>

metadata-service/src/main/java/com/dbsyncer/metadata/controller/ProgressController.java

@@ -76,6 +76,16 @@ public ResponseEntity<Double> getAverageLag(@PathVariable UUID taskId) {
         return ResponseEntity.ok(averageLag);
     }
 
+    /**
+     * Get ETA (seconds) for task completion if it can be estimated, otherwise returns null.
+     */
+    @GetMapping("/eta")
+    public ResponseEntity<Long> getEtaSeconds(@PathVariable UUID taskId) {
+        log.debug("REST request to get ETA for task: {}", taskId);
+        Long etaSeconds = progressTrackingService.estimateEtaSeconds(taskId);
+        return ResponseEntity.ok(etaSeconds);
+    }
+
     /**
      * Get tables with errors.
      */

metadata-service/src/main/java/com/dbsyncer/metadata/controller/TaskController.java

@@ -4,6 +4,7 @@
 import com.dbsyncer.metadata.dto.TaskResponse;
 import com.dbsyncer.metadata.dto.TaskUpdateRequest;
 import com.dbsyncer.metadata.entity.TaskStatus;
+import com.dbsyncer.metadata.service.TaskExecutionService;
 import com.dbsyncer.metadata.service.TaskService;
 import jakarta.validation.Valid;
 import lombok.RequiredArgsConstructor;
@@ -29,6 +30,7 @@
 public class TaskController {
 
     private final TaskService taskService;
+    private final TaskExecutionService executionService;
 
     /**
      * Create a new migration task.
@@ -119,7 +121,7 @@ public ResponseEntity<Void> deleteTask(@PathVariable UUID taskId) {
     @PostMapping("/{taskId}/start")
     public ResponseEntity<TaskResponse> startTask(@PathVariable UUID taskId) {
         log.info("REST request to start task: {}", taskId);
-        TaskResponse response = taskService.updateTaskStatus(taskId, TaskStatus.STARTING);
+        TaskResponse response = executionService.startTask(taskId);
         return ResponseEntity.ok(response);
     }
 
@@ -129,7 +131,7 @@ public ResponseEntity<TaskResponse> startTask(@PathVariable UUID taskId) {
     @PostMapping("/{taskId}/stop")
     public ResponseEntity<TaskResponse> stopTask(@PathVariable UUID taskId) {
         log.info("REST request to stop task: {}", taskId);
-        TaskResponse response = taskService.updateTaskStatus(taskId, TaskStatus.STOPPING);
+        TaskResponse response = executionService.stopTask(taskId);
         return ResponseEntity.ok(response);
     }
 
@@ -139,7 +141,7 @@ public ResponseEntity<TaskResponse> stopTask(@PathVariable UUID taskId) {
     @PostMapping("/{taskId}/pause")
     public ResponseEntity<TaskResponse> pauseTask(@PathVariable UUID taskId) {
         log.info("REST request to pause task: {}", taskId);
-        TaskResponse response = taskService.updateTaskStatus(taskId, TaskStatus.PAUSED);
+        TaskResponse response = executionService.pauseTask(taskId);
         return ResponseEntity.ok(response);
     }
 
@@ -149,7 +151,7 @@ public ResponseEntity<TaskResponse> pauseTask(@PathVariable UUID taskId) {
     @PostMapping("/{taskId}/resume")
     public ResponseEntity<TaskResponse> resumeTask(@PathVariable UUID taskId) {
         log.info("REST request to resume task: {}", taskId);
-        TaskResponse response = taskService.updateTaskStatus(taskId, TaskStatus.RUNNING);
+        TaskResponse response = executionService.resumeTask(taskId);
         return ResponseEntity.ok(response);
     }
 

metadata-service/src/main/java/com/dbsyncer/metadata/dto/TaskCreateRequest.java

@@ -73,6 +73,9 @@ public class TaskCreateRequest {
     private Integer batchSize = 10000;
     private Integer maxQueueSize = 8192;
     private Integer pollIntervalMs = 1000;
+    private Boolean incrementalSnapshot = false;
+    private Integer snapshotChunkSize = 10000;
+    private Integer parallelTables = 1;
 
     // Metadata
     private String createdBy;

metadata-service/src/main/java/com/dbsyncer/metadata/dto/TaskResponse.java

@@ -47,6 +47,9 @@ public class TaskResponse {
     private Integer batchSize;
     private Integer maxQueueSize;
     private Integer pollIntervalMs;
+    private Boolean incrementalSnapshot;
+    private Integer snapshotChunkSize;
+    private Integer parallelTables;
 
     // Status
     private TaskStatus status;
@@ -97,6 +100,9 @@ public static TaskResponse fromEntity(MigrationTask task) {
         response.setBatchSize(task.getBatchSize());
         response.setMaxQueueSize(task.getMaxQueueSize());
         response.setPollIntervalMs(task.getPollIntervalMs());
+        response.setIncrementalSnapshot(task.getIncrementalSnapshot());
+        response.setSnapshotChunkSize(task.getSnapshotChunkSize());
+        response.setParallelTables(task.getParallelTables());
 
         response.setStatus(task.getStatus());
         response.setErrorMessage(task.getErrorMessage());