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

Add bit operation to prevent negative value.
apache · Oct 27, 2023 · 7c1dab4 · 7c1dab4
1 parent 9980aa4
commit 7c1dab4
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);
         }