[test] Cover aggregation recovery across partitions with different bucket counts #4280

Search before asking

• I searched the existing issues and found no duplicate covering this test scope.

Description

Add integration coverage for aggregation merge engine recovery when partitions of the same table have different bucket counts. This follows the per-partition bucket-count work in PR #3908.

The PR already tests mixed-count bucket enumeration and producer-offset registration in RecoveryOffsetManagerTest. These use a test Admin and parallelism 1. The existing UndoRecoveryManagerITCase.testPartitionedTableRecovery exercises real undo, but both partitions have the same bucket count. The rescale tests in UndoRecoveryITCase change Flink parallelism while keeping the table bucket count fixed.

We need to exercise the full recovery path: discover actual partition buckets, obtain recovery offsets, read changelog, apply undo writes, and resume aggregation without missing or double-applying records.

Shared setup

• Create a partitioned primary-key table, e.g. PRIMARY KEY (pt, id), with table.merge-engine = aggregation and an explicit SUM aggregation.
• Create and populate an old partition, alter bucket.num, then create a new partition. Cover both 2 -> 4 and 4 -> 2, and assert that each partition retains its own bucket count.
• Use deterministic keys that route differently under the two layouts. Ensure data reaches the higher-numbered buckets of the larger partition so truncated bucket enumeration cannot pass unnoticed.
• Include multi-subtask runs, such as parallelism 3 and 4, to exercise normal-write sharding and recovery assignment. Parallelism 1 alone cannot expose assignment mismatches.

1. Real checkpoint recovery

1. Run a real Flink aggregation sink job against the mixed-count partitions and wait for a completed checkpoint containing a known baseline.
2. Write additional records to both partitions and wait until those writes are visible in Fluss. Prevent another checkpoint from completing, then inject a job failure.
3. Restore from the completed checkpoint and replay the post-checkpoint input.
4. Verify that recovery actually performed undo for both layouts and that every key has the exact expected aggregate. For example, baseline 10 followed by an increment of 7 must finish at 17, not 24.
5. Check that checkpoint/recovery offsets cover the correct physical buckets, with no nonexistent buckets, omitted high-numbered buckets, or conflicting offsets for a bucket across restored state fragments.

2. Producer-offset recovery without a checkpoint, including another failure during undo

1. Start with no Flink checkpoint/savepoint to restore and a stable sink producer ID. Establish the producer-offset baseline before writing.
2. Write records to both partitions and fail before the first checkpoint completes.
3. Restart with the same producer ID and verify that recovery uses the persisted producer offsets and performs real changelog reads and undo writes.
4. After some undo writes have been acknowledged, but before recovery finishes and normal input processing resumes, inject another failure.
5. Restart again with the same producer ID and no completed checkpoint. Verify that the original recovery baseline remains usable, partially completed undo is safe to repeat, and recovery converges.
6. Resume/replay input and assert exact per-key aggregates against a failure-free reference result: no missing records, duplicate aggregation, or residual effects from the failed attempts. Verify producer-offset cleanup follows the existing lifecycle after successful recovery.

Acceptance criteria

• Both scenarios run through the real Flink sink recovery path, including actual Fluss log scanning and undo writes; offset-map assertions alone are insufficient.
• Both increasing and decreasing bucket counts are covered, with multiple subtasks and data in every relevant bucket.
• Assertions prove that undo occurred and that final per-key aggregates are correct for both partitions.
• The second failure is injected after observable undo progress, making repeated-recovery idempotency part of the test.
• Failure injection and checkpoint boundaries use explicit synchronization/progress signals rather than timing-dependent sleeps.

Extend UndoRecoveryITCase for the job-level scenarios and add focused helpers or coverage in UndoRecoveryManagerITCase where needed.

Willingness to contribute

• I'm willing to submit a PR!