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 2022/04/04 12:33:09 UTC
[pulsar] branch branch-2.8 updated (02b208cc581 -> af554a06720)
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a change to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 02b208cc581 [C++] Fix flaky tests about reference count (#14854)
new e9546904be8 Fixes NPE - ``ReplicatedSubscriptionsController`` send marker message when enable deduplicated. (#14017)
new d9edbef8122 [Broker] Optimize RawReader#create when using Compactor (#14447)
new 35116b7b69e Fixed flaky test MemoryLimitTest#testRejectMessages (#14220) (#14628)
new 9e0b13fc4d8 [pulsar-functions] fix some IOExceptions when create functions from package URL (#14553)
new 365152f27c8 Provide an accurate error message when set ``autoTopicCreation `` (#14684)
new d73942b6ae6 Process maxRedeliverCount is 0 of DeadLeddterPolicy (#14706)
new a68f29c146a [fix][admin-cli]: Remove the trust certs check (#14764)
new d32f99b6ef9 [fix][broker] Fix cannot delete namespace with system topic (#14730)
new e067e008082 [fix][admin] Fix NPE in PulsarAdminBuilder when the service is not set (#14769)
new 4d9f4b3772b [fix][broker] Fix topic policy reader close bug. (#14897)
new 7e8f7a46ee5 [fix][broker] Fix wrong state for non-durable cursor (#14869)
new bb4862e396b [fix][security] Upgrade jackson and jackson-databind (2.13.2.1) to get rid of CVE-2020-36518 (#14871)
new c8b7eba53d1 [CI] Upgrade zlib version to 1.2.12 (#14964)
new a4c6ae42caf [fix][client] ConsumerBuilderImpl can not set null to deadLetterPolicy (#14980)
new af554a06720 [fix][broker] Fix potential NPE in Replicator (#15003)
The 15 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:
distribution/server/src/assemble/LICENSE.bin.txt | 16 ++--
.../bookkeeper/mledger/impl/ManagedCursorImpl.java | 2 +-
.../mledger/impl/NonDurableCursorImpl.java | 4 +-
pom.xml | 9 ++-
.../pulsar/broker/admin/impl/NamespacesBase.java | 7 +-
.../pulsar/broker/service/AbstractReplicator.java | 2 +-
.../SystemTopicBasedTopicPoliciesService.java | 4 +-
.../nonpersistent/NonPersistentReplicator.java | 2 +-
.../service/persistent/MessageDeduplication.java | 7 +-
.../service/persistent/PersistentReplicator.java | 2 +-
.../pulsar/broker/web/PulsarWebResource.java | 7 +-
.../org/apache/pulsar/client/api/RawReader.java | 2 +-
.../apache/pulsar/client/impl/RawReaderImpl.java | 2 +
.../broker/BrokerMessageDeduplicationTest.java | 73 ++++++++++++++++++
.../apache/pulsar/broker/admin/AdminApiTest2.java | 47 +++++++++++-
.../systopic/PartitionedSystemTopicTest.java | 5 ++
.../apache/pulsar/client/api/MemoryLimitTest.java | 86 ++++++++++++----------
.../org/apache/pulsar/client/impl/ReaderTest.java | 21 +++++-
.../data/{BundlesData.java => ValidateResult.java} | 27 ++++---
.../data/impl/AutoTopicCreationOverrideImpl.java | 18 +++--
.../pulsar/client/admin/internal/PackagesImpl.java | 3 +-
.../client/admin/internal/PulsarAdminImpl.java | 3 +
.../admin/internal/PulsarAdminBuilderImplTest.java | 21 +++---
pulsar-client-cpp/docker/alpine/Dockerfile | 8 +-
.../docker/alpine/Dockerfile-alpine-3.8 | 8 +-
pulsar-client-cpp/docker/manylinux1/Dockerfile | 8 +-
pulsar-client-cpp/docker/manylinux2014/Dockerfile | 8 +-
pulsar-client-cpp/pkg/deb/Dockerfile | 8 +-
pulsar-client-cpp/pkg/licenses/LICENSE-zlib.txt | 4 +-
pulsar-client-cpp/pkg/rpm/Dockerfile | 8 +-
.../org/apache/pulsar/admin/cli/CmdClusters.java | 5 --
.../pulsar/client/impl/ConsumerBuilderImpl.java | 3 +-
.../client/impl/ConsumerBuilderImplTest.java | 38 +++++++---
.../data/AutoTopicCreationOverrideTest.java | 12 +--
pulsar-functions/runtime-all/pom.xml | 1 -
.../functions/worker/rest/api/FunctionsImpl.java | 1 +
pulsar-sql/presto-distribution/LICENSE | 28 +++----
pulsar-sql/presto-distribution/pom.xml | 4 +-
.../websocket/AbstractWebSocketHandlerTest.java | 3 +-
39 files changed, 354 insertions(+), 163 deletions(-)
create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/BrokerMessageDeduplicationTest.java
copy pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/{BundlesData.java => ValidateResult.java} (64%)
copy pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/PersistencePoliciesTest.java => pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/PulsarAdminBuilderImplTest.java (62%)
[pulsar] 12/15: [fix][security] Upgrade jackson and jackson-databind (2.13.2.1) to get rid of CVE-2020-36518 (#14871)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit bb4862e396b0c0fa2a98af25679a6d6b70f2fa1d
Author: Nicolò Boschi <bo...@gmail.com>
AuthorDate: Mon Mar 28 18:18:42 2022 +0200
[fix][security] Upgrade jackson and jackson-databind (2.13.2.1) to get rid of CVE-2020-36518 (#14871)
* [fix][security] Upgrade JacksonXML to get rid of CVE-2020-36518
* force jackson-databind version
(cherry picked from commit 6d9ba7b60ab97257aa6e4e032a9764b5c3aefffa)
---
distribution/server/src/assemble/LICENSE.bin.txt | 16 +++++++-------
pom.xml | 9 ++++++--
pulsar-functions/runtime-all/pom.xml | 1 -
pulsar-sql/presto-distribution/LICENSE | 28 ++++++++++++------------
pulsar-sql/presto-distribution/pom.xml | 4 ++--
5 files changed, 31 insertions(+), 27 deletions(-)
diff --git a/distribution/server/src/assemble/LICENSE.bin.txt b/distribution/server/src/assemble/LICENSE.bin.txt
index 285a048a816..42dfa316eb8 100644
--- a/distribution/server/src/assemble/LICENSE.bin.txt
+++ b/distribution/server/src/assemble/LICENSE.bin.txt
@@ -312,14 +312,14 @@ The Apache Software License, Version 2.0
* JCommander -- com.beust-jcommander-1.78.jar
* High Performance Primitive Collections for Java -- com.carrotsearch-hppc-0.7.3.jar
* Jackson
- - com.fasterxml.jackson.core-jackson-annotations-2.12.6.jar
- - com.fasterxml.jackson.core-jackson-core-2.12.6.jar
- - com.fasterxml.jackson.core-jackson-databind-2.12.6.jar
- - com.fasterxml.jackson.dataformat-jackson-dataformat-yaml-2.12.6.jar
- - com.fasterxml.jackson.jaxrs-jackson-jaxrs-base-2.12.6.jar
- - com.fasterxml.jackson.jaxrs-jackson-jaxrs-json-provider-2.12.6.jar
- - com.fasterxml.jackson.module-jackson-module-jaxb-annotations-2.12.6.jar
- - com.fasterxml.jackson.module-jackson-module-jsonSchema-2.12.6.jar
+ - com.fasterxml.jackson.core-jackson-annotations-2.13.2.jar
+ - com.fasterxml.jackson.core-jackson-core-2.13.2.jar
+ - com.fasterxml.jackson.core-jackson-databind-2.13.2.1.jar
+ - com.fasterxml.jackson.dataformat-jackson-dataformat-yaml-2.13.2.jar
+ - com.fasterxml.jackson.jaxrs-jackson-jaxrs-base-2.13.2.jar
+ - com.fasterxml.jackson.jaxrs-jackson-jaxrs-json-provider-2.13.2.jar
+ - com.fasterxml.jackson.module-jackson-module-jaxb-annotations-2.13.2.jar
+ - com.fasterxml.jackson.module-jackson-module-jsonSchema-2.13.2.jar
* Caffeine -- com.github.ben-manes.caffeine-caffeine-2.9.1.jar
* Conscrypt -- org.conscrypt-conscrypt-openjdk-uber-2.5.2.jar
* Proto Google Common Protos -- com.google.api.grpc-proto-google-common-protos-2.0.1.jar
diff --git a/pom.xml b/pom.xml
index 8a60eb624ec..0570850f805 100644
--- a/pom.xml
+++ b/pom.xml
@@ -123,8 +123,8 @@ flexible messaging model and an intuitive client API.</description>
<log4j2.version>2.17.1</log4j2.version>
<bouncycastle.version>1.69</bouncycastle.version>
<bouncycastlefips.version>1.0.2</bouncycastlefips.version>
- <jackson.version>2.12.6</jackson.version>
- <jackson.databind.version>2.12.6</jackson.databind.version>
+ <jackson.version>2.13.2</jackson.version>
+ <jackson.databind.version>2.13.2.1</jackson.databind.version>
<reflections.version>0.9.11</reflections.version>
<swagger.version>1.6.2</swagger.version>
<puppycrawl.checkstyle.version>8.37</puppycrawl.checkstyle.version>
@@ -740,6 +740,11 @@ flexible messaging model and an intuitive client API.</description>
<type>pom</type>
<scope>import</scope>
</dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ <version>${jackson.databind.version}</version>
+ </dependency>
<dependency>
<artifactId>log4j</artifactId>
diff --git a/pulsar-functions/runtime-all/pom.xml b/pulsar-functions/runtime-all/pom.xml
index 747f663d6e5..6b131cf6067 100644
--- a/pulsar-functions/runtime-all/pom.xml
+++ b/pulsar-functions/runtime-all/pom.xml
@@ -76,7 +76,6 @@
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
- <version>${jackson.databind.version}</version>
</dependency>
<!--In order to support protobuf schema, this dependency needs to be added-->
diff --git a/pulsar-sql/presto-distribution/LICENSE b/pulsar-sql/presto-distribution/LICENSE
index f072e95aed7..137993c9934 100644
--- a/pulsar-sql/presto-distribution/LICENSE
+++ b/pulsar-sql/presto-distribution/LICENSE
@@ -207,19 +207,19 @@ This projects includes binary packages with the following licenses:
The Apache Software License, Version 2.0
* Jackson
- - jackson-annotations-2.12.6.jar
- - jackson-core-2.12.6.jar
- - jackson-databind-2.12.6.jar
- - jackson-dataformat-smile-2.12.6.jar
- - jackson-datatype-guava-2.12.6.jar
- - jackson-datatype-jdk8-2.12.6.jar
- - jackson-datatype-joda-2.12.6.jar
- - jackson-datatype-jsr310-2.12.6.jar
- - jackson-dataformat-yaml-2.12.6.jar
- - jackson-jaxrs-base-2.12.6.jar
- - jackson-jaxrs-json-provider-2.12.6.jar
- - jackson-module-jaxb-annotations-2.12.6.jar
- - jackson-module-jsonSchema-2.12.6.jar
+ - jackson-annotations-2.13.2.jar
+ - jackson-core-2.13.2.jar
+ - jackson-databind-2.13.2.1.jar
+ - jackson-dataformat-smile-2.13.2.jar
+ - jackson-datatype-guava-2.13.2.jar
+ - jackson-datatype-jdk8-2.13.2.jar
+ - jackson-datatype-joda-2.13.2.jar
+ - jackson-datatype-jsr310-2.13.2.jar
+ - jackson-dataformat-yaml-2.13.2.jar
+ - jackson-jaxrs-base-2.13.2.jar
+ - jackson-jaxrs-json-provider-2.13.2.jar
+ - jackson-module-jaxb-annotations-2.13.2.jar
+ - jackson-module-jsonSchema-2.13.2.jar
* Guava
- guava-30.1-jre.jar
- listenablefuture-9999.0-empty-to-avoid-conflict-with-guava.jar
@@ -439,7 +439,7 @@ The Apache Software License, Version 2.0
* Snappy
- snappy-java-1.1.7.jar
* Jackson
- - jackson-module-parameter-names-2.12.6.jar
+ - jackson-module-parameter-names-2.13.2.jar
* Java Assist
- javassist-3.25.0-GA.jar
* Java Native Access
diff --git a/pulsar-sql/presto-distribution/pom.xml b/pulsar-sql/presto-distribution/pom.xml
index 22197fb5552..39febca3372 100644
--- a/pulsar-sql/presto-distribution/pom.xml
+++ b/pulsar-sql/presto-distribution/pom.xml
@@ -39,10 +39,10 @@
<objenesis.version>2.6</objenesis.version>
<objectsize.version>0.0.12</objectsize.version>
<guice.version>4.2.0</guice.version>
- <jackson.version>2.12.6</jackson.version>
+ <jackson.version>2.13.2</jackson.version>
<!--fix Security Vulnerabilities-->
<!--https://www.cvedetails.com/vulnerability-list/vendor_id-15866/product_id-42991/Fasterxml-Jackson-databind.html-->
- <jackson.databind.version>2.12.6</jackson.databind.version>
+ <jackson.databind.version>2.13.2.1</jackson.databind.version>
<maven.version>3.0.5</maven.version>
<guava.version>30.1-jre</guava.version>
<asynchttpclient.version>2.12.1</asynchttpclient.version>
[pulsar] 02/15: [Broker] Optimize RawReader#create when using Compactor (#14447)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit d9edbef8122ca1d33185394822bf3143e50a16e1
Author: Qiang Zhao <74...@users.noreply.github.com>
AuthorDate: Thu Feb 24 23:14:31 2022 +0800
[Broker] Optimize RawReader#create when using Compactor (#14447)
(cherry picked from commit 992dfbd81f3dbdc648c30ff73e37b6c9ae2b276e)
---
pulsar-broker/src/main/java/org/apache/pulsar/client/api/RawReader.java | 2 +-
.../src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java | 2 ++
2 files changed, 3 insertions(+), 1 deletion(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/client/api/RawReader.java b/pulsar-broker/src/main/java/org/apache/pulsar/client/api/RawReader.java
index f74157a9389..fe068f2f943 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/client/api/RawReader.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/client/api/RawReader.java
@@ -34,7 +34,7 @@ public interface RawReader {
static CompletableFuture<RawReader> create(PulsarClient client, String topic, String subscription) {
CompletableFuture<Consumer<byte[]>> future = new CompletableFuture<>();
RawReader r = new RawReaderImpl((PulsarClientImpl) client, topic, subscription, future);
- return future.thenCompose(x -> x.seekAsync(MessageId.earliest)).thenApply(__ -> r);
+ return future.thenApply(__ -> r);
}
/**
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java
index 217dd5ccc85..6b032370898 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java
@@ -31,6 +31,7 @@ import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.RawMessage;
import org.apache.pulsar.client.api.RawReader;
import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
import org.apache.pulsar.common.api.proto.CommandAck.AckType;
@@ -56,6 +57,7 @@ public class RawReaderImpl implements RawReader {
consumerConfiguration.setSubscriptionType(SubscriptionType.Exclusive);
consumerConfiguration.setReceiverQueueSize(DEFAULT_RECEIVER_QUEUE_SIZE);
consumerConfiguration.setReadCompacted(true);
+ consumerConfiguration.setSubscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
consumer = new RawConsumerImpl(client, consumerConfiguration,
consumerFuture);
[pulsar] 10/15: [fix][broker] Fix topic policy reader close bug. (#14897)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 4d9f4b3772bf1cf52ec9cc7b2e1b255fcb5b37df
Author: Jiwei Guo <te...@apache.org>
AuthorDate: Sun Mar 27 22:07:03 2022 +0800
[fix][broker] Fix topic policy reader close bug. (#14897)
### Motivation
https://github.com/apache/pulsar/issues/14896 is flaky, after diving into the codes, I find it's a bug about closing topic policy reader. We should use `ex.getCause` instead of `ex`.
Stacktrace：
```
2022-03-27T10:42:16,795+0800 [broker-client-shared-internal-executor-58-1] WARN org.apache.pulsar.broker.service.SystemTopicBasedTopicPoliciesService - Read more topic polices exception, read again.
java.util.concurrent.CompletionException: org.apache.pulsar.client.api.PulsarClientException$AlreadyClosedException: Consumer already closed
at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:292) ~[?:1.8.0_291]
at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:308) ~[?:1.8.0_291]
at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:607) ~[?:1.8.0_291]
at java.util.concurrent.CompletableFuture.uniApplyStage(CompletableFuture.java:628) ~[?:1.8.0_291]
at java.util.concurrent.CompletableFuture.thenApply(CompletableFuture.java:1996) ~[?:1.8.0_291]
at org.apache.pulsar.client.impl.MultiTopicsReaderImpl.readNextAsync(MultiTopicsReaderImpl.java:140) ~[classes/:?]
```
(cherry picked from commit 85c1ba50cb155111f969403a092affd536d9dcb9)
---
.../pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java | 4 +++-
.../apache/pulsar/broker/systopic/PartitionedSystemTopicTest.java | 5 +++++
2 files changed, 8 insertions(+), 1 deletion(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
index 8d002be73de..cccd3a30d1d 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
@@ -47,6 +47,7 @@ import org.apache.pulsar.common.naming.NamespaceBundle;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.TopicPolicies;
+import org.apache.pulsar.common.util.FutureUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -357,7 +358,8 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
notifyListener(msg);
readMorePolicies(reader);
} else {
- if (ex instanceof PulsarClientException.AlreadyClosedException) {
+ Throwable cause = FutureUtil.unwrapCompletionException(ex);
+ if (cause instanceof PulsarClientException.AlreadyClosedException) {
log.error("Read more topic policies exception, close the read now!", ex);
cleanCacheAndCloseReader(
reader.getSystemTopic().getTopicName().getNamespaceObject(), false);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/systopic/PartitionedSystemTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/systopic/PartitionedSystemTopicTest.java
index 7dd02bdeae7..bbd3cae7117 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/systopic/PartitionedSystemTopicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/systopic/PartitionedSystemTopicTest.java
@@ -74,6 +74,7 @@ public class PartitionedSystemTopicTest extends BrokerTestBase {
Assert.assertEquals(admin.topics().getPartitionedTopicList(ns).size(), 1);
Assert.assertEquals(partitions, PARTITIONS);
Assert.assertEquals(admin.topics().getList(ns).size(), PARTITIONS);
+ reader.close();
}
@Test(timeOut = 1000 * 60)
@@ -97,6 +98,10 @@ public class PartitionedSystemTopicTest extends BrokerTestBase {
.subscribeAsync());
}
FutureUtil.waitForAll(futureList).get();
+ // Close all the consumers after check
+ for (CompletableFuture<Consumer<byte[]>> consumer : futureList) {
+ consumer.join().close();
+ }
}
}
[pulsar] 03/15: Fixed flaky test MemoryLimitTest#testRejectMessages (#14220) (#14628)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 35116b7b69eb6158bebbe82523dfb65499603b8e
Author: wuxuanqicn <89...@users.noreply.github.com>
AuthorDate: Sat Mar 12 13:50:05 2022 +0800
Fixed flaky test MemoryLimitTest#testRejectMessages (#14220) (#14628)
Co-authored-by: xuanqi.wu <xu...@weimob.com>
(cherry picked from commit 5f8db372ee3926f93eb109ab3b713038c3b523c8)
---
.../apache/pulsar/client/api/MemoryLimitTest.java | 86 ++++++++++++----------
1 file changed, 47 insertions(+), 39 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MemoryLimitTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MemoryLimitTest.java
index ec98e7d1dbe..431991e61aa 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MemoryLimitTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MemoryLimitTest.java
@@ -18,20 +18,22 @@
*/
package org.apache.pulsar.client.api;
-import static org.testng.Assert.assertEquals;
-import static org.testng.Assert.fail;
-
-import java.util.concurrent.CountDownLatch;
-
import lombok.Cleanup;
-
import org.apache.pulsar.client.api.PulsarClientException.MemoryBufferIsFullError;
-import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.ProducerImpl;
+import org.apache.pulsar.client.impl.PulsarTestClient;
+import org.awaitility.Awaitility;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
+import java.time.Duration;
+import java.util.concurrent.TimeUnit;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.fail;
+
@Test(groups = "broker-api")
public class MemoryLimitTest extends ProducerConsumerBase {
@@ -62,27 +64,30 @@ public class MemoryLimitTest extends ProducerConsumerBase {
throws Exception {
String topic = newTopicName();
- @Cleanup
- PulsarClientImpl client = (PulsarClientImpl) PulsarClient.builder()
+ ClientBuilder clientBuilder = PulsarClient.builder()
.serviceUrl(pulsar.getBrokerServiceUrl())
- .memoryLimit(100, SizeUnit.KILO_BYTES)
- .build();
+ .memoryLimit(100, SizeUnit.KILO_BYTES);
@Cleanup
- Producer<byte[]> producer = client.newProducer()
+ PulsarTestClient client = PulsarTestClient.create(clientBuilder);
+
+ @Cleanup
+ ProducerImpl<byte[]> producer = (ProducerImpl<byte[]>) client.newProducer()
.topic(topic)
.blockIfQueueFull(false)
+ .sendTimeout(5, TimeUnit.SECONDS)
.create();
+ // make sure all message pending at pendingMessages queue
+ // connection with broker can not be established, so handleSendReceipt will not be invoked while sending message
+ client.dropOpSendMessages();
final int n = 101;
- CountDownLatch latch = new CountDownLatch(n);
-
for (int i = 0; i < n; i++) {
- producer.sendAsync(new byte[1024]).thenRun(() -> {
- latch.countDown();
- });
+ producer.sendAsync(new byte[1024]);
}
-
+ Awaitility.await()
+ .atMost(Duration.ofSeconds(5))
+ .until(() -> producer.getPendingQueueSize() == n);
assertEquals(client.getMemoryLimitController().currentUsage(), n * 1024);
try {
@@ -92,8 +97,10 @@ public class MemoryLimitTest extends ProducerConsumerBase {
// Expected
}
- latch.await();
-
+ client.allowReconnecting();
+ Awaitility.await()
+ .atMost(Duration.ofSeconds(30))
+ .until(() -> producer.getPendingQueueSize() == 0);
assertEquals(client.getMemoryLimitController().currentUsage(), 0);
// We should now be able to send again
@@ -105,41 +112,40 @@ public class MemoryLimitTest extends ProducerConsumerBase {
String t1 = newTopicName();
String t2 = newTopicName();
- @Cleanup
- PulsarClientImpl client = (PulsarClientImpl) PulsarClient.builder()
+ ClientBuilder clientBuilder = PulsarClient.builder()
.serviceUrl(pulsar.getBrokerServiceUrl())
- .memoryLimit(100, SizeUnit.KILO_BYTES)
- .build();
+ .memoryLimit(100, SizeUnit.KILO_BYTES);
+
+ @Cleanup
+ PulsarTestClient client = PulsarTestClient.create(clientBuilder);
@Cleanup
- Producer<byte[]> p1 = client.newProducer()
+ ProducerImpl<byte[]> p1 = (ProducerImpl<byte[]>) client.newProducer()
.topic(t1)
.blockIfQueueFull(false)
+ .sendTimeout(5, TimeUnit.SECONDS)
.create();
@Cleanup
- Producer<byte[]> p2 = client.newProducer()
+ ProducerImpl<byte[]> p2 = (ProducerImpl<byte[]>) client.newProducer()
.topic(t2)
.blockIfQueueFull(false)
+ .sendTimeout(5, TimeUnit.SECONDS)
.create();
+ client.dropOpSendMessages();
final int n = 101;
- CountDownLatch latch = new CountDownLatch(n);
-
for (int i = 0; i < n / 2; i++) {
- p1.sendAsync(new byte[1024]).thenRun(() -> {
- latch.countDown();
- });
- p2.sendAsync(new byte[1024]).thenRun(() -> {
- latch.countDown();
- });
+ p1.sendAsync(new byte[1024]);
+ p2.sendAsync(new byte[1024]);
}
// Last message in order to reach the limit
- p1.sendAsync(new byte[1024]).thenRun(() -> {
- latch.countDown();
- });
+ p1.sendAsync(new byte[1024]);
+ Awaitility.await()
+ .atMost(Duration.ofSeconds(5))
+ .until(() -> (p1.getPendingQueueSize() + p2.getPendingQueueSize()) == n);
assertEquals(client.getMemoryLimitController().currentUsage(), n * 1024);
try {
@@ -156,8 +162,10 @@ public class MemoryLimitTest extends ProducerConsumerBase {
// Expected
}
- latch.await();
-
+ client.allowReconnecting();
+ Awaitility.await()
+ .atMost(Duration.ofSeconds(30))
+ .until(() -> (p1.getPendingQueueSize() + p2.getPendingQueueSize()) == 0);
assertEquals(client.getMemoryLimitController().currentUsage(), 0);
// We should now be able to send again
[pulsar] 05/15: Provide an accurate error message when set ``autoTopicCreation `` (#14684)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 365152f27c8fdb8d1ea8ef96265c89d228db55e6
Author: Qiang Zhao <74...@users.noreply.github.com>
AuthorDate: Thu Mar 17 19:47:50 2022 +0800
Provide an accurate error message when set ``autoTopicCreation `` (#14684)
(cherry picked from commit 50a7e50745e5c26200a4abd2e63e3750e34339fb)
---
.../pulsar/broker/admin/impl/NamespacesBase.java | 7 ++--
.../common/policies/data/ValidateResult.java | 40 ++++++++++++++++++++++
.../data/impl/AutoTopicCreationOverrideImpl.java | 18 +++++-----
.../data/AutoTopicCreationOverrideTest.java | 12 +++----
4 files changed, 61 insertions(+), 16 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
index e91314d11d6..264e800b715 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
@@ -99,6 +99,7 @@ import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
import org.apache.pulsar.common.policies.data.SubscribeRate;
import org.apache.pulsar.common.policies.data.SubscriptionAuthMode;
import org.apache.pulsar.common.policies.data.TenantOperation;
+import org.apache.pulsar.common.policies.data.ValidateResult;
import org.apache.pulsar.common.policies.data.impl.AutoTopicCreationOverrideImpl;
import org.apache.pulsar.common.policies.data.impl.DispatchRateImpl;
import org.apache.pulsar.common.util.FutureUtil;
@@ -834,9 +835,11 @@ public abstract class NamespacesBase extends AdminResource {
validateNamespacePolicyOperation(namespaceName, PolicyName.AUTO_TOPIC_CREATION, PolicyOperation.WRITE);
validatePoliciesReadOnlyAccess();
if (autoTopicCreationOverride != null) {
- if (!AutoTopicCreationOverrideImpl.isValidOverride(autoTopicCreationOverride)) {
+ ValidateResult validateResult = AutoTopicCreationOverrideImpl.validateOverride(autoTopicCreationOverride);
+ if (!validateResult.isSuccess()) {
throw new RestException(Status.PRECONDITION_FAILED,
- "Invalid configuration for autoTopicCreationOverride");
+ "Invalid configuration for autoTopicCreationOverride. the detail is "
+ + validateResult.getErrorInfo());
}
if (maxPartitions > 0 && autoTopicCreationOverride.getDefaultNumPartitions() > maxPartitions) {
throw new RestException(Status.NOT_ACCEPTABLE,
diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/ValidateResult.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/ValidateResult.java
new file mode 100644
index 00000000000..13821947cf2
--- /dev/null
+++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/ValidateResult.java
@@ -0,0 +1,40 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.policies.data;
+
+import lombok.Getter;
+
+@Getter
+public class ValidateResult {
+ private final boolean success;
+ private final String errorInfo;
+
+ private ValidateResult(boolean success, String errorInfo) {
+ this.success = success;
+ this.errorInfo = errorInfo;
+ }
+
+ public static ValidateResult fail(String errorInfo) {
+ return new ValidateResult(false, errorInfo);
+ }
+
+ public static ValidateResult success() {
+ return new ValidateResult(true, null);
+ }
+}
diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/impl/AutoTopicCreationOverrideImpl.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/impl/AutoTopicCreationOverrideImpl.java
index 1ce60d1c78b..ba6bc07780a 100644
--- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/impl/AutoTopicCreationOverrideImpl.java
+++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/impl/AutoTopicCreationOverrideImpl.java
@@ -23,6 +23,7 @@ import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.pulsar.common.policies.data.AutoTopicCreationOverride;
import org.apache.pulsar.common.policies.data.TopicType;
+import org.apache.pulsar.common.policies.data.ValidateResult;
/**
* Override of autoTopicCreation settings on a namespace level.
@@ -35,28 +36,29 @@ public final class AutoTopicCreationOverrideImpl implements AutoTopicCreationOve
private String topicType;
private Integer defaultNumPartitions;
- public static boolean isValidOverride(AutoTopicCreationOverride override) {
+ public static ValidateResult validateOverride(AutoTopicCreationOverride override) {
if (override == null) {
- return false;
+ return ValidateResult.fail("[AutoTopicCreationOverride] can not be null");
}
if (override.isAllowAutoTopicCreation()) {
if (!TopicType.isValidTopicType(override.getTopicType())) {
- return false;
+ return ValidateResult.fail(String.format("Unknown topic type [%s]", override.getTopicType()));
}
if (TopicType.PARTITIONED.toString().equals(override.getTopicType())) {
if (override.getDefaultNumPartitions() == null) {
- return false;
+ return ValidateResult.fail("[defaultNumPartitions] cannot be null when the type is partitioned.");
}
- if (!(override.getDefaultNumPartitions() > 0)) {
- return false;
+ if (override.getDefaultNumPartitions() <= 0) {
+ return ValidateResult.fail("[defaultNumPartitions] cannot be less than 1 for partition type.");
}
} else if (TopicType.NON_PARTITIONED.toString().equals(override.getTopicType())) {
if (override.getDefaultNumPartitions() != null) {
- return false;
+ return ValidateResult.fail("[defaultNumPartitions] is not allowed to be"
+ + " set when the type is non-partition.");
}
}
}
- return true;
+ return ValidateResult.success();
}
public static AutoTopicCreationOverrideImplBuilder builder() {
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/AutoTopicCreationOverrideTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/AutoTopicCreationOverrideTest.java
index 5092d433d0d..66769f0bfbc 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/AutoTopicCreationOverrideTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/AutoTopicCreationOverrideTest.java
@@ -32,7 +32,7 @@ public class AutoTopicCreationOverrideTest {
.allowAutoTopicCreation(true)
.topicType(TopicType.NON_PARTITIONED.toString())
.build();
- assertTrue(AutoTopicCreationOverrideImpl.isValidOverride(override));
+ assertTrue(AutoTopicCreationOverrideImpl.validateOverride(override).isSuccess());
}
@Test
@@ -42,7 +42,7 @@ public class AutoTopicCreationOverrideTest {
.topicType(TopicType.PARTITIONED.toString())
.defaultNumPartitions(2)
.build();
- assertTrue(AutoTopicCreationOverrideImpl.isValidOverride(override));
+ assertTrue(AutoTopicCreationOverrideImpl.validateOverride(override).isSuccess());
}
@Test
@@ -51,7 +51,7 @@ public class AutoTopicCreationOverrideTest {
.allowAutoTopicCreation(true)
.topicType("aaa")
.build();
- assertFalse(AutoTopicCreationOverrideImpl.isValidOverride(override));
+ assertFalse(AutoTopicCreationOverrideImpl.validateOverride(override).isSuccess());
}
@Test
@@ -61,7 +61,7 @@ public class AutoTopicCreationOverrideTest {
.topicType(TopicType.PARTITIONED.toString())
.defaultNumPartitions(0)
.build();
- assertFalse(AutoTopicCreationOverrideImpl.isValidOverride(override));
+ assertFalse(AutoTopicCreationOverrideImpl.validateOverride(override).isSuccess());
}
@Test
@@ -70,7 +70,7 @@ public class AutoTopicCreationOverrideTest {
.allowAutoTopicCreation(true)
.topicType(TopicType.PARTITIONED.toString())
.build();
- assertFalse(AutoTopicCreationOverrideImpl.isValidOverride(override));
+ assertFalse(AutoTopicCreationOverrideImpl.validateOverride(override).isSuccess());
}
@Test
@@ -80,6 +80,6 @@ public class AutoTopicCreationOverrideTest {
.topicType(TopicType.NON_PARTITIONED.toString())
.defaultNumPartitions(2)
.build();
- assertFalse(AutoTopicCreationOverrideImpl.isValidOverride(override));
+ assertFalse(AutoTopicCreationOverrideImpl.validateOverride(override).isSuccess());
}
}
[pulsar] 11/15: [fix][broker] Fix wrong state for non-durable cursor (#14869)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 7e8f7a46ee5f5b57da2517908002496a26e8e8d2
Author: Qiang Zhao <74...@users.noreply.github.com>
AuthorDate: Mon Mar 28 16:43:29 2022 +0800
[fix][broker] Fix wrong state for non-durable cursor (#14869)
### Motivation
The current non-durable cursor does not have the correct state.
For example, when the reader is created, I always see the cursor status as ``Uninitialized`` via the ``getInternalStats`` method.
```json
{
"reader-xxxxx" : {
"markDeletePosition" : "19785:18718",
"readPosition" : "19807:42735",
"waitingReadOp" : false,
"pendingReadOps" : 0,
"messagesConsumedCounter" : -2257,
"cursorLedger" : -1,
"cursorLedgerLastEntry" : -1,
"individuallyDeletedMessages" : "[]",
"lastLedgerSwitchTimestamp" : "2022-03-24T20:03:51.85Z",
"state" : "Uninitialized",
"numberOfEntriesSinceFirstNotAckedMessage" : 744993,
"totalNonContiguousDeletedMessagesRange" : 0,
"subscriptionHavePendingRead" : true,
"subscriptionHavePendingReplayRead" : false,
"properties" : { }
}
}
```
### Modifications
- Correct the cursor state.
(cherry picked from commit b477557a6f6d36ae18f09d4edfa076e9092a17fe)
---
.../bookkeeper/mledger/impl/ManagedCursorImpl.java | 2 +-
.../mledger/impl/NonDurableCursorImpl.java | 4 ++--
.../org/apache/pulsar/client/impl/ReaderTest.java | 21 ++++++++++++++++++++-
3 files changed, 23 insertions(+), 4 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
index 7597ba0bb7b..c31e304be85 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
@@ -236,7 +236,7 @@ public class ManagedCursorImpl implements ManagedCursor {
Closed // The managed cursor has been closed
}
- private static final AtomicReferenceFieldUpdater<ManagedCursorImpl, State> STATE_UPDATER =
+ protected static final AtomicReferenceFieldUpdater<ManagedCursorImpl, State> STATE_UPDATER =
AtomicReferenceFieldUpdater.newUpdater(ManagedCursorImpl.class, State.class, "state");
protected volatile State state = null;
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/NonDurableCursorImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/NonDurableCursorImpl.java
index 1d545bd6b48..4625f5b5800 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/NonDurableCursorImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/NonDurableCursorImpl.java
@@ -61,7 +61,7 @@ public class NonDurableCursorImpl extends ManagedCursorImpl {
// read-position
recoverCursor(startCursorPosition);
}
-
+ STATE_UPDATER.set(this, State.Open);
log.info("[{}] Created non-durable cursor read-position={} mark-delete-position={}", ledger.getName(),
readPosition, markDeletePosition);
}
@@ -110,7 +110,7 @@ public class NonDurableCursorImpl extends ManagedCursorImpl {
@Override
public void asyncClose(CloseCallback callback, Object ctx) {
- // No-Op
+ STATE_UPDATER.set(this, State.Closed);
callback.closeComplete(ctx);
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ReaderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ReaderTest.java
index 205264699eb..53ae8f97008 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ReaderTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ReaderTest.java
@@ -36,7 +36,6 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
-
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
@@ -55,6 +54,8 @@ import org.apache.pulsar.client.api.ReaderBuilder;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.ClusterData;
+import org.apache.pulsar.common.policies.data.ManagedLedgerInternalStats;
+import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats;
import org.apache.pulsar.common.policies.data.RetentionPolicies;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
import org.apache.pulsar.common.policies.data.TopicStats;
@@ -602,4 +603,22 @@ public class ReaderTest extends MockedPulsarServiceBaseTest {
});
}
+ @Test
+ public void testReaderCursorStatsCorrect() throws Exception {
+ final String readerNotAckTopic = "persistent://my-property/my-ns/testReaderCursorStatsCorrect";
+ @Cleanup
+ Reader<byte[]> reader = pulsarClient.newReader()
+ .topic(readerNotAckTopic)
+ .startMessageId(MessageId.earliest)
+ .create();
+ PersistentTopicInternalStats internalStats = admin.topics().getInternalStats(readerNotAckTopic);
+ Assert.assertEquals(internalStats.cursors.size(), 1);
+ String key = new ArrayList<>(internalStats.cursors.keySet()).get(0);
+ ManagedLedgerInternalStats.CursorStats cursor = internalStats.cursors.get(key);
+ Assert.assertEquals(cursor.state, "Open");
+ reader.close();
+ internalStats = admin.topics().getInternalStats(readerNotAckTopic);
+ Assert.assertEquals(internalStats.cursors.size(), 0);
+ }
+
}
[pulsar] 01/15: Fixes NPE - ``ReplicatedSubscriptionsController`` send marker message when enable deduplicated. (#14017)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit e9546904be8c5ff197f5f6829a344e5288715d09
Author: Qiang Zhao <74...@users.noreply.github.com>
AuthorDate: Sat Jan 29 01:43:53 2022 +0800
Fixes NPE - ``ReplicatedSubscriptionsController`` send marker message when enable deduplicated. (#14017)
* Fixes marker message cause NPE.
* Fixes marker message cause NPE.
(cherry picked from commit 80f47d14774b85a23af050b09bfb8cb896163a48)
---
.../service/persistent/MessageDeduplication.java | 7 ++-
.../broker/BrokerMessageDeduplicationTest.java | 73 ++++++++++++++++++++++
2 files changed, 77 insertions(+), 3 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java
index b6f51467edf..613c5a3116b 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java
@@ -82,7 +82,8 @@ public class MessageDeduplication {
Failed,
}
- enum MessageDupStatus {
+ @VisibleForTesting
+ public enum MessageDupStatus {
// whether a message is a definitely a duplicate or not cannot be determined at this time
Unknown,
// message is definitely NOT a duplicate
@@ -315,7 +316,7 @@ public class MessageDeduplication {
* @return true if the message should be published or false if it was recognized as a duplicate
*/
public MessageDupStatus isDuplicate(PublishContext publishContext, ByteBuf headersAndPayload) {
- if (!isEnabled()) {
+ if (!isEnabled() || publishContext.isMarkerMessage()) {
return MessageDupStatus.NotDup;
}
@@ -368,7 +369,7 @@ public class MessageDeduplication {
* Call this method whenever a message is persisted to get the chance to trigger a snapshot.
*/
public void recordMessagePersisted(PublishContext publishContext, PositionImpl position) {
- if (!isEnabled()) {
+ if (!isEnabled() || publishContext.isMarkerMessage()) {
return;
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/BrokerMessageDeduplicationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/BrokerMessageDeduplicationTest.java
new file mode 100644
index 00000000000..f2d7a21a40f
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/BrokerMessageDeduplicationTest.java
@@ -0,0 +1,73 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker;
+
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
+import org.apache.bookkeeper.mledger.ManagedLedger;
+import org.apache.pulsar.broker.service.Topic;
+import org.apache.pulsar.broker.service.persistent.MessageDeduplication;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.testng.annotations.Test;
+
+public class BrokerMessageDeduplicationTest {
+
+ @Test
+ public void markerMessageNotDeduplicated() {
+ PulsarService pulsarService = mock(PulsarService.class);
+ ServiceConfiguration configuration = new ServiceConfiguration();
+ doReturn(configuration).when(pulsarService).getConfiguration();
+ MessageDeduplication deduplication = spy(new MessageDeduplication(pulsarService,
+ mock(PersistentTopic.class), mock(ManagedLedger.class)));
+ doReturn(true).when(deduplication).isEnabled();
+ Topic.PublishContext context = mock(Topic.PublishContext.class);
+ doReturn(true).when(context).isMarkerMessage();
+ MessageDeduplication.MessageDupStatus status = deduplication.isDuplicate(context, null);
+ assertEquals(status, MessageDeduplication.MessageDupStatus.NotDup);
+ }
+
+ @Test
+ public void markerMessageNotRecordPersistent() {
+ PulsarService pulsarService = mock(PulsarService.class);
+ ServiceConfiguration configuration = new ServiceConfiguration();
+ doReturn(configuration).when(pulsarService).getConfiguration();
+ MessageDeduplication deduplication = spy(new MessageDeduplication(pulsarService,
+ mock(PersistentTopic.class), mock(ManagedLedger.class)));
+ doReturn(true).when(deduplication).isEnabled();
+ Topic.PublishContext context = mock(Topic.PublishContext.class);
+ // marker message don't record message persisted.
+ doReturn(true).when(context).isMarkerMessage();
+ deduplication.recordMessagePersisted(context, null);
+
+ // if is not a marker message, we will get NPE. because context is mocked with null value fields.
+ doReturn(false).when(context).isMarkerMessage();
+ try {
+ deduplication.recordMessagePersisted(context, null);
+ fail();
+ } catch (Exception npe) {
+ assertTrue(npe instanceof NullPointerException);
+ }
+ }
+
+
+}
[pulsar] 08/15: [fix][broker] Fix cannot delete namespace with system topic (#14730)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit d32f99b6ef9213153af51efa0a485308ef4773ec
Author: Jiwei Guo <te...@apache.org>
AuthorDate: Tue Mar 22 19:26:18 2022 +0800
[fix][broker] Fix cannot delete namespace with system topic (#14730)
(cherry picked from commit 7556c4e0165660e8dbd141c4e93bb9e31a67e6f9)
---
.../pulsar/broker/web/PulsarWebResource.java | 7 +++-
.../apache/pulsar/broker/admin/AdminApiTest2.java | 47 +++++++++++++++++++++-
2 files changed, 51 insertions(+), 3 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
index 8fe63753b17..bc60b4ee326 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
@@ -733,7 +733,12 @@ public abstract class PulsarWebResource {
pulsarService.getConfigurationCache().policiesCache().getAsync(path).thenAccept(policiesResult -> {
if (policiesResult.isPresent()) {
Policies policies = policiesResult.get();
- if (policies.replication_clusters.isEmpty()) {
+ if (policies.deleted) {
+ String msg = String.format("Namespace %s is deleted", namespace.toString());
+ log.warn(msg);
+ validationFuture.completeExceptionally(new RestException(Status.PRECONDITION_FAILED,
+ "Namespace is deleted"));
+ } else if (policies.replication_clusters.isEmpty()) {
String msg = String.format(
"Namespace does not have any clusters configured : local_cluster=%s ns=%s",
localCluster, namespace.toString());
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java
index e43cf4fbd11..581bd33f0a4 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java
@@ -46,7 +46,6 @@ import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.UUID;
-import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;
import javax.ws.rs.core.Response.Status;
import lombok.Cleanup;
@@ -83,8 +82,9 @@ import org.apache.pulsar.common.naming.TopicDomain;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.AutoFailoverPolicyData;
import org.apache.pulsar.common.policies.data.AutoFailoverPolicyType;
-import org.apache.pulsar.common.policies.data.BrokerNamespaceIsolationDataImpl;
+import org.apache.pulsar.common.policies.data.BacklogQuota;
import org.apache.pulsar.common.policies.data.BrokerNamespaceIsolationData;
+import org.apache.pulsar.common.policies.data.BrokerNamespaceIsolationDataImpl;
import org.apache.pulsar.common.policies.data.BundlesData;
import org.apache.pulsar.common.policies.data.ClusterData;
import org.apache.pulsar.common.policies.data.ConsumerStats;
@@ -98,6 +98,7 @@ import org.apache.pulsar.common.policies.data.RetentionPolicies;
import org.apache.pulsar.common.policies.data.SubscriptionStats;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
import org.apache.pulsar.common.policies.data.TopicStats;
+import org.apache.pulsar.common.policies.data.impl.BacklogQuotaImpl;
import org.awaitility.Awaitility;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
@@ -1269,6 +1270,48 @@ public class AdminApiTest2 extends MockedPulsarServiceBaseTest {
assertEquals(topicStats.getSubscriptions().get(subName).getMsgBacklog(), 9);
}
+ @Test
+ public void testDeleteNamespaceWithTopicPolicies() throws Exception {
+ stopBroker();
+ conf.setSystemTopicEnabled(true);
+ conf.setTopicLevelPoliciesEnabled(true);
+ setup();
+
+ String tenant = "test-tenant";
+ assertFalse(admin.tenants().getTenants().contains(tenant));
+
+ // create tenant
+ admin.tenants().createTenant(tenant,
+ new TenantInfoImpl(Sets.newHashSet("role1", "role2"), Sets.newHashSet("test")));
+ assertTrue(admin.tenants().getTenants().contains(tenant));
+
+ // create namespace2
+ String namespace = tenant + "/test-ns2";
+ admin.namespaces().createNamespace(namespace, Sets.newHashSet("test"));
+ // create topic
+ String topic = namespace + "/test-topic2";
+ Producer<byte[]> producer = pulsarClient.newProducer().topic(topic).create();
+ producer.send("test".getBytes(StandardCharsets.UTF_8));
+ BacklogQuota backlogQuota = BacklogQuotaImpl
+ .builder()
+ .limitTime(1000)
+ .limitSize(1000)
+ .retentionPolicy(BacklogQuota.RetentionPolicy.producer_exception)
+ .build();
+ admin.topics().setBacklogQuota(topic, backlogQuota);
+ Awaitility.await().untilAsserted(() -> {
+ Assert.assertEquals(admin.topics()
+ .getBacklogQuotaMap(topic)
+ .get(BacklogQuota.BacklogQuotaType.destination_storage), backlogQuota);
+ });
+ producer.close();
+ admin.topics().delete(topic);
+ admin.namespaces().deleteNamespace(namespace);
+ Awaitility.await().untilAsserted(() -> {
+ assertTrue(admin.namespaces().getNamespaces(tenant).isEmpty());
+ });
+ }
+
@Test(timeOut = 30000)
public void testBacklogNoDelayed() throws PulsarClientException, PulsarAdminException, InterruptedException {
final String topic = "persistent://prop-xyz/ns1/precise-back-log-no-delayed-" + UUID.randomUUID().toString();
[pulsar] 06/15: Process maxRedeliverCount is 0 of DeadLeddterPolicy (#14706)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit d73942b6ae67c1be6a3501d0d221e762df00bcd4
Author: xiaolong ran <xi...@tencent.com>
AuthorDate: Fri Mar 18 12:05:26 2022 +0800
Process maxRedeliverCount is 0 of DeadLeddterPolicy (#14706)
Signed-off-by: xiaolongran <xi...@tencent.com>
Fixes #14704
When the user uses the function of DeadLetterPolicy, it is better than misoperation. MaxRedeliverCount may be set to 0. When it is set to 0, according to the current processing logic of the Java Client, the message will be pushed to the DeadLetter Topic every time.
- When MaxRedeliverCount <= 0 in DeadLetterPolicy, we reset MaxRedeliverCount to default value
(cherry picked from commit 601fbdd40eabfcd5d5519e0c5bcc20ae280a8e18)
---
.../apache/pulsar/client/impl/ConsumerBuilderImpl.java | 2 ++
.../pulsar/client/impl/ConsumerBuilderImplTest.java | 17 ++++++++++++++++-
.../pulsar/websocket/AbstractWebSocketHandlerTest.java | 3 ++-
3 files changed, 20 insertions(+), 2 deletions(-)
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java
index f902d765873..d01a2076d8d 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java
@@ -426,6 +426,8 @@ public class ConsumerBuilderImpl<T> implements ConsumerBuilder<T> {
if (conf.getAckTimeoutMillis() == 0) {
conf.setAckTimeoutMillis(DEFAULT_ACK_TIMEOUT_MILLIS_FOR_DEAD_LETTER);
}
+
+ checkArgument(deadLetterPolicy.getMaxRedeliverCount() > 0, "MaxRedeliverCount must be > 0.");
conf.setDeadLetterPolicy(deadLetterPolicy);
}
return this;
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerBuilderImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerBuilderImplTest.java
index b2ce170b124..36ea53f31ef 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerBuilderImplTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerBuilderImplTest.java
@@ -18,7 +18,13 @@
*/
package org.apache.pulsar.client.impl;
-import org.apache.pulsar.client.api.*;
+import org.apache.pulsar.client.api.BatchReceivePolicy;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.DeadLetterPolicy;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.SubscriptionMode;
import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
import org.testng.annotations.BeforeTest;
import org.testng.annotations.Test;
@@ -283,6 +289,15 @@ public class ConsumerBuilderImplTest {
.build());
}
+ @Test(expectedExceptions = IllegalArgumentException.class)
+ public void testRedeliverCountOfDeadLetterPolicy() {
+ consumerBuilderImpl.deadLetterPolicy(DeadLetterPolicy.builder()
+ .maxRedeliverCount(0)
+ .deadLetterTopic("test-dead-letter-topic")
+ .retryLetterTopic("test-retry-letter-topic")
+ .build());
+ }
+
@Test
public void testConsumerBuilderImplWhenNumericPropertiesAreValid() {
consumerBuilderImpl.negativeAckRedeliveryDelay(1, TimeUnit.MILLISECONDS);
diff --git a/pulsar-websocket/src/test/java/org/apache/pulsar/websocket/AbstractWebSocketHandlerTest.java b/pulsar-websocket/src/test/java/org/apache/pulsar/websocket/AbstractWebSocketHandlerTest.java
index 9bd99076779..782e05ea625 100644
--- a/pulsar-websocket/src/test/java/org/apache/pulsar/websocket/AbstractWebSocketHandlerTest.java
+++ b/pulsar-websocket/src/test/java/org/apache/pulsar/websocket/AbstractWebSocketHandlerTest.java
@@ -369,11 +369,12 @@ public class AbstractWebSocketHandlerTest {
consumerHandler.clearQueryParams();
consumerHandler.putQueryParam("receiverQueueSize", "1001");
consumerHandler.putQueryParam("deadLetterTopic", "dead-letter-topic");
+ consumerHandler.putQueryParam("maxRedeliverCount", "3");
conf = consumerHandler.getConf();
// receive queue size is the minimum value of default value (1000) and user defined value(1001)
assertEquals(conf.getReceiverQueueSize(), 1000);
assertEquals(conf.getDeadLetterPolicy().getDeadLetterTopic(), "dead-letter-topic");
- assertEquals(conf.getDeadLetterPolicy().getMaxRedeliverCount(), 0);
+ assertEquals(conf.getDeadLetterPolicy().getMaxRedeliverCount(), 3);
}
}
[pulsar] 15/15: [fix][broker] Fix potential NPE in Replicator (#15003)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit af554a06720ad88d9fbea78c469a2341c05a6b40
Author: Qiang Huang <HQ...@users.noreply.github.com>
AuthorDate: Sat Apr 2 16:54:58 2022 +0800
[fix][broker] Fix potential NPE in Replicator (#15003)
### Motivation
```java
public CompletableFuture<MessageId> getFuture() {
return null;
}
```
The return value null may cause potential NPE in ` getFuture`. The method does similarly： CompletableFuture<Void> remove()` & .
### Modifications
Use ` return CompletableFuture.completedFuture(null);` instead of `return null;`
(cherry picked from commit b604b4c9fb9fdae832bfabc61a0527ee0aec566f)
---
.../main/java/org/apache/pulsar/broker/service/AbstractReplicator.java | 2 +-
.../pulsar/broker/service/nonpersistent/NonPersistentReplicator.java | 2 +-
.../apache/pulsar/broker/service/persistent/PersistentReplicator.java | 2 +-
3 files changed, 3 insertions(+), 3 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java
index 34d2b6ecb86..a29382fcdef 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java
@@ -203,7 +203,7 @@ public abstract class AbstractReplicator {
public CompletableFuture<Void> remove() {
// No-op
- return null;
+ return CompletableFuture.completedFuture(null);
}
protected boolean isWritable() {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentReplicator.java
index b003db823ed..744f47298c7 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentReplicator.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentReplicator.java
@@ -227,7 +227,7 @@ public class NonPersistentReplicator extends AbstractReplicator implements Repli
@Override
public CompletableFuture<MessageId> getFuture() {
- return null;
+ return CompletableFuture.completedFuture(null);
}
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java
index 0f3f18c4492..03617195cee 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java
@@ -482,7 +482,7 @@ public class PersistentReplicator extends AbstractReplicator
@Override
public CompletableFuture<MessageId> getFuture() {
- return null;
+ return CompletableFuture.completedFuture(null);
}
}
[pulsar] 09/15: [fix][admin] Fix NPE in PulsarAdminBuilder when the service is not set (#14769)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit e067e00808292f3330575aa6409c30e56f703e6b
Author: Zike Yang <zk...@streamnative.io>
AuthorDate: Tue Mar 22 18:29:23 2022 +0800
[fix][admin] Fix NPE in PulsarAdminBuilder when the service is not set (#14769)
(cherry picked from commit 0b6b1e2079a9afe92527df2cdb0e843a0ae391dd)
---
.../client/admin/internal/PulsarAdminImpl.java | 3 ++
.../admin/internal/PulsarAdminBuilderImplTest.java | 38 ++++++++++++++++++++++
2 files changed, 41 insertions(+)
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PulsarAdminImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PulsarAdminImpl.java
index e4fdf132896..b80799fffc9 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PulsarAdminImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PulsarAdminImpl.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.client.admin.internal;
+import static com.google.common.base.Preconditions.checkArgument;
import java.io.IOException;
import java.net.URL;
import java.util.Map;
@@ -146,6 +147,8 @@ public class PulsarAdminImpl implements PulsarAdmin {
int autoCertRefreshTime,
TimeUnit autoCertRefreshTimeUnit,
ClassLoader clientBuilderClassLoader) throws PulsarClientException {
+ checkArgument(StringUtils.isNotBlank(serviceUrl), "Service URL needs to be specified");
+
this.connectTimeout = connectTimeout;
this.connectTimeoutUnit = connectTimeoutUnit;
this.readTimeout = readTimeout;
diff --git a/pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/PulsarAdminBuilderImplTest.java b/pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/PulsarAdminBuilderImplTest.java
new file mode 100644
index 00000000000..1ea45401eec
--- /dev/null
+++ b/pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/PulsarAdminBuilderImplTest.java
@@ -0,0 +1,38 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.admin.internal;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.fail;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.testng.annotations.Test;
+
+public class PulsarAdminBuilderImplTest {
+
+ @Test
+ public void testAdminBuilderWithServiceUrlNotSet() throws PulsarClientException {
+ try{
+ PulsarAdmin.builder().build();
+ fail();
+ } catch (IllegalArgumentException exception) {
+ assertEquals("Service URL needs to be specified", exception.getMessage());
+ }
+ }
+}
[pulsar] 14/15: [fix][client] ConsumerBuilderImpl can not set null to deadLetterPolicy (#14980)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit a4c6ae42caf3563ca17e3425bda0ccba990062f5
Author: JiangHaiting <ji...@apache.org>
AuthorDate: Sat Apr 2 11:16:02 2022 +0800
[fix][client] ConsumerBuilderImpl can not set null to deadLetterPolicy (#14980)
`ConsumerBuilderImpl` can not set null to deadLetterPolicy if it's already assigned a non-null value.
Calls `conf.setDeadLetterPolicy` if `deadLetterPolicy` is null.
(cherry picked from commit 2f530717851d5e852b4489ff70184d14634ef2b9)
---
.../pulsar/client/impl/ConsumerBuilderImpl.java | 3 +--
.../client/impl/ConsumerBuilderImplTest.java | 29 +++++++++++++---------
2 files changed, 18 insertions(+), 14 deletions(-)
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java
index d01a2076d8d..5857006e876 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java
@@ -426,10 +426,9 @@ public class ConsumerBuilderImpl<T> implements ConsumerBuilder<T> {
if (conf.getAckTimeoutMillis() == 0) {
conf.setAckTimeoutMillis(DEFAULT_ACK_TIMEOUT_MILLIS_FOR_DEAD_LETTER);
}
-
checkArgument(deadLetterPolicy.getMaxRedeliverCount() > 0, "MaxRedeliverCount must be > 0.");
- conf.setDeadLetterPolicy(deadLetterPolicy);
}
+ conf.setDeadLetterPolicy(deadLetterPolicy);
return this;
}
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerBuilderImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerBuilderImplTest.java
index 36ea53f31ef..d648ee75af7 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerBuilderImplTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerBuilderImplTest.java
@@ -18,6 +18,17 @@
*/
package org.apache.pulsar.client.impl;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertNotNull;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
import org.apache.pulsar.client.api.BatchReceivePolicy;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.DeadLetterPolicy;
@@ -29,18 +40,6 @@ import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
import org.testng.annotations.BeforeTest;
import org.testng.annotations.Test;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.TimeUnit;
-import java.util.regex.Pattern;
-
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-import static org.testng.Assert.assertNotNull;
-
/**
* Unit tests of {@link ConsumerBuilderImpl}.
*/
@@ -298,6 +297,12 @@ public class ConsumerBuilderImplTest {
.build());
}
+ @Test
+ public void testNullDeadLetterPolicy() {
+ consumerBuilderImpl.deadLetterPolicy(null);
+ verify(consumerBuilderImpl.getConf()).setDeadLetterPolicy(null);
+ }
+
@Test
public void testConsumerBuilderImplWhenNumericPropertiesAreValid() {
consumerBuilderImpl.negativeAckRedeliveryDelay(1, TimeUnit.MILLISECONDS);
[pulsar] 13/15: [CI] Upgrade zlib version to 1.2.12 (#14964)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit c8b7eba53d116162d199986ee806e72a9f1971d6
Author: Lishen Yao <ya...@gmail.com>
AuthorDate: Fri Apr 1 00:42:21 2022 +0800
[CI] Upgrade zlib version to 1.2.12 (#14964)
Co-authored-by: Zike Yang <zi...@apache.org>
(cherry picked from commit 8b2b988654887cdb9d123698d5237381704f89c3)
---
pulsar-client-cpp/docker/alpine/Dockerfile | 8 ++++----
pulsar-client-cpp/docker/alpine/Dockerfile-alpine-3.8 | 8 ++++----
pulsar-client-cpp/docker/manylinux1/Dockerfile | 8 ++++----
pulsar-client-cpp/docker/manylinux2014/Dockerfile | 8 ++++----
pulsar-client-cpp/pkg/deb/Dockerfile | 8 ++++----
pulsar-client-cpp/pkg/licenses/LICENSE-zlib.txt | 4 ++--
pulsar-client-cpp/pkg/rpm/Dockerfile | 8 ++++----
7 files changed, 26 insertions(+), 26 deletions(-)
diff --git a/pulsar-client-cpp/docker/alpine/Dockerfile b/pulsar-client-cpp/docker/alpine/Dockerfile
index ef77284242c..12d1e2f9f97 100644
--- a/pulsar-client-cpp/docker/alpine/Dockerfile
+++ b/pulsar-client-cpp/docker/alpine/Dockerfile
@@ -43,12 +43,12 @@ RUN curl -O -L https://boostorg.jfrog.io/artifactory/main/release/1.72.0/source/
rm -rf /boost_1_72_0.tar.gz /boost_1_72_0
# ZLib
-RUN curl -O -L https://zlib.net/zlib-1.2.11.tar.gz && \
- tar xfz zlib-1.2.11.tar.gz && \
- cd zlib-1.2.11 && \
+RUN curl -O -L https://zlib.net/zlib-1.2.12.tar.gz && \
+ tar xfz zlib-1.2.12.tar.gz && \
+ cd zlib-1.2.12 && \
CFLAGS="-fPIC -O3" ./configure && \
make -j4 && make install && \
- rm -rf /zlib-1.2.11.tar.gz /zlib-1.2.11
+ rm -rf /zlib-1.2.12.tar.gz /zlib-1.2.12
# Compile OpenSSL
RUN curl -O -L https://github.com/openssl/openssl/archive/OpenSSL_1_1_0j.tar.gz && \
diff --git a/pulsar-client-cpp/docker/alpine/Dockerfile-alpine-3.8 b/pulsar-client-cpp/docker/alpine/Dockerfile-alpine-3.8
index 0a9fbb40711..862785e9015 100644
--- a/pulsar-client-cpp/docker/alpine/Dockerfile-alpine-3.8
+++ b/pulsar-client-cpp/docker/alpine/Dockerfile-alpine-3.8
@@ -43,12 +43,12 @@ RUN curl -O -L https://boostorg.jfrog.io/artifactory/main/release/1.72.0/source/
rm -rf /boost_1_72_0.tar.gz /boost_1_72_0
# ZLib
-RUN curl -O -L https://zlib.net/zlib-1.2.11.tar.gz && \
- tar xfz zlib-1.2.11.tar.gz && \
- cd zlib-1.2.11 && \
+RUN curl -O -L https://zlib.net/zlib-1.2.12.tar.gz && \
+ tar xfz zlib-1.2.12.tar.gz && \
+ cd zlib-1.2.12 && \
CFLAGS="-fPIC -O3" ./configure && \
make -j4 && make install && \
- rm -rf /zlib-1.2.11.tar.gz /zlib-1.2.11
+ rm -rf /zlib-1.2.12.tar.gz /zlib-1.2.12
# Compile OpenSSL
RUN curl -O -L https://github.com/openssl/openssl/archive/OpenSSL_1_1_0j.tar.gz && \
diff --git a/pulsar-client-cpp/docker/manylinux1/Dockerfile b/pulsar-client-cpp/docker/manylinux1/Dockerfile
index 0df92401a8f..502c6f6fe77 100644
--- a/pulsar-client-cpp/docker/manylinux1/Dockerfile
+++ b/pulsar-client-cpp/docker/manylinux1/Dockerfile
@@ -46,12 +46,12 @@ RUN curl -O -L https://www.cpan.org/src/5.0/perl-5.10.0.tar.gz && \
####################################
# ZLib
-RUN curl -O -L https://zlib.net/zlib-1.2.11.tar.gz && \
- tar xvfz zlib-1.2.11.tar.gz && \
- cd zlib-1.2.11 && \
+RUN curl -O -L https://zlib.net/zlib-1.2.12.tar.gz && \
+ tar xvfz zlib-1.2.12.tar.gz && \
+ cd zlib-1.2.12 && \
CFLAGS="-fPIC -O3" ./configure && \
make && make install && \
- rm -rf /zlib-1.2.11.tar.gz /zlib-1.2.11
+ rm -rf /zlib-1.2.12.tar.gz /zlib-1.2.12
# Compile OpenSSL
RUN curl -O -L https://github.com/openssl/openssl/archive/OpenSSL_1_1_0j.tar.gz && \
diff --git a/pulsar-client-cpp/docker/manylinux2014/Dockerfile b/pulsar-client-cpp/docker/manylinux2014/Dockerfile
index 20a7247e30a..850c0b7a8d0 100644
--- a/pulsar-client-cpp/docker/manylinux2014/Dockerfile
+++ b/pulsar-client-cpp/docker/manylinux2014/Dockerfile
@@ -46,12 +46,12 @@ RUN curl -O -L https://www.cpan.org/src/5.0/perl-5.10.0.tar.gz && \
####################################
# ZLib
-RUN curl -O -L https://zlib.net/zlib-1.2.11.tar.gz && \
- tar xvfz zlib-1.2.11.tar.gz && \
- cd zlib-1.2.11 && \
+RUN curl -O -L https://zlib.net/zlib-1.2.12.tar.gz && \
+ tar xvfz zlib-1.2.12.tar.gz && \
+ cd zlib-1.2.12 && \
CFLAGS="-fPIC -O3" ./configure && \
make && make install && \
- rm -rf /zlib-1.2.11.tar.gz /zlib-1.2.11
+ rm -rf /zlib-1.2.12.tar.gz /zlib-1.2.12
# Compile OpenSSL
RUN curl -O -L https://github.com/openssl/openssl/archive/OpenSSL_1_1_0j.tar.gz && \
diff --git a/pulsar-client-cpp/pkg/deb/Dockerfile b/pulsar-client-cpp/pkg/deb/Dockerfile
index 77c3f470c06..1925827b56a 100644
--- a/pulsar-client-cpp/pkg/deb/Dockerfile
+++ b/pulsar-client-cpp/pkg/deb/Dockerfile
@@ -49,12 +49,12 @@ RUN curl -O -L https://github.com/google/protobuf/releases/download/v3.3.0/prot
rm -rf /protobuf-cpp-3.3.0.tar.gz /protobuf-3.3.0
# ZLib
-RUN curl -O -L https://github.com/madler/zlib/archive/v1.2.11.tar.gz && \
- tar xvfz v1.2.11.tar.gz && \
- cd zlib-1.2.11 && \
+RUN curl -O -L https://github.com/madler/zlib/archive/v1.2.12.tar.gz && \
+ tar xvfz v1.2.12.tar.gz && \
+ cd zlib-1.2.12 && \
CFLAGS="-fPIC -O3" ./configure && \
make && make install && \
- rm -rf /v1.2.11.tar.gz /zlib-1.2.11
+ rm -rf /v1.2.12.tar.gz /zlib-1.2.12
# Zstandard
RUN curl -O -L https://github.com/facebook/zstd/releases/download/v1.3.7/zstd-1.3.7.tar.gz && \
diff --git a/pulsar-client-cpp/pkg/licenses/LICENSE-zlib.txt b/pulsar-client-cpp/pkg/licenses/LICENSE-zlib.txt
index cf25d04b7e1..f1f93cd6a6a 100644
--- a/pulsar-client-cpp/pkg/licenses/LICENSE-zlib.txt
+++ b/pulsar-client-cpp/pkg/licenses/LICENSE-zlib.txt
@@ -1,7 +1,7 @@
zlib.h -- interface of the 'zlib' general purpose compression library
- version 1.2.11, January 15th, 2017
+ version 1.2.12, March 27th, 2022
- Copyright (C) 1995-2017 Jean-loup Gailly and Mark Adler
+ Copyright (C) 1995-2022 Jean-loup Gailly and Mark Adler
This software is provided 'as-is', without any express or implied
warranty. In no event will the authors be held liable for any damages
diff --git a/pulsar-client-cpp/pkg/rpm/Dockerfile b/pulsar-client-cpp/pkg/rpm/Dockerfile
index c2406c5c7d3..e83290dccca 100644
--- a/pulsar-client-cpp/pkg/rpm/Dockerfile
+++ b/pulsar-client-cpp/pkg/rpm/Dockerfile
@@ -49,12 +49,12 @@ RUN curl -O -L https://github.com/google/protobuf/releases/download/v3.3.0/prot
rm -rf /protobuf-cpp-3.3.0.tar.gz /protobuf-3.3.0
# ZLib
-RUN curl -O -L https://github.com/madler/zlib/archive/v1.2.11.tar.gz && \
- tar xvfz v1.2.11.tar.gz && \
- cd zlib-1.2.11 && \
+RUN curl -O -L https://github.com/madler/zlib/archive/v1.2.12.tar.gz && \
+ tar xvfz v1.2.12.tar.gz && \
+ cd zlib-1.2.12 && \
CFLAGS="-fPIC -O3" ./configure && \
make && make install && \
- rm -rf /v1.2.11.tar.gz /zlib-1.2.11
+ rm -rf /v1.2.12.tar.gz /zlib-1.2.12
# Zstandard
RUN curl -O -L https://github.com/facebook/zstd/releases/download/v1.3.7/zstd-1.3.7.tar.gz && \
[pulsar] 04/15: [pulsar-functions] fix some IOExceptions when create functions from package URL (#14553)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 9e0b13fc4d862e9197e07bfe6cc34d57c07537f5
Author: Rui Fu <fr...@users.noreply.github.com>
AuthorDate: Wed Mar 16 19:01:27 2022 +0800
[pulsar-functions] fix some IOExceptions when create functions from package URL (#14553)
(cherry picked from commit f0d166f36e1fbd4df1e20ae2ccc7fcae822c17b4)
---
.../java/org/apache/pulsar/client/admin/internal/PackagesImpl.java | 3 ++-
.../org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java | 1 +
2 files changed, 3 insertions(+), 1 deletion(-)
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PackagesImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PackagesImpl.java
index 4c7fc4cf299..77749e6421b 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PackagesImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PackagesImpl.java
@@ -25,6 +25,7 @@ import java.io.InputStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
+import java.nio.file.StandardCopyOption;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
@@ -182,7 +183,7 @@ public class PackagesImpl extends ComponentResource implements Packages {
if (destinyPath.getParent() != null) {
Files.createDirectories(destinyPath.getParent());
}
- Files.copy(inputStream, destinyPath);
+ Files.copy(inputStream, destinyPath, StandardCopyOption.REPLACE_EXISTING);
future.complete(null);
} catch (IOException e) {
future.completeExceptionally(e);
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java
index dba3645c3e5..b2d0f36787e 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java
@@ -771,6 +771,7 @@ public class FunctionsImpl extends ComponentImpl implements Functions<PulsarWork
// use the Nar extraction directory as a temporary directory for downloaded files
tempDirectory = Paths.get(worker.getWorkerConfig().getNarExtractionDirectory());
}
+ Files.createDirectories(tempDirectory);
File file = Files.createTempFile(tempDirectory, "function", ".tmp").toFile();
worker.getBrokerAdmin().packages().download(packageName, file.toString());
return file;
[pulsar] 07/15: [fix][admin-cli]: Remove the trust certs check (#14764)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit a68f29c146ad36c12cae8a6dd606e4aa81d345b8
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Tue Mar 22 00:07:52 2022 +0800
[fix][admin-cli]: Remove the trust certs check (#14764)
(cherry picked from commit 44f92a8118e3ce0461cd4dc1736860fab77f0cae)
---
.../src/main/java/org/apache/pulsar/admin/cli/CmdClusters.java | 5 -----
1 file changed, 5 deletions(-)
diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdClusters.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdClusters.java
index eb1ca4f1360..d43c8496472 100644
--- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdClusters.java
+++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdClusters.java
@@ -76,11 +76,6 @@ public class CmdClusters extends CmdBase {
"You must specify tls-trust-store-type, tls-trust-store and tls-trust-store-pwd"
+ " when enable tls-enable-keystore");
}
- } else {
- if (StringUtils.isBlank(clusterData.getBrokerClientTrustCertsFilePath())) {
- throw new RuntimeException("You must specify tls-trust-certs-filepath"
- + " when tls-enable-keystore is not enable");
- }
}
}
}