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/19 10:10:12 UTC
(pulsar) branch branch-4.0 updated (cd1e9dbb282 -> e1d838cc713)
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 cd1e9dbb282 [fix][meta] Fix ephemeral handling of ZK nodes and fix MockZooKeeper ephemeral and ZK stat handling (#23988)
new 1a26a5ac056 [fix][test] Fix flaky test MetadataStoreTest.emptyStoreTest (#23998)
new e623a3c850c [improve][meta] Simplify getting parent path in ZKMetadataStore without using java.io.File (#23996)
new e1d838cc713 [fix][client] fix retry topic with exclusive mode. (#23859)
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:
.../apache/pulsar/client/api/RetryTopicTest.java | 40 ++++++++++++++++++++++
.../apache/pulsar/client/impl/ConsumerBase.java | 8 +++++
.../pulsar/metadata/impl/ZKMetadataStore.java | 7 ++--
.../apache/pulsar/metadata/MetadataStoreTest.java | 2 +-
4 files changed, 53 insertions(+), 4 deletions(-)
(pulsar) 02/03: [improve][meta] Simplify getting parent path in ZKMetadataStore without using java.io.File (#23996)
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 e623a3c850c78be95824b201caf23736991086d7
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Mon Feb 17 23:11:26 2025 -0800
[improve][meta] Simplify getting parent path in ZKMetadataStore without using java.io.File (#23996)
(cherry picked from commit 4bfdcd85ff5c9f21ba4fd976adf7785465053151)
---
.../main/java/org/apache/pulsar/metadata/impl/ZKMetadataStore.java | 7 ++++---
1 file changed, 4 insertions(+), 3 deletions(-)
diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKMetadataStore.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKMetadataStore.java
index dab3553f8ce..56c0962f250 100644
--- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKMetadataStore.java
+++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKMetadataStore.java
@@ -19,7 +19,6 @@
package org.apache.pulsar.metadata.impl;
import com.google.common.annotations.VisibleForTesting;
-import java.io.File;
import java.util.Collections;
import java.util.EnumSet;
import java.util.List;
@@ -610,8 +609,10 @@ public class ZKMetadataStore extends AbstractBatchedMetadataStore
if (rc != Code.NONODE.intValue()) {
callback.processResult(rc, path, ctx, name);
} else {
- String parent = (new File(originalPath)).getParent().replace("\\", "/");
-
+ String parent = parent(originalPath);
+ if (parent == null) {
+ parent = "/";
+ }
// Create parent nodes as "CONTAINER" so that ZK will automatically delete them when they're empty
asyncCreateFullPathOptimistic(zk, parent, new byte[0], CreateMode.CONTAINER,
(rc1, path1, ctx1, name1) -> {
(pulsar) 01/03: [fix][test] Fix flaky test MetadataStoreTest.emptyStoreTest (#23998)
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 1a26a5ac056670cf2f3ddf26cce9d058a2539bbc
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Mon Feb 17 22:22:06 2025 -0800
[fix][test] Fix flaky test MetadataStoreTest.emptyStoreTest (#23998)
(cherry picked from commit bcf8afb991d67f82eca55f76a046c86391364a47)
---
.../src/test/java/org/apache/pulsar/metadata/MetadataStoreTest.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/MetadataStoreTest.java b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/MetadataStoreTest.java
index a7b1dcf6bf0..7a333391dc9 100644
--- a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/MetadataStoreTest.java
+++ b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/MetadataStoreTest.java
@@ -114,7 +114,7 @@ public class MetadataStoreTest extends BaseMetadataStoreTest {
MetadataStoreConfig.builder().fsyncEnable(false).build());
String data = "data";
- String path = "/non-existing-key";
+ String path = "/concurrentPutTest";
int concurrent = 50;
List<CompletableFuture<Stat>> futureList = new ArrayList<>();
for (int i = 0; i < concurrent; i++) {
(pulsar) 03/03: [fix][client] fix retry topic with exclusive mode. (#23859)
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 e1d838cc713f1806edc10a40bb02c9eee922ad24
Author: Wenzhi Feng <52...@users.noreply.github.com>
AuthorDate: Tue Feb 18 17:59:52 2025 +0800
[fix][client] fix retry topic with exclusive mode. (#23859)
(cherry picked from commit 5a59ab7768e11db6ed92dada78b398feca9e24fc)
---
.../apache/pulsar/client/api/RetryTopicTest.java | 40 ++++++++++++++++++++++
.../apache/pulsar/client/impl/ConsumerBase.java | 8 +++++
2 files changed, 48 insertions(+)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/RetryTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/RetryTopicTest.java
index d0e72deb87f..2b897760b6f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/RetryTopicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/RetryTopicTest.java
@@ -136,6 +136,46 @@ public class RetryTopicTest extends ProducerConsumerBase {
checkConsumer.close();
}
+ /**
+ * Retry topic feature relies on the delay queue feature when consumer produce a delayed message
+ * to the retry topic. The delay queue feature is only supported in shared and key-shared subscription type.
+ * As a result, the subscription type of the retry topic should be shared or key-shared.
+ * @throws Exception
+ */
+ @Test
+ public void testRetryTopicWithExclusiveMode() throws Exception {
+ final String topic = "persistent://my-property/my-ns/retry-topic-exclusive";
+ final int maxRedeliveryCount = 2;
+
+ Consumer<byte[]> consumer = pulsarClient.newConsumer(Schema.BYTES)
+ .topic(topic)
+ .subscriptionName("my-subscription")
+ .subscriptionType(SubscriptionType.Exclusive)
+ .enableRetry(true)
+ .deadLetterPolicy(DeadLetterPolicy.builder().maxRedeliverCount(maxRedeliveryCount).build())
+ .receiverQueueSize(100)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+ .subscribe();
+
+ Producer<byte[]> producer = pulsarClient.newProducer(Schema.BYTES)
+ .topic(topic)
+ .create();
+
+ producer.send("Hello Pulsar".getBytes());
+ producer.close();
+
+ // receive message and set delay to 5 seconds
+ Message<byte[]> message = consumer.receive();
+ long timestamp = System.currentTimeMillis();
+ consumer.reconsumeLater(message, 4, TimeUnit.SECONDS);
+
+ // receive message and check the delay is at least 4 seconds
+ consumer.receive();
+ long delay = System.currentTimeMillis() - timestamp;
+ assertTrue(delay >= 2000);
+ consumer.close();
+ }
+
@Data
public static class Foo {
@Nullable
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
index 1ad8c6d28f1..27e2216e589 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
@@ -48,6 +48,7 @@ import org.apache.pulsar.client.api.BatchReceivePolicy;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.ConsumerBuilder;
import org.apache.pulsar.client.api.ConsumerEventListener;
+import org.apache.pulsar.client.api.DeadLetterPolicy;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.MessageIdAdv;
@@ -792,6 +793,13 @@ public abstract class ConsumerBase<T> extends HandlerState implements Consumer<T
}
protected SubType getSubType() {
+ // For retry topic, we always use Shared subscription
+ // Because we will produce delayed messages to retry topic.
+ DeadLetterPolicy deadLetterPolicy = conf.getDeadLetterPolicy();
+ if (deadLetterPolicy != null && topic.equals(deadLetterPolicy.getRetryLetterTopic())) {
+ return SubType.Shared;
+ }
+
SubscriptionType type = conf.getSubscriptionType();
switch (type) {
case Exclusive: