[Umbrella] Support Aggregation Merge Engine

### Search before asking

- [x] I searched in the [issues](https://github.com/apache/fluss/issues) and found nothing similar.

### Motivation

When using Aggregation Merge Engine with multiple concurrent Flink jobs writing to the same primary key table, the following issues occur:

1. **Concurrent Conflicts**: Multiple writers updating the same aggregation columns simultaneously cause data inconsistency
2. **No Exactly-Once Guarantee**: Job failover cannot guarantee aggregation accuracy
3. **Duplicate Calculations**: Data replay after job restart leads to duplicate aggregation
4. **State Loss**: Lack of state management to track committed data

These issues make Aggregation Merge Engine unsuitable for production scenarios requiring strict data accuracy.

### Solution

The complete solution has been proposed in the FIP document [FIP-21: Aggregation Merge Engine](https://cwiki.apache.org/confluence/display/FLUSS/FIP-21%3A+Aggregation+Merge+Engine).

### 1. State Management
- **WriterState**: Tracks maximum committed offset per TableBucket
- **BucketOffsetTracker**: Client-side offset tracking
- **Integration**: Uses Flink's Operator State for persistence across checkpoints

### 2. Undo Recovery
- **Mechanism**: On failover, undoes uncommitted data written after last checkpoint
- **Implementation**: Reads old values at committed offset and overwrites new values
- **Guarantee**: Ensures data consistency after job restart

### Anything else?

_No response_

### Willingness to contribute

- [x] I'm willing to submit a PR!

No files, tests, or code entry points are named. Start by reading FIP-21 and tracing the Aggregation Merge Engine and Flink integration; done means implementing WriterState, BucketOffsetTracker, checkpoint persistence, and undo recovery for consistent failover behavior.

Written by the indexing model from the issue text.