apache / apache/fluss

[flink] Skip unnecessary undo recovery for idempotent aggregations

Open
#3,703 0 comments 0 reactions 0 assignees View on GitHub
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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.