[BUG][Connector-V2] Iceberg source lost data with parallelism option (#…

…5732)
apache · Oct 28, 2023 · 7f3b4be · 7f3b4be
1 parent 4617423
commit 7f3b4be
Showing 1 changed file with 1 addition and 1 deletion.
diff --git a/...che/seatunnel/connectors/seatunnel/iceberg/source/enumerator/AbstractSplitEnumerator.java b/...che/seatunnel/connectors/seatunnel/iceberg/source/enumerator/AbstractSplitEnumerator.java
@@ -108,7 +108,7 @@ protected void refreshPendingSplits() {
     private void addPendingSplits(Collection<IcebergFileScanTaskSplit> newSplits) {
         int numReaders = context.currentParallelism();
         for (IcebergFileScanTaskSplit newSplit : newSplits) {
-            int ownerReader = newSplit.splitId().hashCode() % numReaders;
+            int ownerReader = (newSplit.splitId().hashCode() & Integer.MAX_VALUE) % numReaders;
             pendingSplits.computeIfAbsent(ownerReader, r -> new ArrayList<>()).add(newSplit);
             log.info("Assigning {} to {} reader.", newSplit, ownerReader);
         }