apache / apache/seatunnel

[Feature][API] Add a shared non-negative modulo helper for hash-based routing

Open
#11,976 1 comment 0 reactions 0 assignees View on GitHub
Dominant language
Java
Stars
9.7k
Forks
2.4k
Avg merge
3d 13h
Merged PRs (30d)
203

Description

### Search before asking

- [X] I had searched in the [feature](https://github.com/apache/seatunnel/issues?q=is%3Aissue+label%3A%22Feature%22) and found no similar feature requirement.

### Description

Follow-up to #11720 / #11721, filed at the request of @SEZ9 and @DanielLeens as the agreed successor to finding `PR11721-F2`.

#11721 fixes a real crash in `MultiTableSinkWriter.write()`, where `Math.abs(object.hashCode()) % blockingQueues.size()` returned a **negative** queue index because `Math.abs(Integer.MIN_VALUE)` is still `Integer.MIN_VALUE`. That fix removes the last live instance of the bug. This issue is about the reason the bug survived in the first place: **there is no shared helper for "map a hash onto `[0, n)`", so every call site hand-rolls it, and the codebase currently contains four different spellings of the same idea, one of which was wrong.**

#### The four spellings in `dev` today

| # | Spelling | Main-source sites | Correct? |
|---|---|---|---|
| 1 | `Math.abs(h) % n` | 1 | **No** — negative result for `h == Integer.MIN_VALUE` when `n` is not a power of two |
| 2 | `(h & Integer.MAX_VALUE) % n` | 9 | Yes |
| 3 | `((h * 31) & 0x7FFFFFFF) % n` | 3 | Yes |
| 4 | `Math.floorMod(h, n)`, with an `n <= 0` guard | 1 | Yes |

Spelling 1 is `MultiTableSinkWriter.java:619`, fixed by #11721. Verified with a repo-wide scan of `dev`: after that PR lands, **zero** occurrences of `Math.abs(...) % ...` remain in main source.

#### Why it is worth a helper rather than leaving it

The 12 correct sites are correct by repetition, not by construction. Nothing prevents the 13th author from reaching for `Math.abs` again, which is exactly what happened here. A single well-named, tested helper turns a convention that must be remembered into one that is imported.

#### The call sites

Spelling 2, `(h & Integer.MAX_VALUE) % n` (9 main-source):

| File | Line |
|---|---|
| `seatunnel-connectors-v2/connector-paimon/.../source/enumerator/AbstractSplitEnumerator.java` | 240 |
| `seatunnel-connectors-v2/connector-hbase/.../source/HbaseSourceSplitEnumerator.java` | 323 |
| `seatunnel-connectors-v2/connector-fluss/.../source/FlussSourceSplitEnumerator.java` | 172 |
| `seatunnel-connectors-v2/connector-mongodb/.../source/enumerator/MongodbSplitEnumerator.java` | 170 |
| `seatunnel-connectors-v2/connector-iceberg/.../source/enumerator/AbstractSplitEnumerator.java` | 193 |
| `seatunnel-connectors-v2/connector-jdbc/.../source/JdbcSourceSplitEnumerator.java` | 176 |
| `seatunnel-connectors-v2/connector-google-bigtable/.../source/BigtableSourceSplitEnumerator.java` | 386 |
| `seatunnel-connectors-v2/connector-kafka/.../sink/MessageContentPartitioner.java` | 52 |
| `seatunnel-engine/seatunnel-engine-server/.../statestore/metrics/hazelcast/HazelcastMetricsSnapshotStateStore.java` | 170 |

Spelling 3, `((h * 31) & 0x7FFFFFFF) % n` (3 main-source):

| File | Line |
|---|---|
| `seatunnel-connectors-v2/connector-kafka/.../source/KafkaSourceSplitEnumerator.java` | 449 |
| `seatunnel-connectors-v2/connector-pulsar/.../source/enumerator/PulsarSplitEnumerator.java` | 232 |
| `seatunnel-connectors-v2/connector-rocketmq/.../source/RocketMqSourceSplitEnumerator.java` | 138 |

Spelling 4, the existing precedent, `connector-file`:

```java
// FileSourceDocumentRouting.java:51-57
public static int routeBucket(String documentId, int routeParallelism) {
if (routeParallelism <= 0) {
throw new IllegalArgumentException("routeParallelism must be greater than zero");
}
byte[] digest = sha256(documentId == null ? "" : documentId);
return Math.floorMod(ByteBuffer.wrap(digest).getInt(), routeParallelism);
}
```

Two test files also spell the expression out to mirror the code under test (`BigtableSourceSplitEnumeratorTest.java:290`, `JdbcSourceSplitEnumeratorTest.java:197`); they would follow whatever the production sites do.

#### The part that needs a decision before any code is written

**Spellings 2 and 4 are not the same function.** Masking the sign bit and taking the true mathematical modulus both land in `[0, n)`, but they do not agree on negative hashes when `n` is not a power of two:

```
h = Integer.MIN_VALUE n=3 mask=0 floorMod=1 DIFFER
h = Integer.MIN_VALUE n=4 mask=0 floorMod=0 same
h = -5 n=3 mask=0 floorMod=1 DIFFER
h = -5 n=4 mask=3 floorMod=3 same
h = -1 n=3 mask=1 floorMod=2 DIFFER
```

(Verified by running the three expressions on a JDK, not derived on paper.)

So a naive "replace all 12 sites with `Math.floorMod`" would be a behavioural change, not a refactor: it silently **reshuffles which reader owns which split** at every source enumerator in the list. For an ephemeral sink queue like `MultiTableSinkWriter` that is harmless, but for split enumerators it changes assignment across an upgrade, and per the project's backward-compatibility rules that deserves deliberate handling rather than being smuggled in under a cleanup.

My proposal, which keeps the change a pure refactor:

- Adopt the **masking** semantics (spelling 2) for the shared helper, since that is what 9 of the 12 sites already do, so migrating them is byte-for-byte behaviour-preserving.
- Leave `FileSourceDocumentRouting.routeBucket` alone. It is already correct, already guarded, already shared within its module, and its `floorMod` semantics are part of a documented routing contract (#10914). Changing it would alter document-to-reader mapping for no benefit.
- Treat spelling 3 (`* 31` then mask) as a separate judgement call. The `* 31` pre-mix is not redundant, it changes the distribution, so those three sites should either keep their pre-mix and use the helper only for the mask-and-mod step, or be left as they are. I would not fold them in silently.

#### Proposed helper

Placement: `seatunnel-common` (or `seatunnel-api`), so both connectors and the engine can use it without a new dependency edge. No new third-party dependency.

```java
/**
* Maps a hash code onto the range {@code [0, bucketCount)}.
*
*

Clears the sign bit rather than using {@link Math#abs}: {@code Math.abs(Integer.MIN_VALUE)}
* is still {@code Integer.MIN_VALUE}, which yields a negative index whenever {@code bucketCount}
* is not a power of two.
*
* @param hash any int, including {@link Integer#MIN_VALUE}
* @param bucketCount number of buckets, must be positive
* @return a bucket index in {@code [0, bucketCount)}
* @throws IllegalArgumentException if {@code bucketCount <= 0}
*/
public static int nonNegativeMod(int hash, int bucketCount) {
if (bucketCount <= 0) {
throw new IllegalArgumentException(
"bucketCount must be greater than zero, but was " + bucketCount);
}
return (hash & Integer.MAX_VALUE) % bucketCount;
}
```

The `bucketCount <= 0` guard is deliberate and mirrors `routeBucket`. Several current sites divide by a parallelism or partition count with no lower-bound validation, so a misconfigured value surfaces as a bare `ArithmeticException: / by zero` with no indication of which knob was wrong. Routing it through one helper gives that class of misconfiguration a single, nameable failure message.

Tests, as requested by @SEZ9 on #11721:

- `nonNegativeMod(Integer.MIN_VALUE, n)` is in `[0, n)` for `n` in 1..64, the regression that motivated this
- `nonNegativeMod(h, n)` is in `[0, n)` for a broad sweep of `h` including `Integer.MIN_VALUE`, `Integer.MAX_VALUE`, `-1`, `0`
- equivalence with the pre-existing expression at every migrated site, so the refactor is provably behaviour-preserving
- `bucketCount <= 0` throws `IllegalArgumentException`

#### Out of scope

`ShardRouter.java:122` in `connector-clickhouse` does the same thing for a `long` hash (`& Long.MAX_VALUE` then `%`, then a cast). It is correct as written. A `nonNegativeMod(long, int)` overload could cover it later, but I am not proposing to touch it here.

### Usage Scenario

This is a maintainability and defect-prevention change; it is not user-facing and changes no configuration or output. The user-visible bug that motivated it is already described in #11720 and fixed by #11721.

The scenario it guards against is the one that just occurred: a contributor adding a new source enumerator or sink partitioner writes `Math.abs(key.hashCode()) % parallelism`, it passes review because it reads correctly, and it fails in production only for the one key in 2^32 that hashes to `Integer.MIN_VALUE`, and only when parallelism is not a power of two, which is why it can sit undetected for years.

### Related issues

- #11720 — the bug report
- #11721 — the fix, where this was agreed as follow-up `PR11721-F2`
- #10914 — the document-routing design that introduced `FileSourceDocumentRouting.routeBucket`

### Are you willing to submit a PR?

- [X] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct)

Contributor guide

No contributing guide indexed for this repository

Research direction

Start by reviewing the listed source enumerator and partitioner call sites, the existing FileSourceDocumentRouting.routeBucket precedent, and the two named tests. Resolve the masking-semantics and helper-placement decisions before changing code. Done means a shared guarded helper, behavior-preserving migrations where agreed, and tests covering Integer.MIN_VALUE, range bounds, equivalence, and invalid bucket counts.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
backend, distributed-systems
Issue type
Feature
Difficulty
4/5
Estimated time
3-5 days
Activity status
Active
Clarity
Mostly clear
Newbie friendliness
55/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.