Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[improve][ml] Support Bookkeeper batch read #23180

Draft
wants to merge 12 commits into
base: master
Choose a base branch
from

Conversation

dao-jun
Copy link
Member

@dao-jun dao-jun commented Aug 15, 2024

Motivation

In BP-62, Bookkeeper starting to support batch read API,
the PR is to support Bookkeeper batch reading on the Pulsar side.

Modifications

  1. Support Bookkeeper batch read API
  2. Introduce a new flag bookkeeperEnableBatchRead to control enable BK batch read or not, the default value is false.

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: dao-jun#13

# Conflicts:
#	managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/EntryCacheDisabled.java
#	managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImpl.java
@dao-jun dao-jun changed the title [WIP] Support Bookkeeper batch read [WIP][improve][ml] Support Bookkeeper batch read Aug 15, 2024
@github-actions github-actions bot added doc-not-needed Your PR changes do not impact docs and removed doc-label-missing labels Aug 15, 2024
@apache apache deleted a comment from github-actions bot Aug 16, 2024
@dao-jun dao-jun requested review from BewareMyPower, zymap, horizonzy and hangc0276 and removed request for BewareMyPower August 21, 2024 03:33
@dao-jun dao-jun self-assigned this Aug 21, 2024
@dao-jun dao-jun added category/performance Performance issues fix or improvements area/ML labels Aug 21, 2024
@dao-jun dao-jun added this to the 4.0.0 milestone Aug 21, 2024
@dao-jun dao-jun marked this pull request as ready for review August 21, 2024 03:34
@dao-jun dao-jun changed the title [WIP][improve][ml] Support Bookkeeper batch read [improve][ml] Support Bookkeeper batch read Aug 21, 2024
@dao-jun dao-jun closed this Aug 21, 2024
@dao-jun dao-jun reopened this Aug 21, 2024
if (!useBatchRead(entriesToRead, useBookkeeperV2WireProtocol, isStriped)) {
return handle.readAsync(firstEntry, lastEntry);
}
return handle.batchReadAsync(firstEntry, entriesToRead, 0);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If maxSize = 0, info log will be printed for each call, which will affect performance.
https://github.com/apache/bookkeeper/blob/7c41204506122ed6904289f4814d4130274874aa/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java#L955-L967

private CompletableFuture<LedgerEntries> batchReadEntriesInternalAsync(long startEntry, int maxCount, long maxSize,
            boolean isRecoveryRead) {
        int nettyMaxFrameSizeBytes = clientCtx.getConf().nettyMaxFrameSizeBytes;
        if (maxSize > nettyMaxFrameSizeBytes) {
            LOG.info(
                "The max size is greater than nettyMaxFrameSizeBytes, use nettyMaxFrameSizeBytes:{} to replace it.",
                nettyMaxFrameSizeBytes);
            maxSize = nettyMaxFrameSizeBytes;
        }
        if (maxSize <= 0) {
            LOG.info("The max size is negative, use nettyMaxFrameSizeBytes:{} to replace it.", nettyMaxFrameSizeBytes);
            maxSize = nettyMaxFrameSizeBytes;
        }

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If isAutoSkipNonRecoverableData is set to true, is the batch read failure consistent with the behavior here?

} else if (cursor.getConfig().isAutoSkipNonRecoverableData()
&& exception instanceof NonRecoverableLedgerException) {
log.warn("[{}][{}] read failed from ledger at position:{} : {}", cursor.ledger.getName(), cursor.getName(),
readPosition, exception.getMessage());
final ManagedLedgerImpl ledger = (ManagedLedgerImpl) cursor.getManagedLedger();
Position nexReadPosition;
Long lostLedger = null;
if (exception instanceof ManagedLedgerException.LedgerNotExistException) {
// try to find and move to next valid ledger
nexReadPosition = cursor.getNextLedgerPosition(readPosition.getLedgerId());
lostLedger = readPosition.getLedgerId();
} else {
// Skip this read operation
nexReadPosition = ledger.getValidPositionAfterSkippedEntries(readPosition, count);
}

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If maxSize = 0, info log will be printed for each call, which will affect performance. https://github.com/apache/bookkeeper/blob/7c41204506122ed6904289f4814d4130274874aa/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java#L955-L967

private CompletableFuture<LedgerEntries> batchReadEntriesInternalAsync(long startEntry, int maxCount, long maxSize,
            boolean isRecoveryRead) {
        int nettyMaxFrameSizeBytes = clientCtx.getConf().nettyMaxFrameSizeBytes;
        if (maxSize > nettyMaxFrameSizeBytes) {
            LOG.info(
                "The max size is greater than nettyMaxFrameSizeBytes, use nettyMaxFrameSizeBytes:{} to replace it.",
                nettyMaxFrameSizeBytes);
            maxSize = nettyMaxFrameSizeBytes;
        }
        if (maxSize <= 0) {
            LOG.info("The max size is negative, use nettyMaxFrameSizeBytes:{} to replace it.", nettyMaxFrameSizeBytes);
            maxSize = nettyMaxFrameSizeBytes;
        }

apache/bookkeeper#4485 I created a PR to change the log level to debug

int entriesToRead = (int) (lastEntry - firstEntry + 1);
// Batch read is not supported for striped ledgers.
LedgerMetadata m = handle.getLedgerMetadata();
boolean isStriped = m.getEnsembleSize() != m.getWriteQuorumSize();
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It needs to be made clear in the document that enabling batch does not necessarily mean batch reading.

@dao-jun
Copy link
Member Author

dao-jun commented Aug 21, 2024

Related BK PR: apache/bookkeeper#4485
apache/bookkeeper#4487

@dao-jun dao-jun marked this pull request as draft August 22, 2024 02:30
@lhotari lhotari modified the milestones: 4.0.0, 4.1.0 Oct 11, 2024
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
area/ML category/performance Performance issues fix or improvements doc-not-needed Your PR changes do not impact docs ready-to-test release/4.0.2
Projects
None yet
Development

Successfully merging this pull request may close these issues.

3 participants