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-11987-eae7c236fada9458caf4c5d7cfa25af9a038142c in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit be53a1d3d1e3becba7d4386516b7de14fdc0aa7e Author: Sepuri Sai Krishna <[email protected]> AuthorDate: Tue Sep 1 10:40:59 2026 +0000 [Improve][Common] Add HashUtils.bucketIndex for hash-to-bucket routing (#11987) --- .../apache/seatunnel/common/utils/HashUtils.java | 79 ++++++++++++ .../seatunnel/common/utils/HashUtilsTest.java | 140 +++++++++++++++++++++ .../clickhouse/sink/client/ShardRouter.java | 16 ++- .../fluss/source/FlussSourceSplitEnumerator.java | 3 +- .../source/BigtableSourceSplitEnumerator.java | 5 +- .../source/BigtableSourceSplitEnumeratorTest.java | 3 +- .../hbase/source/HbaseSourceSplitEnumerator.java | 3 +- .../source/enumerator/AbstractSplitEnumerator.java | 3 +- .../jdbc/source/JdbcSourceSplitEnumerator.java | 3 +- .../jdbc/source/JdbcSourceSplitEnumeratorTest.java | 3 +- .../kafka/sink/MessageContentPartitioner.java | 4 +- .../kafka/source/KafkaSourceSplitEnumerator.java | 3 +- .../source/enumerator/MongodbSplitEnumerator.java | 3 +- .../source/enumerator/AbstractSplitEnumerator.java | 3 +- .../source/enumerator/PulsarSplitEnumerator.java | 3 +- .../source/RocketMqSourceSplitEnumerator.java | 3 +- .../HazelcastMetricsSnapshotStateStore.java | 3 +- 17 files changed, 256 insertions(+), 24 deletions(-) diff --git a/seatunnel-common/src/main/java/org/apache/seatunnel/common/utils/HashUtils.java b/seatunnel-common/src/main/java/org/apache/seatunnel/common/utils/HashUtils.java new file mode 100644 index 0000000000..9422dfb570 --- /dev/null +++ b/seatunnel-common/src/main/java/org/apache/seatunnel/common/utils/HashUtils.java @@ -0,0 +1,79 @@ +/* + * 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. + */ + +package org.apache.seatunnel.common.utils; + +/** Utilities for mapping hash codes onto a fixed number of buckets. */ +public final class HashUtils { + + private HashUtils() {} + + /** + * Maps an arbitrary hash code onto {@code [0, bucketCount)}. + * + * <p>Plain {@code hash % bucketCount} is not usable for bucket routing: {@code %} keeps the + * sign of its left operand, so a negative hash yields a negative index for every hash that is + * not an exact multiple of {@code bucketCount}. {@code Math.abs(hash) % bucketCount} is not a + * fix either, because {@code Math.abs(Integer.MIN_VALUE)} overflows back to {@code + * Integer.MIN_VALUE} and still leaves the result negative. Clearing the sign bit has neither + * problem. + * + * <p>Note that this is deliberately <em>not</em> equivalent to {@code Math.floorMod(hash, + * bucketCount)}. Masking clears the sign bit before the division, which adds {@code 2^31} to a + * negative hash, so the two agree only when {@code bucketCount} is a power of two: for {@code + * hash = -5} and {@code bucketCount = 3} masking yields {@code 0} while {@code floorMod} yields + * {@code 1}. Call sites that route splits or partitions by hash must not be switched between + * the two spellings, as that would silently reassign ownership across a version upgrade. + * + * @param hash any hash code, including negative values and {@link Integer#MIN_VALUE} + * @param bucketCount the number of buckets, must be greater than zero + * @return a bucket index in {@code [0, bucketCount)} + * @throws IllegalArgumentException if {@code bucketCount} is not greater than zero + */ + public static int bucketIndex(int hash, int bucketCount) { + checkBucketCount(bucketCount); + return (hash & Integer.MAX_VALUE) % bucketCount; + } + + /** + * Maps an arbitrary 64-bit hash onto {@code [0, bucketCount)}. + * + * <p>Behaves as {@link #bucketIndex(int, int)} does, clearing the sign bit before the division + * so that {@link Long#MIN_VALUE} is handled without overflow. The result always fits in an + * {@code int} because it is smaller than {@code bucketCount}. + * + * <p>This is a distinct mapping from the {@code int} overload rather than a widening of it: a + * 64-bit hash and its truncation to 32 bits generally land in different buckets, so a call site + * must not be switched between the two overloads. + * + * @param hash any 64-bit hash, including negative values and {@link Long#MIN_VALUE} + * @param bucketCount the number of buckets, must be greater than zero + * @return a bucket index in {@code [0, bucketCount)} + * @throws IllegalArgumentException if {@code bucketCount} is not greater than zero + */ + public static int bucketIndex(long hash, int bucketCount) { + checkBucketCount(bucketCount); + return (int) ((hash & Long.MAX_VALUE) % bucketCount); + } + + private static void checkBucketCount(int bucketCount) { + if (bucketCount <= 0) { + throw new IllegalArgumentException( + "bucketCount must be greater than zero, but was " + bucketCount); + } + } +} diff --git a/seatunnel-common/src/test/java/org/apache/seatunnel/common/utils/HashUtilsTest.java b/seatunnel-common/src/test/java/org/apache/seatunnel/common/utils/HashUtilsTest.java new file mode 100644 index 0000000000..aac92f9762 --- /dev/null +++ b/seatunnel-common/src/test/java/org/apache/seatunnel/common/utils/HashUtilsTest.java @@ -0,0 +1,140 @@ +/* + * 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. + */ + +package org.apache.seatunnel.common.utils; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +public class HashUtilsTest { + + @Test + public void testIntegerMinValueIsMappedIntoRange() { + // Math.abs(Integer.MIN_VALUE) overflows back to Integer.MIN_VALUE, so the naive + // Math.abs(hash) % bucketCount spelling still returns a negative index here. + Assertions.assertTrue(Math.abs(Integer.MIN_VALUE) % 3 < 0); + + // Clearing the sign bit leaves 0, which is in range for every bucket count. + for (int bucketCount : new int[] {1, 2, 3, 7, 8, 100}) { + Assertions.assertEquals(0, HashUtils.bucketIndex(Integer.MIN_VALUE, bucketCount)); + } + Assertions.assertEquals(1, HashUtils.bucketIndex(Integer.MIN_VALUE + 1, 3)); + } + + @Test + public void testNegativeHashesStayInRange() { + for (int bucketCount : new int[] {1, 2, 3, 7, 8, 100}) { + for (int hash : + new int[] {-1, -5, -31, -12345, Integer.MIN_VALUE, Integer.MIN_VALUE + 1}) { + int bucket = HashUtils.bucketIndex(hash, bucketCount); + Assertions.assertTrue( + bucket >= 0 && bucket < bucketCount, + "hash=" + hash + " bucketCount=" + bucketCount + " bucket=" + bucket); + } + } + } + + @Test + public void testMatchesExistingMaskedSpelling() { + for (int hash = -1000; hash <= 1000; hash++) { + for (int bucketCount : new int[] {1, 2, 3, 7, 8, 16, 100}) { + Assertions.assertEquals( + (hash & Integer.MAX_VALUE) % bucketCount, + HashUtils.bucketIndex(hash, bucketCount), + "hash=" + hash + " bucketCount=" + bucketCount); + } + } + } + + @Test + public void testSingleBucketAlwaysZero() { + Assertions.assertEquals(0, HashUtils.bucketIndex(Integer.MIN_VALUE, 1)); + Assertions.assertEquals(0, HashUtils.bucketIndex(Integer.MAX_VALUE, 1)); + Assertions.assertEquals(0, HashUtils.bucketIndex(0, 1)); + } + + @Test + public void testNonPositiveBucketCountRejected() { + Assertions.assertThrows(IllegalArgumentException.class, () -> HashUtils.bucketIndex(42, 0)); + Assertions.assertThrows( + IllegalArgumentException.class, () -> HashUtils.bucketIndex(42, -1)); + } + + @Test + public void testNotEquivalentToFloorModForNonPowerOfTwo() { + // Guards the invariant that made this a helper rather than a mechanical rename: + // masking and floorMod agree only when bucketCount is a power of two. + Assertions.assertEquals(0, HashUtils.bucketIndex(-5, 3)); + Assertions.assertEquals(1, Math.floorMod(-5, 3)); + + for (int bucketCount : new int[] {2, 4, 8, 16, 1024}) { + for (int hash = -500; hash <= 500; hash++) { + Assertions.assertEquals( + Math.floorMod(hash, bucketCount), + HashUtils.bucketIndex(hash, bucketCount), + "hash=" + hash + " bucketCount=" + bucketCount); + } + } + } + + @Test + public void testLongHashesStayInRange() { + for (int bucketCount : new int[] {1, 2, 3, 7, 8, 100}) { + for (long hash : + new long[] { + -1L, -5L, -31L, -12345L, Long.MIN_VALUE, Long.MIN_VALUE + 1, Long.MAX_VALUE + }) { + int bucket = HashUtils.bucketIndex(hash, bucketCount); + Assertions.assertTrue( + bucket >= 0 && bucket < bucketCount, + "hash=" + hash + " bucketCount=" + bucketCount + " bucket=" + bucket); + } + } + } + + @Test + public void testLongMatchesExistingMaskedSpelling() { + // The clickhouse shard router already spelled this out inline; the helper must not + // change which shard a value routes to. + for (long hash = -1000L; hash <= 1000L; hash++) { + for (int bucketCount : new int[] {1, 2, 3, 7, 8, 16, 100}) { + Assertions.assertEquals( + (int) ((hash & Long.MAX_VALUE) % bucketCount), + HashUtils.bucketIndex(hash, bucketCount), + "hash=" + hash + " bucketCount=" + bucketCount); + } + } + Assertions.assertEquals(0, HashUtils.bucketIndex(Long.MIN_VALUE, 3)); + } + + @Test + public void testLongOverloadIsDistinctFromIntOverload() { + // A 64-bit hash and its truncation to 32 bits generally land in different buckets, + // so the two overloads are separate mappings and call sites must not be switched. + long hash = 1L << 32; + Assertions.assertEquals(1, HashUtils.bucketIndex(hash, 3)); + Assertions.assertEquals(0, HashUtils.bucketIndex((int) hash, 3)); + } + + @Test + public void testLongNonPositiveBucketCountRejected() { + Assertions.assertThrows( + IllegalArgumentException.class, () -> HashUtils.bucketIndex(42L, 0)); + Assertions.assertThrows( + IllegalArgumentException.class, () -> HashUtils.bucketIndex(42L, -1)); + } +} diff --git a/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/ShardRouter.java b/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/ShardRouter.java index d04c95c9a9..dc903b4360 100644 --- a/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/ShardRouter.java +++ b/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/ShardRouter.java @@ -19,6 +19,7 @@ package org.apache.seatunnel.connectors.seatunnel.clickhouse.sink.client; import org.apache.seatunnel.shade.org.apache.commons.lang3.StringUtils; +import org.apache.seatunnel.common.utils.HashUtils; import org.apache.seatunnel.connectors.seatunnel.clickhouse.exception.ClickhouseConnectorErrorCode; import org.apache.seatunnel.connectors.seatunnel.clickhouse.exception.ClickhouseConnectorException; import org.apache.seatunnel.connectors.seatunnel.clickhouse.shard.Shard; @@ -112,15 +113,12 @@ public class ShardRouter implements Serializable { return shards.lowerEntry(threadLocalRandom.nextInt(shardWeightCount) + 1).getValue(); } int offset = - (int) - ((HASH_INSTANCE.hash( - ByteBuffer.wrap( - shardValue - .toString() - .getBytes(StandardCharsets.UTF_8)), - 0) - & Long.MAX_VALUE) - % shardWeightCount); + HashUtils.bucketIndex( + HASH_INSTANCE.hash( + ByteBuffer.wrap( + shardValue.toString().getBytes(StandardCharsets.UTF_8)), + 0), + shardWeightCount); return shards.lowerEntry(offset + 1).getValue(); } diff --git a/seatunnel-connectors-v2/connector-fluss/src/main/java/org/apache/seatunnel/connectors/seatunnel/fluss/source/FlussSourceSplitEnumerator.java b/seatunnel-connectors-v2/connector-fluss/src/main/java/org/apache/seatunnel/connectors/seatunnel/fluss/source/FlussSourceSplitEnumerator.java index 59629a57d5..29a0990c5f 100644 --- a/seatunnel-connectors-v2/connector-fluss/src/main/java/org/apache/seatunnel/connectors/seatunnel/fluss/source/FlussSourceSplitEnumerator.java +++ b/seatunnel-connectors-v2/connector-fluss/src/main/java/org/apache/seatunnel/connectors/seatunnel/fluss/source/FlussSourceSplitEnumerator.java @@ -18,6 +18,7 @@ package org.apache.seatunnel.connectors.seatunnel.fluss.source; import org.apache.seatunnel.api.source.SourceSplitEnumerator; import org.apache.seatunnel.api.table.catalog.TablePath; +import org.apache.seatunnel.common.utils.HashUtils; import org.apache.seatunnel.connectors.seatunnel.fluss.config.StartMode; import com.alibaba.fluss.client.table.scanner.log.LogScanner; @@ -169,7 +170,7 @@ public class FlussSourceSplitEnumerator private static int getSplitOwner(FlussSourceSplit split, int parallelism) { int hash = split.getTablePath().getFullName().hashCode() * 31 + split.getBucketId(); - return (hash & Integer.MAX_VALUE) % parallelism; + return HashUtils.bucketIndex(hash, parallelism); } @Override diff --git a/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceSplitEnumerator.java b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceSplitEnumerator.java index b8d3f2b488..56654e9460 100644 --- a/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceSplitEnumerator.java +++ b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceSplitEnumerator.java @@ -18,6 +18,7 @@ package org.apache.seatunnel.connectors.seatunnel.bigtable.source; import org.apache.seatunnel.api.source.SourceSplitEnumerator; +import org.apache.seatunnel.common.utils.HashUtils; import org.apache.seatunnel.connectors.seatunnel.bigtable.client.BigtableClient; import org.apache.seatunnel.connectors.seatunnel.bigtable.config.BigtableParameters; @@ -383,8 +384,8 @@ public class BigtableSourceSplitEnumerator } else { for (BigtableSourceSplit split : pendingSplits) { int owner = - (split.splitId().hashCode() & Integer.MAX_VALUE) - % context.currentParallelism(); + HashUtils.bucketIndex( + split.splitId().hashCode(), context.currentParallelism()); if (owner == taskId) { toAssign.add(split); } diff --git a/seatunnel-connectors-v2/connector-google-bigtable/src/test/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceSplitEnumeratorTest.java b/seatunnel-connectors-v2/connector-google-bigtable/src/test/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceSplitEnumeratorTest.java index c69e641672..57e38c2f2f 100644 --- a/seatunnel-connectors-v2/connector-google-bigtable/src/test/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceSplitEnumeratorTest.java +++ b/seatunnel-connectors-v2/connector-google-bigtable/src/test/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceSplitEnumeratorTest.java @@ -21,6 +21,7 @@ import org.apache.seatunnel.api.common.metrics.MetricsContext; import org.apache.seatunnel.api.event.EventListener; import org.apache.seatunnel.api.source.SourceEvent; import org.apache.seatunnel.api.source.SourceSplitEnumerator; +import org.apache.seatunnel.common.utils.HashUtils; import org.apache.seatunnel.connectors.seatunnel.bigtable.client.BigtableClient; import org.apache.seatunnel.connectors.seatunnel.bigtable.config.BigtableParameters; import org.apache.seatunnel.connectors.seatunnel.bigtable.exception.BigtableConnectorErrorCode; @@ -287,7 +288,7 @@ public class BigtableSourceSplitEnumeratorTest { // Each split's owner must match hash(splitId) % parallelism. for (int i = 0; i < parallelism; i++) { for (BigtableSourceSplit split : context.getAssignedSplits(i)) { - int expected = (split.splitId().hashCode() & Integer.MAX_VALUE) % parallelism; + int expected = HashUtils.bucketIndex(split.splitId().hashCode(), parallelism); assertEquals( expected, i, diff --git a/seatunnel-connectors-v2/connector-hbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/hbase/source/HbaseSourceSplitEnumerator.java b/seatunnel-connectors-v2/connector-hbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/hbase/source/HbaseSourceSplitEnumerator.java index 744130240a..23cb742e69 100644 --- a/seatunnel-connectors-v2/connector-hbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/hbase/source/HbaseSourceSplitEnumerator.java +++ b/seatunnel-connectors-v2/connector-hbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/hbase/source/HbaseSourceSplitEnumerator.java @@ -21,6 +21,7 @@ package org.apache.seatunnel.connectors.seatunnel.hbase.source; import org.apache.seatunnel.shade.com.google.common.annotations.VisibleForTesting; import org.apache.seatunnel.api.source.SourceSplitEnumerator; +import org.apache.seatunnel.common.utils.HashUtils; import org.apache.seatunnel.connectors.seatunnel.hbase.client.HbaseClient; import org.apache.seatunnel.connectors.seatunnel.hbase.config.HbaseParameters; import org.apache.seatunnel.connectors.seatunnel.hbase.exception.HbaseConnectorErrorCode; @@ -320,6 +321,6 @@ public class HbaseSourceSplitEnumerator /** Hash algorithm for assigning splits to readers */ private static int getSplitOwner(String tp, int numReaders) { - return (tp.hashCode() & Integer.MAX_VALUE) % numReaders; + return HashUtils.bucketIndex(tp.hashCode(), numReaders); } } diff --git a/seatunnel-connectors-v2/connector-iceberg/src/main/java/org/apache/seatunnel/connectors/seatunnel/iceberg/source/enumerator/AbstractSplitEnumerator.java b/seatunnel-connectors-v2/connector-iceberg/src/main/java/org/apache/seatunnel/connectors/seatunnel/iceberg/source/enumerator/AbstractSplitEnumerator.java index 854bc15f33..2b5afc264e 100644 --- a/seatunnel-connectors-v2/connector-iceberg/src/main/java/org/apache/seatunnel/connectors/seatunnel/iceberg/source/enumerator/AbstractSplitEnumerator.java +++ b/seatunnel-connectors-v2/connector-iceberg/src/main/java/org/apache/seatunnel/connectors/seatunnel/iceberg/source/enumerator/AbstractSplitEnumerator.java @@ -22,6 +22,7 @@ import org.apache.seatunnel.shade.org.apache.commons.lang3.tuple.Pair; import org.apache.seatunnel.api.source.SourceSplitEnumerator; import org.apache.seatunnel.api.table.catalog.CatalogTable; import org.apache.seatunnel.api.table.catalog.TablePath; +import org.apache.seatunnel.common.utils.HashUtils; import org.apache.seatunnel.connectors.seatunnel.iceberg.IcebergCatalogLoader; import org.apache.seatunnel.connectors.seatunnel.iceberg.config.IcebergSourceConfig; import org.apache.seatunnel.connectors.seatunnel.iceberg.source.split.IcebergFileScanTaskSplit; @@ -190,7 +191,7 @@ public abstract class AbstractSplitEnumerator } private static int getSplitOwner(String splitId, int numReaders) { - return (splitId.hashCode() & Integer.MAX_VALUE) % numReaders; + return HashUtils.bucketIndex(splitId.hashCode(), numReaders); } protected void addPendingSplits(Collection<IcebergFileScanTaskSplit> newSplits) { diff --git a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/JdbcSourceSplitEnumerator.java b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/JdbcSourceSplitEnumerator.java index a0948dfe0d..65a599f92a 100644 --- a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/JdbcSourceSplitEnumerator.java +++ b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/JdbcSourceSplitEnumerator.java @@ -20,6 +20,7 @@ package org.apache.seatunnel.connectors.seatunnel.jdbc.source; import org.apache.seatunnel.api.source.SourceSplitEnumerator; import org.apache.seatunnel.api.table.catalog.TablePath; import org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated; +import org.apache.seatunnel.common.utils.HashUtils; import org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcSourceConfig; import org.apache.seatunnel.connectors.seatunnel.jdbc.exception.JdbcConnectorException; import org.apache.seatunnel.connectors.seatunnel.jdbc.state.JdbcSourceState; @@ -173,6 +174,6 @@ public class JdbcSourceSplitEnumerator } private static int getSplitOwner(String tp, int numReaders) { - return (tp.hashCode() & Integer.MAX_VALUE) % numReaders; + return HashUtils.bucketIndex(tp.hashCode(), numReaders); } } diff --git a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/JdbcSourceSplitEnumeratorTest.java b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/JdbcSourceSplitEnumeratorTest.java index 9f0748d320..ce2d7629fd 100644 --- a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/JdbcSourceSplitEnumeratorTest.java +++ b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/JdbcSourceSplitEnumeratorTest.java @@ -25,6 +25,7 @@ import org.apache.seatunnel.api.table.catalog.CatalogTable; import org.apache.seatunnel.api.table.catalog.TableIdentifier; import org.apache.seatunnel.api.table.catalog.TablePath; import org.apache.seatunnel.api.table.catalog.TableSchema; +import org.apache.seatunnel.common.utils.HashUtils; import org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcConnectionConfig; import org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcSourceConfig; @@ -194,7 +195,7 @@ class JdbcSourceSplitEnumeratorTest { Assertions.assertEquals(tables.size(), assignedSplitOwners.size()); assignedSplitOwners.forEach( (splitId, owner) -> { - int expectedOwner = (splitId.hashCode() & Integer.MAX_VALUE) % parallelism; + int expectedOwner = HashUtils.bucketIndex(splitId.hashCode(), parallelism); Assertions.assertEquals(expectedOwner, owner); }); diff --git a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/MessageContentPartitioner.java b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/MessageContentPartitioner.java index 907e87c583..7b40c286ed 100644 --- a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/MessageContentPartitioner.java +++ b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/MessageContentPartitioner.java @@ -17,6 +17,8 @@ package org.apache.seatunnel.connectors.seatunnel.kafka.sink; +import org.apache.seatunnel.common.utils.HashUtils; + import org.apache.kafka.clients.producer.Partitioner; import org.apache.kafka.common.Cluster; import org.apache.kafka.common.PartitionInfo; @@ -49,7 +51,7 @@ public class MessageContentPartitioner implements Partitioner { } } // Choose one of the remaining partitions according to the hashcode. - return ((message.hashCode() & Integer.MAX_VALUE) % (numPartitions - assignPartitionsSize)) + return HashUtils.bucketIndex(message.hashCode(), numPartitions - assignPartitionsSize) + assignPartitionsSize; } diff --git a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceSplitEnumerator.java b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceSplitEnumerator.java index c4260ce588..e7eab6e0ad 100644 --- a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceSplitEnumerator.java +++ b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceSplitEnumerator.java @@ -22,6 +22,7 @@ import org.apache.seatunnel.shade.com.google.common.annotations.VisibleForTestin import org.apache.seatunnel.api.source.SourceSplitEnumerator; import org.apache.seatunnel.api.table.catalog.TablePath; import org.apache.seatunnel.common.config.Common; +import org.apache.seatunnel.common.utils.HashUtils; import org.apache.seatunnel.connectors.seatunnel.kafka.KafkaClientUtils; import org.apache.seatunnel.connectors.seatunnel.kafka.exception.KafkaConnectorErrorCode; import org.apache.seatunnel.connectors.seatunnel.kafka.exception.KafkaConnectorException; @@ -446,7 +447,7 @@ public class KafkaSourceSplitEnumerator } private static int getSplitOwner(TopicPartition tp, int numReaders) { - int startIndex = ((tp.topic().hashCode() * 31) & 0x7FFFFFFF) % numReaders; + int startIndex = HashUtils.bucketIndex(tp.topic().hashCode() * 31, numReaders); return (startIndex + tp.partition()) % numReaders; } diff --git a/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/source/enumerator/MongodbSplitEnumerator.java b/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/source/enumerator/MongodbSplitEnumerator.java index 423092d466..d28bdd10df 100644 --- a/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/source/enumerator/MongodbSplitEnumerator.java +++ b/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/source/enumerator/MongodbSplitEnumerator.java @@ -21,6 +21,7 @@ import org.apache.seatunnel.shade.com.google.common.collect.Lists; import org.apache.seatunnel.api.source.SourceSplitEnumerator; import org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated; +import org.apache.seatunnel.common.utils.HashUtils; import org.apache.seatunnel.connectors.seatunnel.mongodb.exception.MongodbConnectorException; import org.apache.seatunnel.connectors.seatunnel.mongodb.internal.MongodbClientProvider; import org.apache.seatunnel.connectors.seatunnel.mongodb.source.split.MongoSplit; @@ -167,6 +168,6 @@ public class MongodbSplitEnumerator } private static int getSplitOwner(String tp, int numReaders) { - return (tp.hashCode() & Integer.MAX_VALUE) % numReaders; + return HashUtils.bucketIndex(tp.hashCode(), numReaders); } } diff --git a/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/enumerator/AbstractSplitEnumerator.java b/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/enumerator/AbstractSplitEnumerator.java index c9125cfbb1..5021e03f38 100644 --- a/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/enumerator/AbstractSplitEnumerator.java +++ b/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/enumerator/AbstractSplitEnumerator.java @@ -22,6 +22,7 @@ import org.apache.seatunnel.shade.com.google.common.util.concurrent.ThreadFactor import org.apache.seatunnel.api.source.SourceSplitEnumerator; import org.apache.seatunnel.common.constants.JobMode; +import org.apache.seatunnel.common.utils.HashUtils; import org.apache.seatunnel.common.utils.SeaTunnelException; import org.apache.seatunnel.connectors.seatunnel.paimon.source.PaimonSourceSplit; import org.apache.seatunnel.connectors.seatunnel.paimon.source.PaimonSourceSplitGenerator; @@ -237,7 +238,7 @@ public abstract class AbstractSplitEnumerator /** Hash algorithm for assigning splits to readers */ protected static int getSplitOwner(String tp, int numReaders) { - return (tp.hashCode() & Integer.MAX_VALUE) % numReaders; + return HashUtils.bucketIndex(tp.hashCode(), numReaders); } // ------------------------------------------------------------------------ diff --git a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/enumerator/PulsarSplitEnumerator.java b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/enumerator/PulsarSplitEnumerator.java index ab4b31d2b8..80ed21b90f 100644 --- a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/enumerator/PulsarSplitEnumerator.java +++ b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/enumerator/PulsarSplitEnumerator.java @@ -21,6 +21,7 @@ import org.apache.seatunnel.api.source.Boundedness; import org.apache.seatunnel.api.source.SourceSplitEnumerator; import org.apache.seatunnel.api.table.catalog.TablePath; import org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated; +import org.apache.seatunnel.common.utils.HashUtils; import org.apache.seatunnel.connectors.seatunnel.pulsar.config.PulsarAdminConfig; import org.apache.seatunnel.connectors.seatunnel.pulsar.config.PulsarConfigUtil; import org.apache.seatunnel.connectors.seatunnel.pulsar.exception.PulsarConnectorException; @@ -229,7 +230,7 @@ public class PulsarSplitEnumerator } static int getSplitOwner(TopicPartition tp, int numReaders) { - int startIndex = ((tp.getTopic().hashCode() * 31) & 0x7FFFFFFF) % numReaders; + int startIndex = HashUtils.bucketIndex(tp.getTopic().hashCode() * 31, numReaders); // here, the assumption is that the id of pulsar partitions are always ascending // starting from 0, and therefore can be used directly as the offset clockwise from the diff --git a/seatunnel-connectors-v2/connector-rocketmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/source/RocketMqSourceSplitEnumerator.java b/seatunnel-connectors-v2/connector-rocketmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/source/RocketMqSourceSplitEnumerator.java index e4b5bd6eff..4846d6c050 100644 --- a/seatunnel-connectors-v2/connector-rocketmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/source/RocketMqSourceSplitEnumerator.java +++ b/seatunnel-connectors-v2/connector-rocketmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/source/RocketMqSourceSplitEnumerator.java @@ -22,6 +22,7 @@ import org.apache.seatunnel.shade.com.google.common.collect.Sets; import org.apache.seatunnel.api.source.SourceSplitEnumerator; import org.apache.seatunnel.common.config.Common; +import org.apache.seatunnel.common.utils.HashUtils; import org.apache.seatunnel.connectors.seatunnel.rocketmq.common.RocketMqAdminUtil; import org.apache.seatunnel.connectors.seatunnel.rocketmq.common.StartMode; import org.apache.seatunnel.connectors.seatunnel.rocketmq.exception.RocketMqConnectorErrorCode; @@ -135,7 +136,7 @@ public class RocketMqSourceSplitEnumerator } private static int getSplitOwner(MessageQueue messageQueue, int numReaders) { - int startIndex = ((messageQueue.getQueueId() * 31) & 0x7FFFFFFF) % numReaders; + int startIndex = HashUtils.bucketIndex(messageQueue.getQueueId() * 31, numReaders); return (startIndex + messageQueue.getQueueId()) % numReaders; } diff --git a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/metrics/hazelcast/HazelcastMetricsSnapshotStateStore.java b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/metrics/hazelcast/HazelcastMetricsSnapshotStateStore.java index 20df3b848b..5c00c30de8 100644 --- a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/metrics/hazelcast/HazelcastMetricsSnapshotStateStore.java +++ b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/metrics/hazelcast/HazelcastMetricsSnapshotStateStore.java @@ -17,6 +17,7 @@ package org.apache.seatunnel.engine.server.common.statestore.metrics.hazelcast; +import org.apache.seatunnel.common.utils.HashUtils; import org.apache.seatunnel.engine.server.common.statestore.metrics.MetricsSnapshotStateStore; import org.apache.seatunnel.engine.server.dag.physical.PipelineLocation; import org.apache.seatunnel.engine.server.execution.TaskLocation; @@ -167,7 +168,7 @@ public class HazelcastMetricsSnapshotStateStore } private long partition(TaskLocation taskLocation) { - return (taskLocation.hashCode() & Integer.MAX_VALUE) % partitionCount; + return HashUtils.bucketIndex(taskLocation.hashCode(), partitionCount); } @Override
