This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12395-71860b4469bbd83722c81aa85ddd7e57f5f440de in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 36401ab4bbb652c02cd721aa70f53c8ef4fe959a Author: Sepuri Sai Krishna <[email protected]> AuthorDate: Sun Sep 20 02:15:14 2026 +0000 [Improve][Connector-V2] Use HashUtils.bucketIndex in AmazonDocumentDB enumerator (#12395) --- .../AmazonDocumentDBSourceSplitEnumerator.java | 3 +- .../AmazonDocumentDBSourceSplitEnumeratorTest.java | 41 ++++++++++++++++++++++ 2 files changed, 43 insertions(+), 1 deletion(-) diff --git a/seatunnel-connectors-v2/connector-amazondocumentdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondocumentdb/source/AmazonDocumentDBSourceSplitEnumerator.java b/seatunnel-connectors-v2/connector-amazondocumentdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondocumentdb/source/AmazonDocumentDBSourceSplitEnumerator.java index 99c4bd37c2..1919eb6728 100644 --- a/seatunnel-connectors-v2/connector-amazondocumentdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondocumentdb/source/AmazonDocumentDBSourceSplitEnumerator.java +++ b/seatunnel-connectors-v2/connector-amazondocumentdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondocumentdb/source/AmazonDocumentDBSourceSplitEnumerator.java @@ -18,6 +18,7 @@ package org.apache.seatunnel.connectors.seatunnel.amazondocumentdb.source; import org.apache.seatunnel.api.source.SourceSplitEnumerator; +import org.apache.seatunnel.common.utils.HashUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -195,6 +196,6 @@ public class AmazonDocumentDBSourceSplitEnumerator } private static int getSplitOwner(Integer splitId, int readerCount) { - return (splitId.hashCode() & Integer.MAX_VALUE) % readerCount; + return HashUtils.bucketIndex(splitId, readerCount); } } diff --git a/seatunnel-connectors-v2/connector-amazondocumentdb/src/test/java/org/apache/seatunnel/connectors/seatunnel/amazondocumentdb/source/AmazonDocumentDBSourceSplitEnumeratorTest.java b/seatunnel-connectors-v2/connector-amazondocumentdb/src/test/java/org/apache/seatunnel/connectors/seatunnel/amazondocumentdb/source/AmazonDocumentDBSourceSplitEnumeratorTest.java index bfb22403fd..00bb75b323 100644 --- a/seatunnel-connectors-v2/connector-amazondocumentdb/src/test/java/org/apache/seatunnel/connectors/seatunnel/amazondocumentdb/source/AmazonDocumentDBSourceSplitEnumeratorTest.java +++ b/seatunnel-connectors-v2/connector-amazondocumentdb/src/test/java/org/apache/seatunnel/connectors/seatunnel/amazondocumentdb/source/AmazonDocumentDBSourceSplitEnumeratorTest.java @@ -135,6 +135,47 @@ public class AmazonDocumentDBSourceSplitEnumeratorTest { Assertions.assertTrue(context.noMoreSplitReaders.contains(0)); } + @Test + public void testSplitOwnerRoutesSplitIdsByBucketIndex() { + RecordingContext context = new RecordingContext(3, new HashSet<>(Arrays.asList(0, 1, 2))); + AmazonDocumentDBSourceSplitEnumerator enumerator = + new AmazonDocumentDBSourceSplitEnumerator(context, null, "{}", null); + // bucketIndex(4, 3) is 1 and bucketIndex(5, 3) is 2, so two splits handed back together + // are owned by different readers. + enumerator.addSplitsBack( + Arrays.asList( + new AmazonDocumentDBSourceSplit(4, "{}", null), + new AmazonDocumentDBSourceSplit(5, "{}", null)), + 0); + + enumerator.registerReader(1); + enumerator.registerReader(2); + + Assertions.assertEquals(1, context.assignedSplits.get(1).size()); + Assertions.assertEquals("4", context.assignedSplits.get(1).get(0).splitId()); + Assertions.assertEquals(1, context.assignedSplits.get(2).size()); + Assertions.assertEquals("5", context.assignedSplits.get(2).get(0).splitId()); + } + + @Test + public void testSplitOwnerKeepsIntegerMinValueSplitIdInRange() { + RecordingContext context = new RecordingContext(3, new HashSet<>(Arrays.asList(0, 1, 2))); + AmazonDocumentDBSourceSplitEnumerator enumerator = + new AmazonDocumentDBSourceSplitEnumerator(context, null, "{}", null); + // Clearing the sign bit is what keeps this in range. Math.abs(Integer.MIN_VALUE) is + // itself negative, so routing by absolute value would yield a negative owner index. + enumerator.addSplitsBack( + Collections.singletonList( + new AmazonDocumentDBSourceSplit(Integer.MIN_VALUE, "{}", null)), + 0); + + enumerator.registerReader(0); + + Assertions.assertEquals(1, context.assignedSplits.get(0).size()); + Assertions.assertEquals( + String.valueOf(Integer.MIN_VALUE), context.assignedSplits.get(0).get(0).splitId()); + } + private static class RecordingContext implements SourceSplitEnumerator.Context<AmazonDocumentDBSourceSplit> {
