1996fanrui opened a new pull request, #29272:
URL: https://github.com/apache/flink/pull/29272
## What is the purpose of the change
When the producer and consumer of a FORWARD edge end up with different
parallelism (e.g. after rescaling via the AdaptiveScheduler or
`pipeline.jobvertex-parallelism-overrides`), `StreamTask` silently replaces the
`ForwardPartitioner` with a `RebalancePartitioner` (FLINK-30213). That's fine
for append-only streams, but it reorders records in changelog streams and
produces wrong results. This PR makes the behavior configurable.
## Brief change log
- Add `pipeline.forward-edge.parallelism-mismatch-mode` (enum):
- `REBALANCE` (default): current behavior.
- `FAIL`: fail the job with an exception.
- `KEEP_FORWARD`: keep the forward partitioner (order is preserved, and
records go to a single consumer subtask).
- `StreamTask#replaceForwardPartitionerIfConsumerParallelismDoesNotMatch`
applies the configured mode.
## Verifying this change
- Unit tests in `StreamTaskTest` cover all three modes.
- End-to-end: I verified all three modes with a real job on a MiniCluster
running the AdaptiveScheduler. The job is `source -> forward -> sink` with
chaining disabled. It starts at parallelism 1, and the sink is then rescaled to
2 through the resource-requirements REST API, which creates the mismatched
FORWARD edge. All three modes behaved as expected: `REBALANCE` spreads records
across both sink subtasks, `KEEP_FORWARD` sends everything to sink subtask 0,
and `FAIL` fails the job with the new exception. The ITCase is on a separate
branch and not part of this PR:
https://github.com/1996fanrui/flink/tree/FLINK-40780-e2e-itcase
## Does this pull request potentially affect one of the following parts
- Dependencies: no
- The public API: yes (new `@PublicEvolving` config option)
- The serializers: no
- The runtime per-record code paths: no (the check runs once when record
writers are created)
- Anything that affects deployment or recovery: no
- The S3 file system connector: no
## Documentation
- New feature: yes
- Docs: the config docs were regenerated (`pipeline_configuration.html`)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
