|
public CompletableFuture<CompactedTopicContext> newCompactedLedger(Position p, long compactedLedgerId) { |
|
synchronized (this) { |
|
CompletableFuture<CompactedTopicContext> previousContext = compactedTopicContext; |
|
compactedTopicContext = openCompactedLedger(bk, compactedLedgerId); |
|
|
|
compactionHorizon = (PositionImpl) p; |
|
|
|
// delete the ledger from the old context once the new one is open |
|
return compactedTopicContext.thenCompose( |
|
__ -> previousContext != null ? previousContext : CompletableFuture.completedFuture(null)); |
|
} |
|
} |
|
|
|
@Override |
|
public CompletableFuture<Void> deleteCompactedLedger(long compactedLedgerId) { |
|
return tryDeleteCompactedLedger(bk, compactedLedgerId); |
|
} |
|
|
|
@Override |
|
@Deprecated |
|
public void asyncReadEntriesOrWait(ManagedCursor cursor, |
|
int maxEntries, |
|
long bytesToRead, |
|
PositionImpl maxReadPosition, |
|
boolean isFirstRead, |
|
ReadEntriesCallback callback, Consumer consumer) { |
|
PositionImpl cursorPosition; |
|
if (isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId())){ |
|
cursorPosition = PositionImpl.EARLIEST; |
|
} else { |
|
cursorPosition = (PositionImpl) cursor.getReadPosition(); |
|
} |
|
|
|
// TODO: redeliver epoch link https://github.com/apache/pulsar/issues/13690 |
|
ReadEntriesCtx readEntriesCtx = ReadEntriesCtx.create(consumer, DEFAULT_CONSUMER_EPOCH); |
|
|
|
final PositionImpl currentCompactionHorizon = compactionHorizon; |
|
|
|
if (currentCompactionHorizon == null |
|
|| currentCompactionHorizon.compareTo(cursorPosition) < 0) { |
|
cursor.asyncReadEntriesOrWait(maxEntries, bytesToRead, callback, readEntriesCtx, maxReadPosition); |
|
} else { |
|
ManagedCursorImpl managedCursor = (ManagedCursorImpl) cursor; |
|
int numberOfEntriesToRead = managedCursor.applyMaxSizeCap(maxEntries, bytesToRead); |
|
|
|
compactedTopicContext.thenCompose( |
|
(context) -> findStartPoint(cursorPosition, context.ledger.getLastAddConfirmed(), context.cache) |
|
.thenCompose((startPoint) -> { |
|
// do not need to read the compaction ledger if it is empty. |
|
// the cursor just needs to be set to the compaction horizon |
|
if (startPoint == COMPACT_LEDGER_EMPTY || startPoint == NEWER_THAN_COMPACTED) { |
|
cursor.seek(currentCompactionHorizon.getNext()); |
|
callback.readEntriesComplete(Collections.emptyList(), readEntriesCtx); |
|
return CompletableFuture.completedFuture(null); |
|
} else { |
|
long endPoint = Math.min(context.ledger.getLastAddConfirmed(), |
|
startPoint + (numberOfEntriesToRead - 1)); |
|
return readEntries(context.ledger, startPoint, endPoint) |
|
.thenAccept((entries) -> { |
|
long entriesSize = 0; |
|
for (Entry entry : entries) { |
|
entriesSize += entry.getLength(); |
|
} |
|
managedCursor.updateReadStats(entries.size(), entriesSize); |
|
|
|
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); |
|
callback.readEntriesComplete(entries, readEntriesCtx); |
|
}); |
|
} |
|
})) |
|
.exceptionally((exception) -> { |
|
if (exception.getCause() instanceof NoSuchElementException) { |
|
cursor.seek(currentCompactionHorizon.getNext()); |
|
callback.readEntriesComplete(Collections.emptyList(), readEntriesCtx); |
|
} else { |
|
callback.readEntriesFailed(new ManagedLedgerException(exception), readEntriesCtx); |
|
} |
|
return null; |
|
}); |
|
} |
|
} |
|
|
|
static CompletableFuture<Long> findStartPoint(PositionImpl p, |
|
long lastEntryId, |
|
AsyncLoadingCache<Long, MessageIdData> cache) { |
|
CompletableFuture<Long> promise = new CompletableFuture<>(); |
|
// if lastEntryId is less than zero it means there are no entries in the compact ledger |
|
if (lastEntryId < 0) { |
|
promise.complete(COMPACT_LEDGER_EMPTY); |
|
} else { |
|
findStartPointLoop(p, 0, lastEntryId, promise, cache); |
|
} |
|
return promise; |
|
} |
|
|
|
@VisibleForTesting |
|
static void findStartPointLoop(PositionImpl p, long start, long end, |
|
CompletableFuture<Long> promise, |
|
AsyncLoadingCache<Long, MessageIdData> cache) { |
|
long midpoint = start + ((end - start) / 2); |
|
|
|
CompletableFuture<MessageIdData> startEntry = cache.get(start); |
|
CompletableFuture<MessageIdData> middleEntry = cache.get(midpoint); |
|
CompletableFuture<MessageIdData> endEntry = cache.get(end); |
|
|
|
CompletableFuture.allOf(startEntry, middleEntry, endEntry).thenRun( |
|
() -> { |
|
if (comparePositionAndMessageId(p, startEntry.join()) <= 0) { |
|
promise.complete(start); |
|
} else if (comparePositionAndMessageId(p, middleEntry.join()) <= 0) { |
|
findStartPointLoop(p, start + 1, midpoint, promise, cache); |
|
} else if (comparePositionAndMessageId(p, endEntry.join()) <= 0) { |
|
findStartPointLoop(p, midpoint + 1, end, promise, cache); |
|
} else { |
|
promise.complete(NEWER_THAN_COMPACTED); |
|
} |
|
}).exceptionally((exception) -> { |
|
promise.completeExceptionally(exception); |
|
return null; |
|
}); |
|
} |
|
|
|
static AsyncLoadingCache<Long, MessageIdData> createCache(LedgerHandle lh, |
|
long maxSize) { |
|
return Caffeine.newBuilder() |
|
.maximumSize(maxSize) |
|
.buildAsync((entryId, executor) -> readOneMessageId(lh, entryId)); |
|
} |
Search before asking
Version
branch-2.9 (guess master branch also have this problem)
Minimal reproduce step
reproduce step:
The error log is as following:
We found that the reason is reading compacted ledger 1705537 failed in the process of TwoPhaseCompaction.
When do TwoPhaseCompaction. The step is :
In the phaseOneLoop(), it would try to read message, trigger CompactedTopicImpl#asyncReadEntriesOrWait(). And then it would use compactedTopicContext to findStartPoint. compactedTopicContext is a local variable, containing ReadOnlyLedgerHandle and AsyncLoadingCache. compactedTopicContext is updated in the previous compaction, and then it would not change until the next compaction finish.
Previous compaction make compactedTopicContext, 1705537 LedgerHandle's metadata is [bookie-1, bookie-2, bookie-3]. However, after bookie shutdown and do bookie autoRecovery, the 3 quorum on 1705537 has been changed to [bookie-4, bookie-5, bookie-6]. But the LedgerHandle's metadata in broker is not change. So it still try to read message of 1705537 from bookie-1,2,3 and fail, causing the compaction process can not success.
The relevant code is:
pulsar/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java
Lines 76 to 209 in 51202a6
pulsar/pulsar-broker/src/main/java/org/apache/pulsar/compaction/TwoPhaseCompactor.java
Lines 74 to 199 in 51202a6
We temporarily fix the problem by restart all the broker. Because after broker restart, the previous compactedTopicContext is not exist, compaction can succeed.
But this is not a user-friendly way since each time bookie shutdown, we may need to restart broker. We‘d better fix this issue.
After diving to the code, we found that there are two way to open ReadOnlyLedgerHandle in bookie, asyncOpenLedgerNoRecovery and asyncOpenLedger.
We can replace asyncOpenLedger to asyncOpenLedgerNoRecovery. I guess the compactedLedger would not in open state, and it need to update metadata.
What did you expect to see?
compaction can succeed after compactedLedger 3 quorum is autoRecover
What did you see instead?
compaction keep failed after compactedLedger 3 quorum is autoRecover
Anything else?
No response
Are you willing to submit a PR?