[FLINK-35661][table] Fix MiniBatchGroupAggFunction silently dropping records (backport #27505)

[nateab]

Backport of #27505 to release-1.20.

What is the purpose of the change

This pull request fixes a bug in MiniBatchGroupAggFunction.finishBundle() where records were being silently dropped when a mini-batch bundle contained a key with only retraction messages and no existing accumulator state.

The root cause was using return instead of continue when inputRows became empty after filtering out leading retraction messages for a key with no state. This caused the method to exit entirely, abandoning processing of all remaining keys in the bundle.

Brief change log

• Changed return; to continue; in MiniBatchGroupAggFunction.finishBundle() so processing continues to the next key instead of exiting the entire method
• Added unit test MiniBatchGroupAggFunctionTest that directly verifies the fix by simulating a bundle with multiple keys where the first key has only retractions

Verifying this change

This change added tests and can be verified as follows:

• Added MiniBatchGroupAggFunctionTest.testFinishBundleContinuesAfterEmptyInputRows() which creates a mock bundle with three keys where the first key has only a DELETE message (no existing state). The test verifies that:
  • Without the fix: 0 outputs are produced (all subsequent keys dropped)
  • With the fix: 2 outputs are produced (keys "bbb" and "ccc" processed correctly)

Backport notes:

• Cherry-picked cleanly from b81b124 (master). The source file is byte-identical between master pre-fix and release-1.20.
• Test passes locally on release-1.20: Tests run: 1, Failures: 0, Errors: 0.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): no
• The serializers: no
• The runtime per-record code paths (performance sensitive): yes (mini-batch aggregation path, but change is minimal - only control flow fix)
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? no
• If yes, how is the feature documented? not applicable

[flinkbot]

CI report:

• b49ec0c Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[nateab]

@flinkbot run azure

Previous failure was in MinioTestContainerTest (flink-s3-fs-base) due to a testcontainers/docker Date deserialization issue (Can not parse date "2022-09-25T13:55:20.239867619+0800") — unrelated to this change. The branch is on the current tip of release-1.20 (dd21933a4ce).

[nateab]

@flinkbot run azure

[nateab]

Force-pushed identical content with new timestamp to trigger a fresh Azure build (@flinkbot run azure was being deduped against the original SHA eb3440cc2e9).

New SHA: b49ec0cb5aa

The previous test_ci connect failure was the known Minio testcontainer infrastructure issue that's currently affecting ALL recent release-1.20 PRs (e.g. #28204 which was merged with the same failure, #28258, #28271). The table-runtime change itself is verified locally on 1.20: Tests run: 1, Failures: 0, Errors: 0.

[weiqingy]

Thanks for the backport, and for the provenance notes — the cherry-pick SHA and the byte-identical claim made this quick to verify.

I checked the backport against master:

• The return→continue fix in finishBundle() is byte-identical to the merged master commit.
• MiniBatchGroupAggFunctionTest.java is byte-identical to the test merged in #27505.
• The Azure failure is the MinioTestContainerTest flake in flink-s3-fs-base, unrelated to the table-runtime change and consistent with the other release-1.20 PRs you referenced.

Nothing further from me — the backport looks faithful. I'll leave the final call to the maintainers.