Repository: apache/flink
File: flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/aggregate/MiniBatchGroupAggFunctionTest.java (Java)

Code before:
        GenericRowData keyB = GenericRowData.of(StringData.fromString("bbb"));
        List<RowData> rowsB = new ArrayList<>();
        GenericRowData insertB = GenericRowData.of(StringData.fromString("bbb"), 2L);
        insertB.setRowKind(RowKind.INSERT);
        rowsB.add(insertB);
        buffer.put(keyB, rowsB);

        // Key "ccc" - normal insert
        GenericRowData keyC = GenericRowData.of(StringData.fromString("ccc"));
        List<RowData> rowsC = new ArrayList<>();
        GenericRowData insertC = GenericRowData.of(StringData.fromString("ccc"), 3L);
        insertC.setRowKind(RowKind.INSERT);
        rowsC.add(insertC);
        buffer.put(keyC, rowsC);

        // Collect output
        List<RowData> output = new ArrayList<>();
        TestCollector collector = new TestCollector(output);

        // Create a test instance with mocks
        TestMiniBatchGroupAggFunction function = new TestMiniBatchGroupAggFunction();
        function.finishBundle(buffer, collector);

        // Verify that keys "bbb" and "ccc" were processed (2 outputs)
        // Before the fix, only key "aaa" would be processed (and filtered out),
        // then the method would return early, producing 0 outputs.
        assertThat(output).hasSize(2);

        // Verify the output contains results for both "bbb" and "ccc"
        assertThat(output.stream().anyMatch(row -> row.getString(0).toString().equals("bbb")))
                .isTrue();
        assertThat(output.stream().anyMatch(row -> row.getString(0).toString().equals("ccc")))
                .isTrue();
    }

    /** Test collector that stores output rows. */
    private static class TestCollector implements Collector<RowData> {
        private final List<RowData> output;

        TestCollector(List<RowData> output) {
            this.output = output;
        }

        @Override
        public void collect(RowData record) {
            // Make a copy since the record might be reused
            GenericRowData copy = GenericRowData.of(record.getString(0), record.getLong(1));
            copy.setRowKind(record.getRowKind());
            output.add(copy);
        }


Review comment (bug):
nit: drop this comment. First sentence is obvious from the code and the assertions below. The next two lines are `claude` explanation for the generated test code, not relevant once the bug is fixed. 

Code after:
    private final int keyIdx = 0;
    private final RowDataKeySelector keySelector =
            HandwrittenSelectorUtil.getRowDataSelector(new int[] {keyIdx}, inputFieldTypes);

    private final RowDataHarnessAssertor assertor =
            new RowDataHarnessAssertor(
                    outputTypes, new GenericRowRecordSortComparator(keyIdx, outputTypes[keyIdx]));

    private final GeneratedAggsHandleFunction genAggsHandler =
            new GeneratedAggsHandleFunction("SumAgg", "", new Object[0]) {
                @Override
                public SumAggsHandleFunction newInstance(ClassLoader classLoader) {
                    return new SumAggsHandleFunction(1); // inputIndex = 1 (the value field)
                }
            };

    private final GeneratedRecordEqualiser genRecordEqualiser =
            new GeneratedRecordEqualiser("Equaliser", "", new Object[0]) {
                @Override
                public RecordEqualiser newInstance(ClassLoader classLoader) {
                    return (row1, row2) -> {
                        if (row1 instanceof GenericRowData && row2 instanceof GenericRowData) {
                            return row1.equals(row2);
                        }
                        return false;
                    };
                }
            };

    private MiniBatchGroupAggFunction createFunction(boolean generateUpdateBefore) {
        return new MiniBatchGroupAggFunction(
                genAggsHandler,
                genRecordEqualiser,
                accTypes,
                inputRowType,
                -1, // no COUNT(*) for this test
                generateUpdateBefore,
                0); // no state retention
    }

    @SuppressWarnings({"unchecked", "rawtypes"})
    private OneInputStreamOperatorTestHarness<RowData, RowData> createTestHarness(
            MiniBatchGroupAggFunction function) throws Exception {
        CountBundleTrigger trigger = new CountBundleTrigger<>(10);
        KeyedMapBundleOperator operator = new KeyedMapBundleOperator(function, trigger);
        return new KeyedOneInputStreamOperatorTestHarness<>(
                operator, keySelector, keySelector.getProducedType());
    }

    /**
     * Verifies that when finishBundle processes a key with only retraction messages (which gets