[improve][broker] Make read compacted entries support maxReadSizeBytes limitation

[coderzc]

Motivation

In #7060, we allow managed cursor to specify the max read batch in bytes to avoid reading too large data at a time, but reading compacted entries did not consider maxReadSizeBytes, we should make reading compacted entries also support maxReadSizeBytes limitation.

Modifications

• Using the applyMaxSizeCap method to recalculate numberOfEntriesToRead before reading compacted entries.
• Update read stats after completion of reading compacted entries.

Verifying this change

• Make sure that the change passes the CI checks.

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

(example:)

• Added integration tests for end-to-end deployment with large payloads (10MB)
• Extended integration test for recovery after broker failure

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

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository:

[gaoran10]

LGTM

[coderzc]

org.apache.pulsar.compaction.StrategicCompactionTest.testCompactCompressedBatching always failed after applying this change. I noticed the StrategicTwoPhaseCompactor may write messages from the same batch into different entries, which will result in skipping some messages when reading compacted entries. Is that so?
Refer to:

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java

Lines 129 to 135 in d099ac4

	Entry lastEntry = entries.get(entries.size() - 1);
	// The compaction task depends on the last snapshot and the incremental
	// entries to build the new snapshot. So for the compaction cursor, we
	// need to force seek the read position to ensure the compactor can read
	// the complete last snapshot because of the compactor will read the data
	// before the compaction cursor mark delete position
	cursor.seek(lastEntry.getPosition().getNext(), true);

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java

Lines 444 to 455 in d099ac4

	<T> CompletableFuture<Boolean> addToCompactedLedger(
	LedgerHandle lh, Message<T> m, String topic, Semaphore outstanding) {
	CompletableFuture<Boolean> bkf = new CompletableFuture<>();
	if (m == null || batchMessageContainer.add((MessageImpl<?>) m, null)) {
	if (batchMessageContainer.getNumMessagesInBatch() > 0) {
	try {
	ByteBuf serialized = batchMessageContainer.toByteBuf();
	outstanding.acquire();
	mxBean.addCompactionWriteOp(topic, serialized.readableBytes());
	long start = System.nanoTime();
	lh.asyncAddEntry(serialized,
	(rc, ledger, eid, ctx) -> {

/cc @heesung-sn @Demogorgon314

[Demogorgon314]

@coderzc Yes, this is a bug. We will fix this issue.

[Demogorgon314]

@heesung-sn Should we limit the batchingMaxMessages to 200(default is 1000), so the ServiceUnitStateChannelImpl will not hit this issue? Since the StrategicTwoPhaseCompactor uses 1000 as the max num messages in batch.

https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java#L301-L307

[heesung-sn]

@heesung-sn Should we limit the batchingMaxMessages to 200(default is 1000), so the ServiceUnitStateChannelImpl will not hit this issue? Since the StrategicTwoPhaseCompactor uses 1000 as the max num messages in batch.

https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java#L301-L307

public class StrategicTwoPhaseCompactor extends TwoPhaseCompactor {
    private static final Logger log = LoggerFactory.getLogger(StrategicTwoPhaseCompactor.class);
    private static final int MAX_OUTSTANDING = 500;
    private static final int MAX_NUM_MESSAGES_IN_BATCH = 1000;
    private static final int MAX_BYTES_IN_BATCH = 128 * 1024;

Yes, if the big batching matters, then we could limit the batching in both the producer and the compactor side in BSC.

[lhotari]

There's a related really flaky test #21284 which slows down CI , @coderzc do you have a chance to check that? Thanks

[shibd]

@coderzc Can you help create a PR to cherry-pick this change to branch-2.11?