[flink] Skip unnecessary undo recovery for idempotent aggregations
- Dominant language
- Java
- Stars
- 2.1k
- Forks
- 625
- Avg merge
- 3d 14h
- Merged PRs (30d)
- 97
Description
### Search before asking
- [x] I searched in the [issues](https://github.com/apache/fluss/issues) and found nothing similar.
### Motivation
When the aggregation merge engine is enabled, the Flink sink currently performs undo recovery unconditionally after a failover.
A failover may cause records that were already applied to Fluss to be emitted again. Undo recovery is required for non-idempotent aggregations such as `SUM`, because applying the same input twice changes the materialized result. However, it is unnecessary for idempotent aggregations such as `MAX` and `MIN`, where applying the same aggregation input again leaves the final result unchanged.
Performing undo recovery for these aggregations introduces unnecessary recovery RPCs, bucket-offset tracking, checkpoint state, and failover latency.
The connector should automatically skip undo recovery only when duplicate inputs can be proven to be safe, while conservatively retaining undo recovery for all other cases.
### Solution
Introduce an automatically selected failover recovery action for aggregation sinks:
- Use `NO_OP` when all effective target aggregations are idempotent and no effective delete or update-before record can reach the server.
- Use `UNDO` for non-idempotent aggregations, effective deletes, incomplete metadata, or any other case whose safety cannot be established.
- Consider only the columns actually written by partial updates.
- Apply the decision consistently to both SQL and DataStream sinks.
- Keep the existing undo recovery operator in the topology for checkpoint and savepoint compatibility, while avoiding recovery connections, RPCs, callbacks, and offset state in `NO_OP` mode.
- Record the selected recovery action in checkpoint state so that an irreversible `NO_OP` checkpoint cannot later be restored using `UNDO`.
The optimization should require no new user-facing configuration.
### Anything else?
The initial idempotent aggregation set includes:
- `MAX` and `MIN`
- `FIRST_VALUE` and `LAST_VALUE` variants
- `BOOL_AND` and `BOOL_OR`
- `RBM32` and `RBM64`
Aggregations such as `SUM`, `PRODUCT`, `LISTAGG`, and `STRING_AGG` continue to use undo recovery.
Legacy checkpoints without recovery-action metadata are treated as `UNDO` checkpoints for backward compatibility.
### Willingness to contribute
- [x] I'm willing to submit a PR!
Contributor guide
No contributing guide indexed for this repository
Research direction
Start at the Flink aggregation sink and trace the existing undo recovery operator, then compare the SQL and DataStream sink paths. Check how checkpoint state stores recovery information and how aggregation metadata exposes effective updates, deletes, and idempotent functions. Done means safe cases use NO_OP without recovery connections or offset state, while unsafe or incomplete cases retain UNDO and legacy checkpoints remain compatible.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- stream-processing
- Issue type
- Feature
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 48/100