You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by pe...@apache.org on 2023/07/04 00:56:56 UTC
[pulsar] branch master updated: [improve][broker] Improve CompactedTopicImpl lock (#20697)
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new e231ee9cb99 [improve][broker] Improve CompactedTopicImpl lock (#20697)
e231ee9cb99 is described below
commit e231ee9cb998469a2999d23b7c198e8a7aa331c0
Author: Cong Zhao <zh...@apache.org>
AuthorDate: Tue Jul 4 08:56:46 2023 +0800
[improve][broker] Improve CompactedTopicImpl lock (#20697)
---
.../pulsar/compaction/CompactedTopicImpl.java | 29 ++++++++--------------
1 file changed, 10 insertions(+), 19 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java
index e2d3de9c19a..1f117338ea0 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java
@@ -60,7 +60,7 @@ public class CompactedTopicImpl implements CompactedTopic {
private final BookKeeper bk;
- private PositionImpl compactionHorizon = null;
+ private volatile PositionImpl compactionHorizon = null;
private volatile CompletableFuture<CompactedTopicContext> compactedTopicContext = null;
public CompactedTopicImpl(BookKeeper bk) {
@@ -91,7 +91,6 @@ public class CompactedTopicImpl implements CompactedTopic {
int numberOfEntriesToRead,
boolean isFirstRead,
ReadEntriesCallback callback, Consumer consumer) {
- synchronized (this) {
PositionImpl cursorPosition;
if (isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId())){
cursorPosition = PositionImpl.EARLIEST;
@@ -101,8 +100,11 @@ public class CompactedTopicImpl implements CompactedTopic {
// TODO: redeliver epoch link https://github.com/apache/pulsar/issues/13690
ReadEntriesCtx readEntriesCtx = ReadEntriesCtx.create(consumer, DEFAULT_CONSUMER_EPOCH);
- if (compactionHorizon == null
- || compactionHorizon.compareTo(cursorPosition) < 0) {
+
+ final PositionImpl currentCompactionHorizon = compactionHorizon;
+
+ if (currentCompactionHorizon == null
+ || currentCompactionHorizon.compareTo(cursorPosition) < 0) {
cursor.asyncReadEntriesOrWait(numberOfEntriesToRead, callback, readEntriesCtx, PositionImpl.LATEST);
} else {
compactedTopicContext.thenCompose(
@@ -110,23 +112,13 @@ public class CompactedTopicImpl implements CompactedTopic {
.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) {
- cursor.seek(compactionHorizon.getNext());
+ if (startPoint == COMPACT_LEDGER_EMPTY || startPoint == NEWER_THAN_COMPACTED) {
+ cursor.seek(currentCompactionHorizon.getNext());
callback.readEntriesComplete(Collections.emptyList(), readEntriesCtx);
return CompletableFuture.completedFuture(null);
- }
- if (startPoint == NEWER_THAN_COMPACTED && compactionHorizon.compareTo(cursorPosition) < 0) {
- cursor.asyncReadEntriesOrWait(numberOfEntriesToRead, callback, readEntriesCtx,
- PositionImpl.LATEST);
- return CompletableFuture.completedFuture(null);
} else {
long endPoint = Math.min(context.ledger.getLastAddConfirmed(),
startPoint + numberOfEntriesToRead);
- if (startPoint == NEWER_THAN_COMPACTED) {
- cursor.seek(compactionHorizon.getNext());
- callback.readEntriesComplete(Collections.emptyList(), readEntriesCtx);
- return CompletableFuture.completedFuture(null);
- }
return readEntries(context.ledger, startPoint, endPoint)
.thenAccept((entries) -> {
Entry lastEntry = entries.get(entries.size() - 1);
@@ -142,7 +134,7 @@ public class CompactedTopicImpl implements CompactedTopic {
}))
.exceptionally((exception) -> {
if (exception.getCause() instanceof NoSuchElementException) {
- cursor.seek(compactionHorizon.getNext());
+ cursor.seek(currentCompactionHorizon.getNext());
callback.readEntriesComplete(Collections.emptyList(), readEntriesCtx);
} else {
callback.readEntriesFailed(new ManagedLedgerException(exception), readEntriesCtx);
@@ -150,7 +142,6 @@ public class CompactedTopicImpl implements CompactedTopic {
return null;
});
}
- }
}
static CompletableFuture<Long> findStartPoint(PositionImpl p,
@@ -318,7 +309,7 @@ public class CompactedTopicImpl implements CompactedTopic {
.compare(p.getEntryId(), m.getEntryId()).result();
}
- public synchronized Optional<Position> getCompactionHorizon() {
+ public Optional<Position> getCompactionHorizon() {
return Optional.ofNullable(this.compactionHorizon);
}