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 2025/02/25 11:09:19 UTC
(pulsar) branch branch-4.0 updated (56233eae41e -> 657d5ee17e5)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 56233eae41e [fix][broker] PIP-322 Fix issue with rate limiters where rates can exceed limits initially and consumption pauses until token balance is positive (#24012)
new c8d8b6f6392 [fix][test]Fix flaky test V1_ProducerConsumerTest.testConcurrentConsumerReceiveWhileReconnect (#24019)
new 50997afa08d [fix][broker] Fix BucketDelayedDeliveryTracker thread safety (#24014)
new 657d5ee17e5 [improve][cli] Support additional msg metadata for V1 topic on peek message cmd (#23978)
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:
.../bucket/BucketDelayedDeliveryTracker.java | 2 +-
.../client/api/v1/V1_ProducerConsumerTest.java | 8 +-
.../pulsar/admin/cli/CmdPersistentTopics.java | 24 +-----
.../org/apache/pulsar/admin/cli/CmdTopics.java | 94 ++++++++++++----------
4 files changed, 58 insertions(+), 70 deletions(-)
(pulsar) 03/03: [improve][cli] Support additional msg metadata for V1 topic on peek message cmd (#23978)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 657d5ee17e59e2cd72d1f4dce703e35b3fe2422a
Author: Rajan Dhabalia <rd...@apache.org>
AuthorDate: Mon Feb 24 22:52:08 2025 -0800
[improve][cli] Support additional msg metadata for V1 topic on peek message cmd (#23978)
(cherry picked from commit 626b211f91fd8d1e9821ae8e2b9b29520e72ac63)
---
.../pulsar/admin/cli/CmdPersistentTopics.java | 24 +-----
.../org/apache/pulsar/admin/cli/CmdTopics.java | 94 ++++++++++++----------
2 files changed, 52 insertions(+), 66 deletions(-)
diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdPersistentTopics.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdPersistentTopics.java
index 3dc0ba7b6f2..7b86e2af7f5 100644
--- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdPersistentTopics.java
+++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdPersistentTopics.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.admin.cli;
import static org.apache.commons.lang3.StringUtils.isNotBlank;
+import static org.apache.pulsar.admin.cli.CmdTopics.printMessages;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import io.netty.buffer.ByteBuf;
@@ -37,8 +38,6 @@ import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.admin.Topics;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
-import org.apache.pulsar.client.impl.BatchMessageIdImpl;
-import org.apache.pulsar.client.impl.MessageIdImpl;
import picocli.CommandLine.Command;
import picocli.CommandLine.Option;
import picocli.CommandLine.Parameters;
@@ -589,26 +588,7 @@ public class CmdPersistentTopics extends CmdBase {
void run() throws PulsarAdminException {
String persistentTopic = validatePersistentTopic(topicName);
List<Message<byte[]>> messages = getPersistentTopics().peekMessages(persistentTopic, subName, numMessages);
- int position = 0;
- for (Message<byte[]> msg : messages) {
- if (++position != 1) {
- System.out.println("-------------------------------------------------------------------------\n");
- }
- if (msg.getMessageId() instanceof BatchMessageIdImpl) {
- BatchMessageIdImpl msgId = (BatchMessageIdImpl) msg.getMessageId();
- System.out.println("Batch Message ID: " + msgId.getLedgerId() + ":" + msgId.getEntryId() + ":"
- + msgId.getBatchIndex());
- } else {
- MessageIdImpl msgId = (MessageIdImpl) msg.getMessageId();
- System.out.println("Message ID: " + msgId.getLedgerId() + ":" + msgId.getEntryId());
- }
- if (msg.getProperties().size() > 0) {
- System.out.println("Properties:");
- print(msg.getProperties());
- }
- ByteBuf data = Unpooled.wrappedBuffer(msg.getData());
- System.out.println(ByteBufUtil.prettyHexDump(data));
- }
+ printMessages(messages, false, this);
}
}
diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java
index 22073b1a89d..ca15e111390 100644
--- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java
+++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java
@@ -1117,50 +1117,7 @@ public class CmdTopics extends CmdBase {
String persistentTopic = validatePersistentTopic(topicName);
List<Message<byte[]>> messages = getTopics().peekMessages(persistentTopic, subName, numMessages,
showServerMarker, transactionIsolationLevel);
- int position = 0;
- for (Message<byte[]> msg : messages) {
- MessageImpl message = (MessageImpl) msg;
- if (++position != 1) {
- System.out.println("-------------------------------------------------------------------------\n");
- }
- if (message.getMessageId() instanceof BatchMessageIdImpl) {
- BatchMessageIdImpl msgId = (BatchMessageIdImpl) message.getMessageId();
- System.out.println("Batch Message ID: " + msgId.getLedgerId() + ":" + msgId.getEntryId() + ":"
- + msgId.getBatchIndex());
- } else {
- MessageIdImpl msgId = (MessageIdImpl) msg.getMessageId();
- System.out.println("Message ID: " + msgId.getLedgerId() + ":" + msgId.getEntryId());
- }
-
- System.out.println("Publish time: " + message.getPublishTime());
- System.out.println("Event time: " + message.getEventTime());
-
- if (message.getDeliverAtTime() != 0) {
- System.out.println("Deliver at time: " + message.getDeliverAtTime());
- }
- MessageMetadata msgMetaData = message.getMessageBuilder();
- if (showServerMarker && msgMetaData.hasMarkerType()) {
- System.out.println("Marker Type: " + MarkerType.valueOf(msgMetaData.getMarkerType()));
- }
-
- if (message.getBrokerEntryMetadata() != null) {
- if (message.getBrokerEntryMetadata().hasBrokerTimestamp()) {
- System.out.println("Broker entry metadata timestamp: "
- + message.getBrokerEntryMetadata().getBrokerTimestamp());
- }
- if (message.getBrokerEntryMetadata().hasIndex()) {
- System.out.println("Broker entry metadata index: "
- + message.getBrokerEntryMetadata().getIndex());
- }
- }
-
- if (message.getProperties().size() > 0) {
- System.out.println("Properties:");
- print(msg.getProperties());
- }
- ByteBuf data = Unpooled.wrappedBuffer(msg.getData());
- System.out.println(ByteBufUtil.prettyHexDump(data));
- }
+ printMessages(messages, showServerMarker, this);
}
}
@@ -1379,6 +1336,55 @@ public class CmdTopics extends CmdBase {
return null;
}
+ public static void printMessages(List<Message<byte[]>> messages, boolean showServerMarker, CliCommand cli) {
+ if (messages == null) {
+ return;
+ }
+ int position = 0;
+ for (Message<byte[]> msg : messages) {
+ MessageImpl message = (MessageImpl) msg;
+ if (++position != 1) {
+ System.out.println("-------------------------------------------------------------------------\n");
+ }
+ if (message.getMessageId() instanceof BatchMessageIdImpl) {
+ BatchMessageIdImpl msgId = (BatchMessageIdImpl) message.getMessageId();
+ System.out.println("Batch Message ID: " + msgId.getLedgerId() + ":" + msgId.getEntryId() + ":"
+ + msgId.getBatchIndex());
+ } else {
+ MessageIdImpl msgId = (MessageIdImpl) msg.getMessageId();
+ System.out.println("Message ID: " + msgId.getLedgerId() + ":" + msgId.getEntryId());
+ }
+
+ System.out.println("Publish time: " + message.getPublishTime());
+ System.out.println("Event time: " + message.getEventTime());
+
+ if (message.getDeliverAtTime() != 0) {
+ System.out.println("Deliver at time: " + message.getDeliverAtTime());
+ }
+ MessageMetadata msgMetaData = message.getMessageBuilder();
+ if (showServerMarker && msgMetaData.hasMarkerType()) {
+ System.out.println("Marker Type: " + MarkerType.valueOf(msgMetaData.getMarkerType()));
+ }
+
+ if (message.getBrokerEntryMetadata() != null) {
+ if (message.getBrokerEntryMetadata().hasBrokerTimestamp()) {
+ System.out.println("Broker entry metadata timestamp: "
+ + message.getBrokerEntryMetadata().getBrokerTimestamp());
+ }
+ if (message.getBrokerEntryMetadata().hasIndex()) {
+ System.out.println("Broker entry metadata index: " + message.getBrokerEntryMetadata().getIndex());
+ }
+ }
+
+ if (message.getProperties().size() > 0) {
+ System.out.println("Properties:");
+ cli.print(msg.getProperties());
+ }
+ ByteBuf data = Unpooled.wrappedBuffer(msg.getData());
+ System.out.println(ByteBufUtil.prettyHexDump(data));
+ }
+ }
+
@Command(description = "Trigger offload of data from a topic to long-term storage (e.g. Amazon S3)")
private class Offload extends CliCommand {
@Option(names = { "-s", "--size-threshold" },
(pulsar) 01/03: [fix][test]Fix flaky test V1_ProducerConsumerTest.testConcurrentConsumerReceiveWhileReconnect (#24019)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit c8d8b6f63921b26b8eff465acc100d3fe75aa534
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Tue Feb 25 14:48:55 2025 +0800
[fix][test]Fix flaky test V1_ProducerConsumerTest.testConcurrentConsumerReceiveWhileReconnect (#24019)
(cherry picked from commit 2c686d0ab1ea3f697dea4f5ce7938282b9b06f88)
---
.../org/apache/pulsar/client/api/v1/V1_ProducerConsumerTest.java | 8 +++++---
1 file changed, 5 insertions(+), 3 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v1/V1_ProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v1/V1_ProducerConsumerTest.java
index 0b3ff345acf..255591ea7da 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v1/V1_ProducerConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v1/V1_ProducerConsumerTest.java
@@ -589,9 +589,11 @@ public class V1_ProducerConsumerTest extends V1_ProducerConsumerBase {
restartBroker();
// The available permits should be 10 and num messages in the queue should be 90
- Awaitility.await().untilAsserted(() ->
- Assert.assertEquals(consumerImpl.getAvailablePermits(), numConsumersThreads));
- Assert.assertEquals(consumerImpl.numMessagesInQueue(), recvQueueSize - numConsumersThreads);
+ Awaitility.await().untilAsserted(() -> {
+ Assert.assertEquals(consumerImpl.getAvailablePermits(), numConsumersThreads);
+ Assert.assertEquals(consumerImpl.numMessagesInQueue(), recvQueueSize - numConsumersThreads);
+ });
+
consumer.close();
}
(pulsar) 02/03: [fix][broker] Fix BucketDelayedDeliveryTracker thread safety (#24014)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 50997afa08d4773b00de43deff2546c10fb22441
Author: 道君 <da...@apache.org>
AuthorDate: Tue Feb 25 14:51:19 2025 +0800
[fix][broker] Fix BucketDelayedDeliveryTracker thread safety (#24014)
(cherry picked from commit a83dd86756bd540c8cc0991afcd3d2ba333603b7)
---
.../pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java
index 08f3ae1fa6e..686daa92d13 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java
@@ -755,7 +755,7 @@ public class BucketDelayedDeliveryTracker extends AbstractDelayedDeliveryTracker
.orElse(false);
}
- public boolean containsMessage(long ledgerId, long entryId) {
+ public synchronized boolean containsMessage(long ledgerId, long entryId) {
if (lastMutableBucket.containsMessage(ledgerId, entryId)) {
return true;
}