KAFKA-20452: Avoid creating unnecessary empty batches in LogCleaner below the High Watermark

[JiayaoS]

This patch replaces upperBoundOffset with highWatermark in
Cleaner.cleanSegments() / cleanInto(), update related Javadoc and
comments, and adds a test to verify that empty batches are not retained
when the highWatermark is beyond the cleaned range.

Reviewers: Jun Rao junrao@gmail.com, PoAn Yang payang@apache.org,
Chia-Ping Tsai chia7712@gmail.com

[JiayaoS]

PR #17193 (KAFKA-17076) introduced logic to retain the last batch in each cleaning round even if empty, to preserve logEndOffset after compaction. As noted by @junrao, the current check uses upperBoundOffset, which may produce unnecessary empty batch.

We only need to retain an empty batch when its nextOffset equals the highWatermark. This PR replaces upperBoundOffset with highWatermark in Cleaner.java, update related comments and Javadoc, and adds testCleanSegmentsNotRetainingLastEmptyBatch to verify empty batch are not retained when the highWatermark is beyond the cleaned range.

testCleanSegmentsRetainingLastEmptyBatch — empty batch retained when nextOffset == highWatermark.
testCleanSegmentsNotRetainingLastEmptyBatch — no empty batch retained when high watermark is beyond the cleaned segments.

[chia7712]

Could we get the hw directly within this method instead of passing it in from the caller?

[JiayaoS]

If we use log.highWatermark() directly, I think the main impact would be on unit tests. If the HW is modified during cleaning, different batches may see different HW values, but I don't think this would cause any issues since the HW typically only moves forward.

[chia7712]

yes, since hw is monotonic, we can simplify the method by fetching it internally rather than capturing the value too early.

[JiayaoS]

I adjusted how the highWatermark parameter is passed, updated the corresponding unit tests, and all tests pass locally.

[JiayaoS]

Hi @junrao , Could you please take a look when you have a moment? Thanks!

[junrao]

@JiayaoS : Thanks for the PR. LGTM