[Fix][Doc] Fix LocalFile doc (#7887)

docs/en/connector-v2/source/LocalFile.md

@@ -254,6 +254,12 @@ Specifies Whether to process data using the tag attribute format.
 
 Filter pattern, which used for filtering files.
 
+The filtering format is similar to wildcard matching file names in Linux.
+
+However, it should be noted that unlike Linux wildcard characters, when encountering file suffixes, the middle dot cannot be omitted.
+
+For example, `abc20241022.csv`, the normal Linux wildcard `abc*` is sufficient, but here we need to use `abc*.*` , Pay attention to a point in the middle.
+
 ### compress_codec [string]
 
 The compress codec of files and the details that supported as the following shown:
@@ -406,6 +412,29 @@ sink {
 
 ```
 
+### Filter File
+
+```hocon
+env {
+  parallelism = 1
+  job.mode = "BATCH"
+}
+
+source {
+  LocalFile {
+    path = "/seatunnel/read/"
+    file_format_type = "csv"
+    skip_header_row_number = 1
+    // file example abcD2024.csv
+    file_filter_pattern = "abc[DX]*.*"
+  }
+}
+sink {
+  Console {
+  }
+}
+```
+
 ## Changelog
 
 ### 2.2.0-beta 2022-09-26

seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/source/ClickhouseSourceReader.java

@@ -17,6 +17,7 @@
 
 package org.apache.seatunnel.connectors.seatunnel.clickhouse.source;
 
+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;
@@ -28,13 +29,15 @@
 import com.clickhouse.client.ClickHouseNode;
 import com.clickhouse.client.ClickHouseRequest;
 import com.clickhouse.client.ClickHouseResponse;
+import lombok.extern.slf4j.Slf4j;
 
 import java.io.IOException;
 import java.util.ArrayList;
 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 +46,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 +101,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 +121,9 @@ public void addSplits(List<ClickhouseSourceSplit> splits) {
     }
 
     @Override
-    public void handleNoMoreSplits() {}
+    public void handleNoMoreSplits() {
+        noMoreSplit = true;
+    }
 
     @Override
     public void notifyCheckpointComplete(long checkpointId) throws Exception {}

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);
         }
     }
 

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-clickhouse-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/ClickhouseIT.java

@@ -101,6 +101,13 @@ public void testClickhouse(TestContainer container) throws Exception {
         clearSinkTable();
     }
 
+    @TestTemplate
+    public void testSourceParallelism(TestContainer container) throws Exception {
+        LOG.info("=========多并行度测试===========");
+        Container.ExecResult execResult = container.executeJob("/clickhouse_to_console.conf");
+        Assertions.assertEquals(0, execResult.getExitCode());
+    }
+
     @BeforeAll
     @Override
     public void startUp() throws Exception {

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-clickhouse-e2e/src/test/resources/clickhouse_to_console.conf

@@ -0,0 +1,45 @@
+#
+# 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.
+#
+######
+###### This config file is a demonstration of streaming processing in seatunnel config
+######
+
+env {
+  parallelism = 3
+  job.mode = "BATCH"
+}
+
+source {
+  # This is a example source plugin **only for test and demonstrate the feature source plugin**
+  Clickhouse {
+    host = "clickhouse:8123"
+    database = "default"
+    sql = "select * from source_table"
+    username = "default"
+    password = ""
+    result_table_name = "source_table"
+  }
+  # If you would like to get more information about how to configure seatunnel and see full list of source plugins,
+  # please go to https://seatunnel.apache.org/docs/connector-v2/source/ClickhouseSource
+}
+
+sink {
+  console {
+  }
+  # If you would like to get more information about how to configure seatunnel and see full list of sink plugins,
+  # please go to https://seatunnel.apache.org/docs/connector-v2/sink
+}

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: