[Improve] Remove SourceSplitEnumerator::currentUnassignedSplitSize

seatunnel-api/src/main/java/org/apache/seatunnel/api/source/SourceSplitEnumerator.java

@@ -70,8 +70,6 @@ public interface SourceSplitEnumerator<SplitT extends SourceSplit, StateT>
      */
     void addSplitsBack(List<SplitT> splits, int subtaskId);
 
-    int currentUnassignedSplitSize();
-
     void handleSplitRequest(int subtaskId);
 
     void registerReader(int subtaskId);

seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSourceSplitEnumerator.java

@@ -143,11 +143,6 @@ public void addSplitsBack(List<AmazonDynamoDBSourceSplit> splits, int subtaskId)
         }
     }
 
-    @Override
-    public int currentUnassignedSplitSize() {
-        return pendingSplits.size();
-    }
-
     @Override
     public void handleSplitRequest(int subtaskId) {}
 

seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/IncrementalSourceEnumerator.java

@@ -89,11 +89,6 @@ public void addSplitsBack(List<SourceSplitBase> splits, int subtaskId) {
         splitAssigner.addSplits(splits);
     }
 
-    @Override
-    public int currentUnassignedSplitSize() {
-        return 0;
-    }
-
     @Override
     public void registerReader(int subtaskId) {
         // do nothing

seatunnel-connectors-v2/connector-cdc/connector-cdc-tidb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/tidb/source/enumerator/TiDBSourceSplitEnumerator.java

@@ -188,11 +188,6 @@ public void addSplitsBack(List<TiDBSourceSplit> splits, int subtaskId) {
         }
     }
 
-    @Override
-    public int currentUnassignedSplitSize() {
-        return pendingSplit.size();
-    }
-
     @Override
     public void handleSplitRequest(int subtaskId) {}
 

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

@@ -130,11 +130,6 @@ public void addSplitsBack(List<ClickhouseSourceSplit> splits, int subtaskId) {
         LOG.info("Add back splits {} to JdbcSourceSplitEnumerator.", splits.size());
     }
 
-    @Override
-    public int currentUnassignedSplitSize() {
-        return this.pendingSplit.size();
-    }
-
     @Override
     public void handleSplitRequest(int subtaskId) {
         throw new ClickhouseConnectorException(

seatunnel-connectors-v2/connector-common/src/main/java/org/apache/seatunnel/connectors/seatunnel/common/source/SingleSplitEnumerator.java

@@ -70,11 +70,6 @@ protected void assignSplit() {
         }
     }
 
-    @Override
-    public int currentUnassignedSplitSize() {
-        return 0;
-    }
-
     @Override
     public void handleSplitRequest(int subtaskId) {
         // nothing

seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/source/split/DorisSourceSplitEnumerator.java

@@ -119,11 +119,6 @@ public void addSplitsBack(List<DorisSourceSplit> splits, int subtaskId) {
         }
     }
 
-    @Override
-    public int currentUnassignedSplitSize() {
-        return this.pendingSplit.size();
-    }
-
     @Override
     public void handleSplitRequest(int subtaskId) {
         throw new DorisConnectorException(

seatunnel-connectors-v2/connector-doris/src/test/java/org/apache/seatunnel/connectors/doris/split/DorisSourceSplitEnumeratorTest.java

@@ -97,9 +97,6 @@ public void dorisSourceSplitEnumeratorTest() {
             Assertions.assertEquals(
                     allocateFiles(i, PARALLELISM, PARTITION_NUMS), splitAllValues.get(i).size());
         }
-
-        // check no duplicate file assigned
-        Assertions.assertEquals(0, dorisSourceSplitEnumerator.currentUnassignedSplitSize());
     }
 
     private List<PartitionDefinition> buildPartitionDefinitions() {

seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/source/EasysearchSourceSplitEnumerator.java

@@ -174,11 +174,6 @@ public void addSplitsBack(List<EasysearchSourceSplit> splits, int subtaskId) {
         }
     }
 
-    @Override
-    public int currentUnassignedSplitSize() {
-        return pendingSplit.size();
-    }
-
     @Override
     public void handleSplitRequest(int subtaskId) {
         throw new EasysearchConnectorException(

seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/source/ElasticsearchSourceSplitEnumerator.java

@@ -172,11 +172,6 @@ public void addSplitsBack(List<ElasticsearchSourceSplit> splits, int subtaskId)
         }
     }
 
-    @Override
-    public int currentUnassignedSplitSize() {
-        return pendingSplit.size();
-    }
-
     @Override
     public void handleSplitRequest(int subtaskId) {
         throw new ElasticsearchConnectorException(

seatunnel-connectors-v2/connector-fake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeSourceSplitEnumerator.java

@@ -73,11 +73,6 @@ public void addSplitsBack(List<FakeSourceSplit> splits, int subtaskId) {
         addSplitChangeToPendingAssignments(splits);
     }
 
-    @Override
-    public int currentUnassignedSplitSize() {
-        return pendingSplits.size();
-    }
-
     @Override
     public void handleSplitRequest(int subtaskId) {}
 

seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/split/FileSourceSplitEnumerator.java

@@ -130,11 +130,6 @@ private static int getSplitOwner(int assignCount, int numReaders) {
         return assignCount % numReaders;
     }
 
-    @Override
-    public int currentUnassignedSplitSize() {
-        return allSplit.size() - assignedSplit.size();
-    }
-
     @Override
     public void registerReader(int subtaskId) {
         // do nothing

seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/split/MultipleTableFileSourceSplitEnumerator.java

@@ -95,11 +95,6 @@ public void addSplitsBack(List<FileSourceSplit> splits, int subtaskId) {
         assignSplit(subtaskId);
     }
 
-    @Override
-    public int currentUnassignedSplitSize() {
-        return allSplit.size() - assignedSplit.size();
-    }
-
     @Override
     public void handleSplitRequest(int subtaskId) {}
 

seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/split/FileSourceSplitEnumeratorTest.java

@@ -89,9 +89,6 @@ public EventListener getEventListener() {
 
         fileSourceSplitEnumerator.run();
 
-        // check all files are assigned
-        Assertions.assertEquals(fileSourceSplitEnumerator.currentUnassignedSplitSize(), 0);
-
         Set<FileSourceSplit> valueSet =
                 assignSplitMap.values().stream().flatMap(List::stream).collect(Collectors.toSet());
 

seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/split/MultipleTableFileSourceSplitEnumeratorTest.java

@@ -81,7 +81,6 @@ void assignSplitTest() throws Exception {
                         context, baseMultipleTableFileSourceConfig);
 
         enumerator.open();
-        Assertions.assertEquals(50, enumerator.currentUnassignedSplitSize());
         IntStream.range(0, parallelism).forEach(enumerator::registerReader);
         enumerator.run();
 
@@ -99,9 +98,6 @@ void assignSplitTest() throws Exception {
             Assertions.assertEquals(
                     allocateFiles(i, parallelism, fileSize), splitAllValues.get(i).size());
         }
-
-        // check no duplicate file assigned
-        Assertions.assertEquals(0, enumerator.currentUnassignedSplitSize());
     }
 
     /**

seatunnel-connectors-v2/connector-hbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/hbase/source/HbaseSourceSplitEnumerator.java

@@ -94,11 +94,6 @@ public void addSplitsBack(List<HbaseSourceSplit> splits, int subtaskId) {
         }
     }
 
-    @Override
-    public int currentUnassignedSplitSize() {
-        return pendingSplit.size();
-    }
-
     @Override
     public void registerReader(int subtaskId) {
         pendingSplit = getTableSplits();

seatunnel-connectors-v2/connector-hive/src/main/java/org/apache/seatunnel/connectors/seatunnel/hive/source/split/MultipleTableHiveSourceSplitEnumerator.java

@@ -96,11 +96,6 @@ public void addSplitsBack(List<FileSourceSplit> splits, int subtaskId) {
         assignSplit(subtaskId);
     }
 
-    @Override
-    public int currentUnassignedSplitSize() {
-        return allSplit.size() - assignedSplit.size();
-    }
-
     @Override
     public void handleSplitRequest(int subtaskId) {}
 

seatunnel-connectors-v2/connector-hive/src/test/java/org/apache/seatunnel/connectors/seatunnel/hive/split/MultipleTableHiveSourceSplitEnumeratorTest.java

@@ -82,7 +82,6 @@ void assignSplitRoundTest() throws Exception {
                 new MultipleTableHiveSourceSplitEnumerator(context, mockConfig);
 
         enumerator.open();
-        Assertions.assertEquals(50, enumerator.currentUnassignedSplitSize());
         IntStream.range(0, parallelism).forEach(enumerator::registerReader);
         enumerator.run();
 
@@ -100,9 +99,6 @@ void assignSplitRoundTest() throws Exception {
             Assertions.assertEquals(
                     allocateFiles(i, parallelism, fileSize), splitAllValues.get(i).size());
         }
-
-        // check no duplicate file assigned
-        Assertions.assertEquals(0, enumerator.currentUnassignedSplitSize());
     }
 
     /**

seatunnel-connectors-v2/connector-iceberg/src/main/java/org/apache/seatunnel/connectors/seatunnel/iceberg/source/enumerator/AbstractSplitEnumerator.java

@@ -144,17 +144,6 @@ public void addSplitsBack(List<IcebergFileScanTaskSplit> splits, int subtaskId)
         log.info("Add back splits {} to JdbcSourceSplitEnumerator.", splits.size());
     }
 
-    @Override
-    public int currentUnassignedSplitSize() {
-        if (!pendingTables.isEmpty()) {
-            return pendingTables.size();
-        }
-        if (!pendingSplits.isEmpty()) {
-            return pendingSplits.values().stream().mapToInt(List::size).sum();
-        }
-        return 0;
-    }
-
     @Override
     public void handleSplitRequest(int subtaskId) {}
 

seatunnel-connectors-v2/connector-influxdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/influxdb/source/InfluxDBSourceSplitEnumerator.java

@@ -93,11 +93,6 @@ public void addSplitsBack(List splits, int subtaskId) {
         }
     }
 
-    @Override
-    public int currentUnassignedSplitSize() {
-        return pendingSplit.size();
-    }
-
     @Override
     public void registerReader(int subtaskId) {
         log.debug("Register reader {} to InfluxDBSourceSplitEnumerator.", subtaskId);

seatunnel-connectors-v2/connector-iotdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdb/source/IoTDBSourceSplitEnumerator.java

@@ -188,11 +188,6 @@ public void addSplitsBack(List<IoTDBSourceSplit> splits, int subtaskId) {
         }
     }
 
-    @Override
-    public int currentUnassignedSplitSize() {
-        return pendingSplit.size();
-    }
-
     @Override
     public void registerReader(int subtaskId) {
         log.debug("Register reader {} to IoTDBSourceSplitEnumerator.", subtaskId);

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/JdbcSourceSplitEnumerator.java

@@ -116,11 +116,6 @@ public void addSplitsBack(List<JdbcSourceSplit> splits, int subtaskId) {
         LOG.info("Add back splits {} to JdbcSourceSplitEnumerator.", splits.size());
     }
 
-    @Override
-    public int currentUnassignedSplitSize() {
-        return pendingTables.isEmpty() && pendingSplits.isEmpty() ? 0 : 1;
-    }
-
     @Override
     public void handleSplitRequest(int subtaskId) {
         throw new JdbcConnectorException(

seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceSplitEnumerator.java

@@ -298,11 +298,6 @@ public void addSplitsBack(List<KafkaSourceSplit> splits, int subtaskId) {
         }
     }
 
-    @Override
-    public int currentUnassignedSplitSize() {
-        return pendingSplit.size();
-    }
-
     @Override
     public void handleSplitRequest(int subtaskId) {
         // Do nothing because Kafka source push split.

seatunnel-connectors-v2/connector-kudu/src/main/java/org/apache/seatunnel/connectors/seatunnel/kudu/source/KuduSourceSplitEnumerator.java

@@ -175,11 +175,6 @@ private int getSplitOwner(String splitId, int numReaders) {
         return (splitId.hashCode() & Integer.MAX_VALUE) % numReaders;
     }
 
-    @Override
-    public int currentUnassignedSplitSize() {
-        return pendingSplits.size();
-    }
-
     @Override
     public void handleSplitRequest(int subtaskId) {
         throw new KuduConnectorException(

seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/source/MaxcomputeSourceSplitEnumerator.java

@@ -87,11 +87,6 @@ public void addSplitsBack(List<MaxcomputeSourceSplit> splits, int subtaskId) {
         addSplitChangeToPendingAssignments(splits);
     }
 
-    @Override
-    public int currentUnassignedSplitSize() {
-        return pendingSplits.size();
-    }
-
     @Override
     public void registerReader(int subtaskId) {}
 

seatunnel-connectors-v2/connector-milvus/src/main/java/org/apache/seatunnel/connectors/seatunnel/milvus/source/MilvusSourceSplitEnumerator.java

@@ -217,11 +217,6 @@ private void addPendingSplit(Collection<MilvusSourceSplit> splits, int ownerRead
         pendingSplits.computeIfAbsent(ownerReader, r -> new ArrayList<>()).addAll(splits);
     }
 
-    @Override
-    public int currentUnassignedSplitSize() {
-        return pendingTables.isEmpty() && pendingSplits.isEmpty() ? 0 : 1;
-    }
-
     @Override
     public void handleSplitRequest(int subtaskId) {
         throw new MilvusConnectorException(

seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/source/enumerator/MongodbSplitEnumerator.java

@@ -103,11 +103,6 @@ public void addSplitsBack(List<MongoSplit> splits, int subtaskId) {
         }
     }
 
-    @Override
-    public int currentUnassignedSplitSize() {
-        return pendingSplits.size();
-    }
-
     @Override
     public void handleSplitRequest(int subtaskId) {
         throw new MongodbConnectorException(

seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/enumerator/AbstractSplitEnumerator.java

@@ -133,11 +133,6 @@ public void addSplitsBack(List<PaimonSourceSplit> splits, int subtaskId) {
         }
     }
 
-    @Override
-    public int currentUnassignedSplitSize() {
-        return pendingSplits.size();
-    }
-
     @Override
     public void registerReader(int subtaskId) {
         readersAwaitingSplit.add(subtaskId);

seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/enumerator/PulsarSplitEnumerator.java

@@ -281,11 +281,6 @@ public void addSplitsBack(List<PulsarPartitionSplit> splits, int subtaskId) {
         }
     }
 
-    @Override
-    public int currentUnassignedSplitSize() {
-        return pendingPartitionSplits.size();
-    }
-
     @Override
     public void handleSplitRequest(int subtaskId) {
         // Do nothing because Pulsar source push split.

seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/source/RabbitmqSplitEnumerator.java

@@ -44,11 +44,6 @@ public void addSplitsBack(List splits, int subtaskId) {
         // do nothing
     }
 
-    @Override
-    public int currentUnassignedSplitSize() {
-        return 0;
-    }
-
     @Override
     public void handleSplitRequest(int subtaskId) {
         // do nothing

seatunnel-connectors-v2/connector-rocketmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/source/RocketMqSourceSplitEnumerator.java

@@ -173,11 +173,6 @@ public void addSplitsBack(List<RocketMqSourceSplit> splits, int subtaskId) {
         }
     }
 
-    @Override
-    public int currentUnassignedSplitSize() {
-        return pendingSplit.size();
-    }
-
     @Override
     public void handleSplitRequest(int subtaskId) {
         // No-op

seatunnel-connectors-v2/connector-sls/src/main/java/org/apache/seatunnel/connectors/seatunnel/sls/source/SlsSourceSplitEnumerator.java

@@ -138,11 +138,6 @@ public void addSplitsBack(List<SlsSourceSplit> splits, int subtaskId) {
         }
     }
 
-    @Override
-    public int currentUnassignedSplitSize() {
-        return 0;
-    }
-
     @Override
     public void handleSplitRequest(int subtaskId) {}