You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2024/06/13 05:22:55 UTC
(pulsar) branch branch-3.0 updated (caf08c26f58 -> 912ae3c5590)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from caf08c26f58 [improve] Upgrade IPAddress to 5.5.0 (#22886)
new 77966e2a76a [improve][broker] Follow up #4196 use `PulsarByteBufAllocator` handle OOM (#20837)
new dae7d8bcf3e [fix][misc] Topic name from persistence name should decode local name (#22879)
new 912ae3c5590 [improve][broker] Optimize PersistentTopic.getLastDispatchablePosition (#22707)
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../broker/service/persistent/PersistentTopic.java | 66 ++++++++++++++++++----
.../buffer/impl/InMemTransactionBuffer.java | 14 ++++-
.../buffer/impl/TopicTransactionBuffer.java | 11 ++++
.../buffer/impl/TransactionBufferDisable.java | 14 ++++-
.../org/apache/pulsar/common/naming/TopicName.java | 5 +-
.../org/apache/pulsar/common/protocol/Markers.java | 12 ++--
.../apache/pulsar/common/naming/TopicNameTest.java | 6 ++
.../impl/FileStoreBackedReadHandleImpl.java | 4 +-
8 files changed, 105 insertions(+), 27 deletions(-)
(pulsar) 01/03: [improve][broker] Follow up #4196 use `PulsarByteBufAllocator` handle OOM (#20837)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 77966e2a76acfe1c618dd9626d277b951ea3f81b
Author: Qiang Zhao <ma...@apache.org>
AuthorDate: Thu Jul 20 23:20:44 2023 +0800
[improve][broker] Follow up #4196 use `PulsarByteBufAllocator` handle OOM (#20837)
(cherry picked from commit ac94296d2488b2b13d76fa2b1bfa71e9e3cfbb45)
---
.../main/java/org/apache/pulsar/common/protocol/Markers.java | 12 ++++++------
.../filesystem/impl/FileStoreBackedReadHandleImpl.java | 4 ++--
2 files changed, 8 insertions(+), 8 deletions(-)
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Markers.java b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Markers.java
index 50b036f99ee..2291aee781f 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Markers.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Markers.java
@@ -19,13 +19,13 @@
package org.apache.pulsar.common.protocol;
import io.netty.buffer.ByteBuf;
-import io.netty.buffer.PooledByteBufAllocator;
import io.netty.util.concurrent.FastThreadLocal;
import java.io.IOException;
import java.util.Map;
import java.util.Optional;
import lombok.SneakyThrows;
import lombok.experimental.UtilityClass;
+import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;
import org.apache.pulsar.common.api.proto.MarkerType;
import org.apache.pulsar.common.api.proto.MarkersMessageIdData;
import org.apache.pulsar.common.api.proto.MessageMetadata;
@@ -109,7 +109,7 @@ public class Markers {
.clear()
.setSnapshotId(snapshotId)
.setSourceCluster(sourceCluster);
- ByteBuf payload = PooledByteBufAllocator.DEFAULT.buffer(req.getSerializedSize());
+ ByteBuf payload = PulsarByteBufAllocator.DEFAULT.buffer(req.getSerializedSize());
try {
req.writeTo(payload);
@@ -138,7 +138,7 @@ public class Markers {
.setLedgerId(ledgerId)
.setEntryId(entryId);
- ByteBuf payload = PooledByteBufAllocator.DEFAULT.buffer(response.getSerializedSize());
+ ByteBuf payload = PulsarByteBufAllocator.DEFAULT.buffer(response.getSerializedSize());
try {
response.writeTo(payload);
return newMessage(MarkerType.REPLICATED_SUBSCRIPTION_SNAPSHOT_RESPONSE, Optional.of(replyToCluster),
@@ -172,7 +172,7 @@ public class Markers {
});
int size = snapshot.getSerializedSize();
- ByteBuf payload = PooledByteBufAllocator.DEFAULT.buffer(size);
+ ByteBuf payload = PulsarByteBufAllocator.DEFAULT.buffer(size);
try {
snapshot.writeTo(payload);
return newMessage(MarkerType.REPLICATED_SUBSCRIPTION_SNAPSHOT, Optional.of(sourceCluster), payload);
@@ -201,7 +201,7 @@ public class Markers {
.setMessageId().copyFrom(msgId);
});
- ByteBuf payload = PooledByteBufAllocator.DEFAULT.buffer(update.getSerializedSize());
+ ByteBuf payload = PulsarByteBufAllocator.DEFAULT.buffer(update.getSerializedSize());
try {
update.writeTo(payload);
@@ -258,7 +258,7 @@ public class Markers {
.setTxnidMostBits(txnMostBits)
.setTxnidLeastBits(txnLeastBits);
- ByteBuf payload = PooledByteBufAllocator.DEFAULT.buffer(0);
+ ByteBuf payload = PulsarByteBufAllocator.DEFAULT.buffer(0);
try {
return Commands.serializeMetadataAndPayload(ChecksumType.Crc32c,
diff --git a/tiered-storage/file-system/src/main/java/org/apache/bookkeeper/mledger/offload/filesystem/impl/FileStoreBackedReadHandleImpl.java b/tiered-storage/file-system/src/main/java/org/apache/bookkeeper/mledger/offload/filesystem/impl/FileStoreBackedReadHandleImpl.java
index e31c62b2bd3..91e7e902eab 100644
--- a/tiered-storage/file-system/src/main/java/org/apache/bookkeeper/mledger/offload/filesystem/impl/FileStoreBackedReadHandleImpl.java
+++ b/tiered-storage/file-system/src/main/java/org/apache/bookkeeper/mledger/offload/filesystem/impl/FileStoreBackedReadHandleImpl.java
@@ -20,7 +20,6 @@ package org.apache.bookkeeper.mledger.offload.filesystem.impl;
import static org.apache.bookkeeper.mledger.offload.OffloadUtils.parseLedgerMetadata;
import io.netty.buffer.ByteBuf;
-import io.netty.buffer.PooledByteBufAllocator;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
@@ -42,6 +41,7 @@ import org.apache.bookkeeper.mledger.ManagedLedgerException;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.MapFile;
+import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;
import org.apache.pulsar.common.naming.TopicName;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -153,7 +153,7 @@ public class FileStoreBackedReadHandleImpl implements ReadHandle {
int length = value.getLength();
long entryId = key.get();
if (entryId == nextExpectedId) {
- ByteBuf buf = PooledByteBufAllocator.DEFAULT.buffer(length, length);
+ ByteBuf buf = PulsarByteBufAllocator.DEFAULT.buffer(length, length);
entries.add(LedgerEntryImpl.create(ledgerId, entryId, length, buf));
buf.writeBytes(value.copyBytes());
entriesToRead--;
(pulsar) 03/03: [improve][broker] Optimize PersistentTopic.getLastDispatchablePosition (#22707)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 912ae3c5590b9ec143b2c6e52daf3ecc1a89d07d
Author: 道君 <da...@apache.org>
AuthorDate: Tue Jun 11 23:45:12 2024 +0800
[improve][broker] Optimize PersistentTopic.getLastDispatchablePosition (#22707)
[PersistentTopic#getLastDispatchablePosition](https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java#L3776-L3788) is using by
[Reader#hasMessageAvailable](https://github.com/apache/pulsar/blob/master/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Reader.java#L116) , [ConsumerImpl#hasMessageAvailable](https://github.com/apache/pulsar/blob/master/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java#L2440-L2448), [Consumer#getLastMessageIdAsync](https://github.com/apache/pulsar/blob/master/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Consumer.java#L591-L615).
The current implementation is read entries from Bookkeeper(or sth else), which leads to low throughput, high latency and heavy load, this PR is for the purpose of optimization.
(cherry picked from commit 266243cae246a6fa52b4b6c626932885ad44cbf4)
---
.../broker/service/persistent/PersistentTopic.java | 66 ++++++++++++++++++----
.../buffer/impl/InMemTransactionBuffer.java | 14 ++++-
.../buffer/impl/TopicTransactionBuffer.java | 11 ++++
.../buffer/impl/TransactionBufferDisable.java | 14 ++++-
4 files changed, 89 insertions(+), 16 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index 89a899a07e3..2ac140585b5 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -270,6 +270,9 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
private volatile CloseFutures closeFutures;
+ // The last position that can be dispatched to consumers
+ private volatile Position lastDispatchablePosition;
+
/***
* We use 2 futures to prevent a new closing if there is an in-progress deletion or closing. We make Pulsar return
* the in-progress one when it is called the second time.
@@ -3459,18 +3462,57 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
@Override
public CompletableFuture<Position> getLastDispatchablePosition() {
- return ManagedLedgerImplUtils.asyncGetLastValidPosition((ManagedLedgerImpl) ledger, entry -> {
- MessageMetadata md = Commands.parseMessageMetadata(entry.getDataBuffer());
- // If a messages has marker will filter by AbstractBaseDispatcher.filterEntriesForConsumer
- if (Markers.isServerOnlyMarker(md)) {
- return false;
- } else if (md.hasTxnidMostBits() && md.hasTxnidLeastBits()) {
- // Filter-out transaction aborted messages.
- TxnID txnID = new TxnID(md.getTxnidMostBits(), md.getTxnidLeastBits());
- return !isTxnAborted(txnID, (PositionImpl) entry.getPosition());
- }
- return true;
- }, getMaxReadPosition());
+ if (lastDispatchablePosition != null) {
+ return CompletableFuture.completedFuture(lastDispatchablePosition);
+ }
+ return ManagedLedgerImplUtils
+ .asyncGetLastValidPosition((ManagedLedgerImpl) ledger, entry -> {
+ MessageMetadata md = Commands.parseMessageMetadata(entry.getDataBuffer());
+ // If a messages has marker will filter by AbstractBaseDispatcher.filterEntriesForConsumer
+ if (Markers.isServerOnlyMarker(md)) {
+ return false;
+ } else if (md.hasTxnidMostBits() && md.hasTxnidLeastBits()) {
+ // Filter-out transaction aborted messages.
+ TxnID txnID = new TxnID(md.getTxnidMostBits(), md.getTxnidLeastBits());
+ return !isTxnAborted(txnID, (PositionImpl) entry.getPosition());
+ }
+ return true;
+ }, getMaxReadPosition())
+ .thenApply(position -> {
+ // Update lastDispatchablePosition to the given position
+ updateLastDispatchablePosition(position);
+ return position;
+ });
+ }
+
+ /**
+ * Update lastDispatchablePosition if the given position is greater than the lastDispatchablePosition.
+ *
+ * @param position
+ */
+ public synchronized void updateLastDispatchablePosition(Position position) {
+ // Update lastDispatchablePosition to null if the position is null, fallback to
+ // ManagedLedgerImplUtils#asyncGetLastValidPosition
+ if (position == null) {
+ lastDispatchablePosition = null;
+ return;
+ }
+
+ PositionImpl position0 = (PositionImpl) position;
+ // If the position is greater than the maxReadPosition, ignore
+ if (position0.compareTo(getMaxReadPosition()) > 0) {
+ return;
+ }
+ // If the lastDispatchablePosition is null, set it to the position
+ if (lastDispatchablePosition == null) {
+ lastDispatchablePosition = position;
+ return;
+ }
+ // If the position is greater than the lastDispatchablePosition, update it
+ PositionImpl lastDispatchablePosition0 = (PositionImpl) lastDispatchablePosition;
+ if (position0.compareTo(lastDispatchablePosition0) > 0) {
+ lastDispatchablePosition = position;
+ }
}
@Override
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBuffer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBuffer.java
index 050aeed7734..1fc5a598565 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBuffer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBuffer.java
@@ -376,8 +376,11 @@ class InMemTransactionBuffer implements TransactionBuffer {
@Override
public void syncMaxReadPositionForNormalPublish(PositionImpl position, boolean isMarkerMessage) {
- if (!isMarkerMessage && maxReadPositionCallBack != null) {
- maxReadPositionCallBack.maxReadPositionMovedForward(null, position);
+ if (!isMarkerMessage) {
+ updateLastDispatchablePosition(position);
+ if (maxReadPositionCallBack != null) {
+ maxReadPositionCallBack.maxReadPositionMovedForward(null, position);
+ }
}
}
@@ -424,4 +427,11 @@ class InMemTransactionBuffer implements TransactionBuffer {
.filter(txnBuffer -> txnBuffer.status.equals(TxnStatus.COMMITTED))
.count();
}
+
+ // ThreadSafe
+ private void updateLastDispatchablePosition(Position position) {
+ if (topic instanceof PersistentTopic t) {
+ t.updateLastDispatchablePosition(position);
+ }
+ }
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java
index 290fb3fcd18..45dea472dfe 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java
@@ -293,6 +293,11 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
}
}
+ // ThreadSafe
+ private void updateLastDispatchablePosition(Position position) {
+ topic.updateLastDispatchablePosition(position);
+ }
+
@Override
public CompletableFuture<TransactionBufferReader> openTransactionBufferReader(TxnID txnID, long startSequenceId) {
return null;
@@ -455,6 +460,8 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
} else {
updateMaxReadPosition((PositionImpl) topic.getManagedLedger().getLastConfirmedEntry(), false);
}
+ // Update the last dispatchable position to null if there is a TXN finished.
+ updateLastDispatchablePosition(null);
}
/**
@@ -519,6 +526,10 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
}
}
}
+ // If the message is a normal message, update the last dispatchable position.
+ if (!isMarkerMessage) {
+ updateLastDispatchablePosition(position);
+ }
}
@Override
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TransactionBufferDisable.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TransactionBufferDisable.java
index ff61760036e..8fa308fccbb 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TransactionBufferDisable.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TransactionBufferDisable.java
@@ -97,8 +97,11 @@ public class TransactionBufferDisable implements TransactionBuffer {
@Override
public void syncMaxReadPositionForNormalPublish(PositionImpl position, boolean isMarkerMessage) {
- if (!isMarkerMessage && maxReadPositionCallBack != null) {
- maxReadPositionCallBack.maxReadPositionMovedForward(null, position);
+ if (!isMarkerMessage) {
+ updateLastDispatchablePosition(position);
+ if (maxReadPositionCallBack != null) {
+ maxReadPositionCallBack.maxReadPositionMovedForward(null, position);
+ }
}
}
@@ -136,4 +139,11 @@ public class TransactionBufferDisable implements TransactionBuffer {
public long getCommittedTxnCount() {
return 0;
}
+
+ // ThreadSafe
+ private void updateLastDispatchablePosition(Position position) {
+ if (topic instanceof PersistentTopic t) {
+ t.updateLastDispatchablePosition(position);
+ }
+ }
}
(pulsar) 02/03: [fix][misc] Topic name from persistence name should decode local name (#22879)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit dae7d8bcf3ed701ba0c77ffb0be221757418d278
Author: 萧易客 <km...@live.com>
AuthorDate: Tue Jun 11 12:46:04 2024 +0800
[fix][misc] Topic name from persistence name should decode local name (#22879)
(cherry picked from commit c326d8e2203b6e9be37f4f2066fd7e90a9b9fb54)
---
.../src/main/java/org/apache/pulsar/common/naming/TopicName.java | 5 ++---
.../test/java/org/apache/pulsar/common/naming/TopicNameTest.java | 6 ++++++
2 files changed, 8 insertions(+), 3 deletions(-)
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java b/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java
index 69d3aaeeebb..5ef8f15b800 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java
@@ -358,17 +358,16 @@ public class TopicName implements ServiceUnitId {
String localName;
if (parts.size() == 4) {
tenant = parts.get(0);
- cluster = null;
namespacePortion = parts.get(1);
domain = parts.get(2);
- localName = parts.get(3);
+ localName = Codec.decode(parts.get(3));
return String.format("%s://%s/%s/%s", domain, tenant, namespacePortion, localName);
} else if (parts.size() == 5) {
tenant = parts.get(0);
cluster = parts.get(1);
namespacePortion = parts.get(2);
domain = parts.get(3);
- localName = parts.get(4);
+ localName = Codec.decode(parts.get(4));
return String.format("%s://%s/%s/%s/%s", domain, tenant, cluster, namespacePortion, localName);
} else {
throw new IllegalArgumentException("Invalid managedLedger name: " + mlName);
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/naming/TopicNameTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/naming/TopicNameTest.java
index 6d3ef7599d3..abd9d5b4278 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/naming/TopicNameTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/naming/TopicNameTest.java
@@ -267,6 +267,12 @@ public class TopicNameTest {
} catch (IllegalArgumentException e) {
// Exception is expected.
}
+
+ // case5: local name with special characters e.g. a:b:c
+ String topicName = "persistent://tenant/namespace/a:b:c";
+ String persistentNamingEncoding = "tenant/namespace/persistent/a%3Ab%3Ac";
+ assertEquals(TopicName.get(topicName).getPersistenceNamingEncoding(), persistentNamingEncoding);
+ assertEquals(TopicName.fromPersistenceNamingEncoding(persistentNamingEncoding), topicName);
}