[FLINK-40331][cdc-connect][fluss] Support forward shuffle in fluss sink.

[loserwang1024]

What is the purpose of this pull request?

https://issues.apache.org/jira/browse/FLINK-40331

Brief change log

Add sink.partitioning.strategy in fluss sink to support FORWARD.

---

Verifying this change

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change added tests and can be verified as follows:

• Added/Updated unit tests in ...*
• Added/Updated integration tests in ...*
• Manually tested by ...

Documentation

• Does this pull request introduce a new feature? (yes / no)
• If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

[lvyanquan]

Hi @loserwang1024, thanks for the PR! One suggestion on the HashFunction change: instead of a raw int sourceIndex, consider passing a small immutable HashContext object:

@Internal
public class HashContext {

    private final int sourceSubtaskIndex;
    private final int downstreamParallelism;

    public HashContext(int sourceSubtaskIndex, int downstreamParallelism) { ... }

    public int getSourceSubtaskIndex() { ... }

    public int getDownstreamParallelism() { ... }
}

default int hashcode(HashContext context, T event) {
    return hashcode(event);
}

Reasons:

1. Self-documenting call sites — hashcode(subTaskId, event) gives no hint what the bare int means; hashcode(hashContext, event) does.
2. Extensible without signature churn — future strategies (bucket-aware mapping, fallback when sourceIndex >= downstreamParallelism) can read from the context instead of stacking more int overloads.
3. Fails fast instead of silently skewing — ForwardHashFunction.hashcode(event) can throw UnsupportedOperationException rather than returning 0 and routing everything to subtask 0 if someone calls the single-arg method by mistake.

The context can be created once in open() and reused, so there's no per-record allocation cost. Since HashFunction is @Internal, this is the cheapest time to make the change. WDYT?