[Fix][Connector-V2][ClickHouse] Fix ClickHouse Bug (#7897)

YOMO-Lee · YOMO-Lee · commit 427668128471 · 2024-10-25T18:34:23.000+08:00
1、When the ClickHouse connector is set to multi parallelism, the task extraction is completed but cannot be stopped normally [(#7897)](#7897) 2、Added E2E test cases for this issue [(#7897)](#7897) 3、Local developers want to observe **Job Progress Information** in a timely manner, Need to modify the following configuration.The configuration in config is invalid ``` seatunnel engine/seatunnel-engineer-common/src/main/resources/seatunnely.yaml ```
diff --git a/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/source/ClickhouseSourceReader.java b/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/source/ClickhouseSourceReader.java
@@ -17,6 +17,8 @@
 
 package org.apache.seatunnel.connectors.seatunnel.clickhouse.source;
 
+import lombok.extern.slf4j.Slf4j;
+import org.apache.seatunnel.api.source.Boundedness;
 import org.apache.seatunnel.api.source.Collector;
 import org.apache.seatunnel.api.source.SourceReader;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
@@ -34,7 +36,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Random;
-
+@Slf4j
 public class ClickhouseSourceReader implements SourceReader<SeaTunnelRow, ClickhouseSourceSplit> {
 
     private final List<ClickHouseNode> servers;
@@ -43,6 +45,7 @@ public class ClickhouseSourceReader implements SourceReader<SeaTunnelRow, Clickh
     private final SourceReader.Context readerContext;
     private ClickHouseRequest<?> request;
     private final String sql;
+    private volatile boolean noMoreSplit;
 
     private final List<ClickhouseSourceSplit> splits;
 
@@ -97,6 +100,12 @@ record -> {
             }
             this.readerContext.signalNoMoreElement();
             this.splits.clear();
+        } else if (noMoreSplit
+                && splits.isEmpty()
+                && Boundedness.BOUNDED.equals(readerContext.getBoundedness())) {
+            log.info("Closed the bounded ClickHouse source");
+            this.readerContext.signalNoMoreElement();
+            this.splits.clear();
         }
     }
 
@@ -111,7 +120,7 @@ public void addSplits(List<ClickhouseSourceSplit> splits) {
     }
 
     @Override
-    public void handleNoMoreSplits() {}
+    public void handleNoMoreSplits() {noMoreSplit = true;}
 
     @Override
     public void notifyCheckpointComplete(long checkpointId) throws Exception {}
diff --git a/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/source/ClickhouseSourceSplitEnumerator.java b/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/source/ClickhouseSourceSplitEnumerator.java
@@ -77,6 +77,8 @@ public void registerReader(int subtaskId) {
         if (assigned < 0) {
             assigned = subtaskId;
             context.assignSplit(subtaskId, new ClickhouseSourceSplit());
+        } else {
+            context.signalNoMoreSplits(subtaskId);
         }
     }
 
diff --git a/seatunnel-engine/seatunnel-engine-common/src/main/resources/seatunnel.yaml b/seatunnel-engine/seatunnel-engine-common/src/main/resources/seatunnel.yaml
@@ -20,6 +20,7 @@ seatunnel:
         backup-count: 1
         queue-type: blockingqueue
         print-execution-info-interval: 60
+        print-job-metrics-info-interval: 60
         slot-service:
             dynamic-slot: true
         checkpoint:

Original file line number	Diff line number	Diff line change
`@@ -77,6 +77,8 @@ public void registerReader(int subtaskId) {`
`77`	`77`	`if (assigned < 0) {`
`78`	`78`	`assigned = subtaskId;`
`79`	`79`	`context.assignSplit(subtaskId, new ClickhouseSourceSplit());`
	`80`	`+ } else {`
	`81`	`+ context.signalNoMoreSplits(subtaskId);`
`80`	`82`	`}`
`81`	`83`	`}`
`82`	`84`