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 2020/12/21 13:04:06 UTC
[pulsar] branch branch-2.7 updated (be5c038 -> 4da0baf)
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a change to branch branch-2.7
in repository https://gitbox.apache.org/repos/asf/pulsar.git.
from be5c038 Support configure max subscriptions per topic on the namespace level policy (#8924)
new 5a13be8 [ISSUE 8832] Avro custom schema not working in consumer (#8939)
new c83da2c Expose non-contiguous deleted messages ranges stats. (#8936)
new 934d3b8 Fix NSQ source META-INF file name and sourceConfigClass (#8941)
new 7468543 Fix subscription dispatch rate does not work after the topic unload without dispatch rate limit. (#8947)
new 40135da [pulsar-broker] capture stats with precise backlog (#8928)
new 61a6aff Intercept beforeSendMessage calls (#8932)
new 4da0baf [C++] Add consumer's configs for reader (#8905)
The 7 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/bookkeeper/mledger/ManagedCursor.java | 5 +
.../bookkeeper/mledger/impl/ManagedCursorImpl.java | 13 +-
.../mledger/impl/ManagedCursorContainerTest.java | 5 +
.../pulsar/broker/intercept/BrokerInterceptor.java | 26 +++-
.../BrokerInterceptorWithClassLoader.java | 12 ++
.../broker/intercept/BrokerInterceptors.java | 17 +++
.../broker/service/AbstractBaseDispatcher.java | 16 ++-
.../apache/pulsar/broker/service/Dispatcher.java | 5 +
.../apache/pulsar/broker/service/Subscription.java | 3 +
.../nonpersistent/NonPersistentSubscription.java | 6 +
.../PersistentDispatcherMultipleConsumers.java | 15 +++
.../PersistentDispatcherSingleActiveConsumer.java | 15 +++
.../service/persistent/PersistentSubscription.java | 9 ++
.../broker/service/persistent/PersistentTopic.java | 14 +-
.../pulsar/broker/admin/TopicPoliciesTest.java | 31 +++++
.../broker/intercept/BrokerInterceptorTest.java | 29 +++++
.../broker/intercept/CounterBrokerInterceptor.java | 18 +++
.../pulsar/broker/stats/SubscriptionStatsTest.java | 43 +++++++
.../client/impl/BrokerClientIntegrationTest.java | 39 +++++-
.../include/pulsar/ReaderConfiguration.h | 93 ++++++++++++++
pulsar-client-cpp/lib/ReaderConfiguration.cc | 76 +++++++++++
pulsar-client-cpp/lib/ReaderConfigurationImpl.h | 20 +--
pulsar-client-cpp/lib/ReaderImpl.cc | 18 +++
pulsar-client-cpp/lib/ReaderImpl.h | 10 +-
pulsar-client-cpp/tests/ReaderConfigurationTest.cc | 141 +++++++++++++++++++++
.../pulsar/client/impl/schema/AvroSchema.java | 6 +-
.../common/policies/data/SubscriptionStats.java | 11 ++
.../pulsar/common/policies/data/TopicStats.java | 10 ++
.../services/{pulsar-io.yml => pulsar-io.yaml} | 2 +-
29 files changed, 679 insertions(+), 29 deletions(-)
create mode 100644 pulsar-client-cpp/tests/ReaderConfigurationTest.cc
rename pulsar-io/nsq/src/main/resources/META-INF/services/{pulsar-io.yml => pulsar-io.yaml} (93%)
[pulsar] 06/07: Intercept beforeSendMessage calls (#8932)
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.7
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 61a6affc1af775b925060765e6ca59adda406a09
Author: Sijie Guo <si...@apache.org>
AuthorDate: Mon Dec 14 16:26:16 2020 -0800
Intercept beforeSendMessage calls (#8932)
*Motivation*
Currently the message metadata headers are deserialized for filtering
out the entries to dispatch. Add a `beforeSendMessage` method to
intercept entries before sending them to consumers
(cherry picked from commit f76655a9d23d1130cb89e4bf5431722f470efbf4)
---
.../pulsar/broker/intercept/BrokerInterceptor.java | 26 +++++++++++++++----
.../BrokerInterceptorWithClassLoader.java | 12 +++++++++
.../broker/intercept/BrokerInterceptors.java | 17 +++++++++++++
.../broker/service/AbstractBaseDispatcher.java | 16 +++++++++++-
.../apache/pulsar/broker/service/Subscription.java | 3 +++
.../nonpersistent/NonPersistentSubscription.java | 6 +++++
.../service/persistent/PersistentSubscription.java | 6 +++++
.../broker/intercept/BrokerInterceptorTest.java | 29 ++++++++++++++++++++++
.../broker/intercept/CounterBrokerInterceptor.java | 18 ++++++++++++++
9 files changed, 127 insertions(+), 6 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptor.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptor.java
index c4fc628..0f43a96 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptor.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptor.java
@@ -18,18 +18,20 @@
*/
package org.apache.pulsar.broker.intercept;
+import java.io.IOException;
+import javax.servlet.ServletException;
+import javax.servlet.ServletRequest;
+import javax.servlet.ServletResponse;
+import org.apache.bookkeeper.mledger.Entry;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.service.ServerCnx;
+import org.apache.pulsar.broker.service.Subscription;
import org.apache.pulsar.common.api.proto.PulsarApi.BaseCommand;
+import org.apache.pulsar.common.api.proto.PulsarApi.MessageMetadata;
import org.apache.pulsar.common.classification.InterfaceAudience;
import org.apache.pulsar.common.classification.InterfaceStability;
import org.apache.pulsar.common.intercept.InterceptException;
-import javax.servlet.ServletException;
-import javax.servlet.ServletRequest;
-import javax.servlet.ServletResponse;
-import java.io.IOException;
-
/**
* A plugin interface that allows you to intercept the
* client requests to the Pulsar brokers.
@@ -42,6 +44,20 @@ import java.io.IOException;
public interface BrokerInterceptor extends AutoCloseable {
/**
+ * Intercept messages before sending them to the consumers.
+ *
+ * @param subscription pulsar subscription
+ * @param entry entry
+ * @param ackSet entry ack bitset. it is either <tt>null</tt> or an array of long-based bitsets.
+ * @param msgMetadata message metadata. The message metadata will be recycled after this call.
+ */
+ default void beforeSendMessage(Subscription subscription,
+ Entry entry,
+ long[] ackSet,
+ MessageMetadata msgMetadata) {
+ }
+
+ /**
* Called by the broker while new command incoming.
*/
void onPulsarCommand(BaseCommand command, ServerCnx cnx) throws InterceptException;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptorWithClassLoader.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptorWithClassLoader.java
index 00c960c..68710cd 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptorWithClassLoader.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptorWithClassLoader.java
@@ -21,9 +21,12 @@ package org.apache.pulsar.broker.intercept;
import lombok.Data;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.Entry;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.service.ServerCnx;
+import org.apache.pulsar.broker.service.Subscription;
import org.apache.pulsar.common.api.proto.PulsarApi.BaseCommand;
+import org.apache.pulsar.common.api.proto.PulsarApi.MessageMetadata;
import org.apache.pulsar.common.intercept.InterceptException;
import org.apache.pulsar.common.nar.NarClassLoader;
@@ -44,6 +47,15 @@ public class BrokerInterceptorWithClassLoader implements BrokerInterceptor {
private final NarClassLoader classLoader;
@Override
+ public void beforeSendMessage(Subscription subscription,
+ Entry entry,
+ long[] ackSet,
+ MessageMetadata msgMetadata) {
+ this.interceptor.beforeSendMessage(
+ subscription, entry, ackSet, msgMetadata);
+ }
+
+ @Override
public void onPulsarCommand(BaseCommand command, ServerCnx cnx) throws InterceptException {
this.interceptor.onPulsarCommand(command, cnx);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptors.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptors.java
index 6f0b7b8..77cca6d 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptors.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptors.java
@@ -20,10 +20,13 @@ package org.apache.pulsar.broker.intercept;
import com.google.common.collect.ImmutableMap;
import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.Entry;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.service.ServerCnx;
+import org.apache.pulsar.broker.service.Subscription;
import org.apache.pulsar.common.api.proto.PulsarApi.BaseCommand;
+import org.apache.pulsar.common.api.proto.PulsarApi.MessageMetadata;
import org.apache.pulsar.common.intercept.InterceptException;
import javax.servlet.ServletException;
@@ -86,6 +89,20 @@ public class BrokerInterceptors implements BrokerInterceptor {
}
@Override
+ public void beforeSendMessage(Subscription subscription,
+ Entry entry,
+ long[] ackSet,
+ MessageMetadata msgMetadata) {
+ for (BrokerInterceptorWithClassLoader value : interceptors.values()) {
+ value.beforeSendMessage(
+ subscription,
+ entry,
+ ackSet,
+ msgMetadata);
+ }
+ }
+
+ @Override
public void onPulsarCommand(BaseCommand command, ServerCnx cnx) throws InterceptException {
for (BrokerInterceptorWithClassLoader value : interceptors.values()) {
value.onPulsarCommand(command, cnx);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java
index 513d6b7..2813d70 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java
@@ -32,6 +32,7 @@ import org.apache.bookkeeper.mledger.ManagedCursor;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.impl.PositionImpl;
import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pulsar.broker.intercept.BrokerInterceptor;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.common.api.proto.PulsarApi;
import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
@@ -135,14 +136,27 @@ public abstract class AbstractBaseDispatcher implements Dispatcher {
totalBytes += metadataAndPayload.readableBytes();
totalChunkedMessages += msgMetadata.hasChunkId() ? 1: 0;
batchSizes.setBatchSize(i, batchSize);
+ long[] ackSet = null;
if (indexesAcks != null && cursor != null) {
- long[] ackSet = cursor.getDeletedBatchIndexesAsLongArray(PositionImpl.get(entry.getLedgerId(), entry.getEntryId()));
+ ackSet = cursor.getDeletedBatchIndexesAsLongArray(
+ PositionImpl.get(entry.getLedgerId(), entry.getEntryId()));
if (ackSet != null) {
indexesAcks.setIndexesAcks(i, Pair.of(batchSize, ackSet));
} else {
indexesAcks.setIndexesAcks(i,null);
}
}
+
+ BrokerInterceptor interceptor = subscription.interceptor();
+ if (null != interceptor) {
+ interceptor.beforeSendMessage(
+ subscription,
+ entry,
+ ackSet,
+ msgMetadata
+ );
+ }
+
} finally {
msgMetadata.recycle();
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Subscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Subscription.java
index 8f6b5fc..1c05399 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Subscription.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Subscription.java
@@ -25,12 +25,15 @@ import java.util.concurrent.CompletableFuture;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.pulsar.broker.intercept.BrokerInterceptor;
import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
import org.apache.pulsar.common.api.proto.PulsarApi.CommandSubscribe.SubType;
import org.apache.pulsar.common.api.proto.PulsarMarkers.ReplicatedSubscriptionsSnapshot;
public interface Subscription {
+ BrokerInterceptor interceptor();
+
Topic getTopic();
String getName();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java
index 954d4c9..635ce3e 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java
@@ -29,6 +29,7 @@ import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.impl.PositionImpl;
import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.intercept.BrokerInterceptor;
import org.apache.pulsar.broker.service.BrokerServiceException;
import org.apache.pulsar.broker.service.BrokerServiceException.ServerMetadataException;
import org.apache.pulsar.broker.service.BrokerServiceException.SubscriptionBusyException;
@@ -77,6 +78,11 @@ public class NonPersistentSubscription implements Subscription {
}
@Override
+ public BrokerInterceptor interceptor() {
+ return this.topic.getBrokerService().getInterceptor();
+ }
+
+ @Override
public String getName() {
return this.subName;
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
index 57b5bbc..0cc4e97 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
@@ -46,6 +46,7 @@ import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
import org.apache.bookkeeper.mledger.impl.PositionImpl;
import org.apache.commons.lang3.tuple.MutablePair;
+import org.apache.pulsar.broker.intercept.BrokerInterceptor;
import org.apache.pulsar.broker.service.BrokerServiceException;
import org.apache.pulsar.broker.service.BrokerServiceException.NotAllowedException;
import org.apache.pulsar.broker.service.BrokerServiceException.ServerMetadataException;
@@ -139,6 +140,11 @@ public class PersistentSubscription implements Subscription {
}
@Override
+ public BrokerInterceptor interceptor() {
+ return topic.getBrokerService().getInterceptor();
+ }
+
+ @Override
public String getName() {
return this.subName;
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/BrokerInterceptorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/BrokerInterceptorTest.java
index 781e85f..4a1504c 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/BrokerInterceptorTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/BrokerInterceptorTest.java
@@ -18,7 +18,11 @@
*/
package org.apache.pulsar.broker.intercept;
+import lombok.Cleanup;
import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.ProducerConsumerBase;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
@@ -35,6 +39,7 @@ import static org.mockito.ArgumentMatchers.same;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
+import static org.testng.Assert.assertEquals;
public class BrokerInterceptorTest extends ProducerConsumerBase {
@@ -108,4 +113,28 @@ public class BrokerInterceptorTest extends ProducerConsumerBase {
// CONNECT and PRODUCER
Assert.assertTrue(((CounterBrokerInterceptor)listener).getCount() >= 2);
}
+
+ @Test
+ public void testBeforeSendMessage() throws PulsarClientException {
+ BrokerInterceptor listener = pulsar.getBrokerInterceptor();
+ Assert.assertTrue(listener instanceof CounterBrokerInterceptor);
+
+ @Cleanup
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic("test-before-send-message")
+ .create();
+
+ Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
+ .topic("test-before-send-message")
+ .subscriptionName("test")
+ .subscribe();
+
+ producer.send("hello world");
+
+ Message<String> msg = consumer.receive();
+
+ assertEquals(msg.getValue(), "hello world");
+
+ Assert.assertTrue(((CounterBrokerInterceptor) listener).getBeforeSendCount() == 1);
+ }
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/CounterBrokerInterceptor.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/CounterBrokerInterceptor.java
index defe386..8f3fa51 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/CounterBrokerInterceptor.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/CounterBrokerInterceptor.java
@@ -19,9 +19,12 @@
package org.apache.pulsar.broker.intercept;
import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.Entry;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.service.ServerCnx;
+import org.apache.pulsar.broker.service.Subscription;
import org.apache.pulsar.common.api.proto.PulsarApi;
+import org.apache.pulsar.common.api.proto.PulsarApi.MessageMetadata;
import org.apache.pulsar.common.intercept.InterceptException;
import javax.servlet.ServletException;
@@ -33,9 +36,20 @@ import java.io.IOException;
@Slf4j
public class CounterBrokerInterceptor implements BrokerInterceptor {
+ int beforeSendCount = 0;
int count = 0;
@Override
+ public void beforeSendMessage(Subscription subscription,
+ Entry entry,
+ long[] ackSet,
+ MessageMetadata msgMetadata) {
+ log.info("Send message to topic {}, subscription {}",
+ subscription.getTopic(), subscription.getName());
+ beforeSendCount++;
+ }
+
+ @Override
public void onPulsarCommand(PulsarApi.BaseCommand command, ServerCnx cnx) throws InterceptException {
log.info("[{}] On [{}] Pulsar command", count, command.getType().name());
count ++;
@@ -71,4 +85,8 @@ public class CounterBrokerInterceptor implements BrokerInterceptor {
public int getCount() {
return count;
}
+
+ public int getBeforeSendCount() {
+ return beforeSendCount;
+ }
}
[pulsar] 05/07: [pulsar-broker] capture stats with precise backlog
(#8928)
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.7
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 40135da18063d59a3474aab7f97168ac7cd3af29
Author: Rajan Dhabalia <rd...@apache.org>
AuthorDate: Mon Dec 14 14:25:11 2020 -0800
[pulsar-broker] capture stats with precise backlog (#8928)
Co-authored-by: Sijie Guo <si...@apache.org>
(cherry picked from commit a7f692f981ddf2f86a69809867797f7269196884)
---
.../org/apache/pulsar/broker/service/persistent/PersistentTopic.java | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index 486d2d7..00dcd81 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -1500,7 +1500,8 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
topicStatsStream.endList();
// Populate subscription specific stats here
- topicStatsStream.writePair("msgBacklog", subscription.getNumberOfEntriesInBacklog(false));
+ topicStatsStream.writePair("msgBacklog",
+ subscription.getNumberOfEntriesInBacklog(true));
topicStatsStream.writePair("msgRateExpired", subscription.getExpiredMessageRate());
topicStatsStream.writePair("msgRateOut", subMsgRateOut);
topicStatsStream.writePair("msgThroughputOut", subMsgThroughputOut);
[pulsar] 03/07: Fix NSQ source META-INF file name and
sourceConfigClass (#8941)
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.7
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 934d3b85c787332c5199cd39a16734f6fc23ea74
Author: Andy Walker <wa...@gmail.com>
AuthorDate: Mon Dec 14 00:38:21 2020 -0500
Fix NSQ source META-INF file name and sourceConfigClass (#8941)
(cherry picked from commit 4df7bb0dcb646d6a8548555ca335d63851663f0b)
---
.../main/resources/META-INF/services/{pulsar-io.yml => pulsar-io.yaml} | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/pulsar-io/nsq/src/main/resources/META-INF/services/pulsar-io.yml b/pulsar-io/nsq/src/main/resources/META-INF/services/pulsar-io.yaml
similarity index 93%
rename from pulsar-io/nsq/src/main/resources/META-INF/services/pulsar-io.yml
rename to pulsar-io/nsq/src/main/resources/META-INF/services/pulsar-io.yaml
index 6f2a347..f1a4470 100644
--- a/pulsar-io/nsq/src/main/resources/META-INF/services/pulsar-io.yml
+++ b/pulsar-io/nsq/src/main/resources/META-INF/services/pulsar-io.yaml
@@ -20,4 +20,4 @@
name: nsq
description: Ingest data from an NSQ topic
sourceClass: org.apache.pulsar.io.nsq.NSQSource
-sourceConfigClass: org.apache.pulsar.io.nsq.NSQConfig
+sourceConfigClass: org.apache.pulsar.io.nsq.NSQSourceConfig
[pulsar] 04/07: Fix subscription dispatch rate does not work after
the topic unload without dispatch rate limit. (#8947)
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.7
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 746854305708bf4d00923307c000c8cd23b79ef2
Author: lipenghui <pe...@apache.org>
AuthorDate: Tue Dec 15 01:51:43 2020 +0800
Fix subscription dispatch rate does not work after the topic unload without dispatch rate limit. (#8947)
### Motivation
Fix subscription dispatch rate does not work after the topic unload without dispatch rate limit.
Currently, if the subscription dispatch rate is present then update the rate limit, the subscription can be affected by the new policy. But if the subscription dispatch rate is absent, update the rate limit, the subscription cannot be affected.
### Verifying this change
New unitest added.
(cherry picked from commit 37e02fa02e0437cbc1254147b04c68ec77a4e76b)
---
.../apache/pulsar/broker/service/Dispatcher.java | 5 ++++
.../PersistentDispatcherMultipleConsumers.java | 15 +++++++++++
.../PersistentDispatcherSingleActiveConsumer.java | 15 +++++++++++
.../broker/service/persistent/PersistentTopic.java | 8 +-----
.../pulsar/broker/admin/TopicPoliciesTest.java | 31 ++++++++++++++++++++++
5 files changed, 67 insertions(+), 7 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Dispatcher.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Dispatcher.java
index 483190f..43b02d1 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Dispatcher.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Dispatcher.java
@@ -26,6 +26,7 @@ import org.apache.bookkeeper.mledger.impl.PositionImpl;
import org.apache.pulsar.broker.service.persistent.DispatchRateLimiter;
import org.apache.pulsar.common.api.proto.PulsarApi.CommandSubscribe.SubType;
import org.apache.pulsar.common.api.proto.PulsarApi.MessageMetadata;
+import org.apache.pulsar.common.policies.data.DispatchRate;
import org.apache.pulsar.common.policies.data.Policies;
public interface Dispatcher {
@@ -92,6 +93,10 @@ public interface Dispatcher {
return Optional.empty();
}
+ default void updateRateLimiter(DispatchRate dispatchRate) {
+
+ }
+
default void initializeDispatchRateLimiterIfNeeded(Optional<Policies> policies) {
//No-op
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
index 265186b..af22fa5 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
@@ -64,6 +64,7 @@ import org.apache.pulsar.client.impl.Backoff;
import org.apache.pulsar.common.api.proto.PulsarApi.CommandSubscribe.SubType;
import org.apache.pulsar.common.api.proto.PulsarApi.MessageMetadata;
import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.policies.data.DispatchRate;
import org.apache.pulsar.common.policies.data.Policies;
import org.apache.pulsar.common.policies.data.TopicPolicies;
import org.apache.pulsar.common.util.Codec;
@@ -754,6 +755,20 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
}
@Override
+ public void updateRateLimiter(DispatchRate dispatchRate) {
+ if (!this.dispatchRateLimiter.isPresent() && dispatchRate != null) {
+ this.dispatchRateLimiter = Optional.of(new DispatchRateLimiter(topic, Type.SUBSCRIPTION));
+ }
+ this.dispatchRateLimiter.ifPresent(limiter -> {
+ if (dispatchRate != null) {
+ this.dispatchRateLimiter.get().updateDispatchRate(dispatchRate);
+ } else {
+ this.dispatchRateLimiter.get().updateDispatchRate();
+ }
+ });
+ }
+
+ @Override
public void initializeDispatchRateLimiterIfNeeded(Optional<Policies> policies) {
if (!dispatchRateLimiter.isPresent() && DispatchRateLimiter
.isDispatchRateNeeded(topic.getBrokerService(), policies, topic.getName(), Type.SUBSCRIPTION)) {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
index 89b1920..a5c8a5f 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
@@ -54,6 +54,7 @@ import org.apache.pulsar.broker.transaction.buffer.exceptions.TransactionNotSeal
import org.apache.pulsar.client.impl.Backoff;
import org.apache.pulsar.common.api.proto.PulsarApi.CommandSubscribe.SubType;
import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.policies.data.DispatchRate;
import org.apache.pulsar.common.policies.data.Policies;
import org.apache.pulsar.common.policies.data.TopicPolicies;
import org.apache.pulsar.common.util.Codec;
@@ -568,6 +569,20 @@ public final class PersistentDispatcherSingleActiveConsumer extends AbstractDisp
}
@Override
+ public void updateRateLimiter(DispatchRate dispatchRate) {
+ if (!this.dispatchRateLimiter.isPresent() && dispatchRate != null) {
+ this.dispatchRateLimiter = Optional.of(new DispatchRateLimiter(topic, Type.SUBSCRIPTION));
+ }
+ this.dispatchRateLimiter.ifPresent(limiter -> {
+ if (dispatchRate != null) {
+ this.dispatchRateLimiter.get().updateDispatchRate(dispatchRate);
+ } else {
+ this.dispatchRateLimiter.get().updateDispatchRate();
+ }
+ });
+ }
+
+ @Override
public void initializeDispatchRateLimiterIfNeeded(Optional<Policies> policies) {
if (!dispatchRateLimiter.isPresent() && DispatchRateLimiter
.isDispatchRateNeeded(topic.getBrokerService(), policies, topic.getName(), Type.SUBSCRIPTION)) {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index 3b91cf80..486d2d7 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -2597,13 +2597,7 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
subscriptions.forEach((subName, sub) -> {
sub.getConsumers().forEach(Consumer::checkPermissions);
Dispatcher dispatcher = sub.getDispatcher();
- if (policies.isSubscriptionDispatchRateSet()) {
- dispatcher.getRateLimiter().ifPresent(rateLimiter ->
- rateLimiter.updateDispatchRate(policies.getSubscriptionDispatchRate()));
- } else {
- dispatcher.getRateLimiter().ifPresent(rateLimiter ->
- rateLimiter.updateDispatchRate());
- }
+ dispatcher.updateRateLimiter(policies.getSubscriptionDispatchRate());
});
if (policies.getPublishRate() != null) {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java
index 8fe2be2..99cb773 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java
@@ -645,6 +645,37 @@ public class TopicPoliciesTest extends MockedPulsarServiceBaseTest {
}
@Test
+ public void testGetSetSubscriptionDispatchRateAfterTopicLoaded() throws Exception {
+ final String topic = testTopic + UUID.randomUUID();
+ admin.topics().createNonPartitionedTopic(topic);
+
+ Awaitility.await().atMost(3, TimeUnit.SECONDS)
+ .until(() -> pulsar.getTopicPoliciesService().cacheIsInitialized(TopicName.get(topic)));
+
+ DispatchRate dispatchRate = new DispatchRate(1000,
+ 1024 * 1024, 1);
+ log.info("Subscription Dispatch Rate: {} will set to the topic: {}", dispatchRate, topic);
+
+ String subscriptionName = "test_subscription_rate";
+ Consumer<byte[]> consumer = pulsarClient.newConsumer().subscriptionName(subscriptionName).topic(topic).subscribe();
+
+ admin.topics().setSubscriptionDispatchRate(topic, dispatchRate);
+ log.info("Subscription dispatch rate set success on topic: {}", topic);
+
+ Awaitility.await().atMost(3, TimeUnit.SECONDS)
+ .untilAsserted(() -> Assert.assertEquals(admin.topics().getSubscriptionDispatchRate(topic), dispatchRate));
+
+ DispatchRateLimiter dispatchRateLimiter = pulsar.getBrokerService().getTopicIfExists(topic)
+ .get().get().getSubscription(subscriptionName).getDispatcher().getRateLimiter().get();
+ Assert.assertNotNull(dispatchRateLimiter);
+ Assert.assertEquals(dispatchRateLimiter.getDispatchRateOnByte(), dispatchRate.dispatchThrottlingRateInByte);
+ Assert.assertEquals(dispatchRateLimiter.getDispatchRateOnMsg(), dispatchRate.dispatchThrottlingRateInMsg);
+
+ consumer.close();
+ admin.topics().delete(topic, true);
+ }
+
+ @Test
public void testRemoveSubscriptionDispatchRate() throws Exception {
final String topic = testTopic + UUID.randomUUID();
admin.topics().createNonPartitionedTopic(topic);
[pulsar] 02/07: Expose non-contiguous deleted messages ranges
stats. (#8936)
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.7
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit c83da2c6cc90cf157506e6fa983b217c4826ebd9
Author: lipenghui <pe...@apache.org>
AuthorDate: Mon Dec 14 12:39:23 2020 +0800
Expose non-contiguous deleted messages ranges stats. (#8936)
Expose non-contiguous deleted messages ranges stats.
New test added.
(cherry picked from commit 1654e8872d389ec0a9e9dd22203267fa6333dead)
---
.../apache/bookkeeper/mledger/ManagedCursor.java | 5 +++
.../bookkeeper/mledger/impl/ManagedCursorImpl.java | 13 ++++++-
.../mledger/impl/ManagedCursorContainerTest.java | 5 +++
.../service/persistent/PersistentSubscription.java | 3 ++
.../broker/service/persistent/PersistentTopic.java | 3 ++
.../pulsar/broker/stats/SubscriptionStatsTest.java | 43 ++++++++++++++++++++++
.../common/policies/data/SubscriptionStats.java | 11 ++++++
.../pulsar/common/policies/data/TopicStats.java | 10 +++++
8 files changed, 92 insertions(+), 1 deletion(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java
index 94642aa..7a7ac25 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java
@@ -609,6 +609,11 @@ public interface ManagedCursor {
int getTotalNonContiguousDeletedMessagesRange();
/**
+ * Returns the serialized size of mark-Delete ranges.
+ */
+ int getNonContiguousDeletedMessagesRangeSerializedSize();
+
+ /**
* Returns the estimated size of the unacknowledged backlog for this cursor
*
* @return the estimated size from the mark delete position of the cursor
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 1b7d22e..e9b96be 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
@@ -54,6 +54,7 @@ import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicLongFieldUpdater;
@@ -185,6 +186,7 @@ public class ManagedCursorImpl implements ManagedCursor {
private long entriesReadCount;
private long entriesReadSize;
+ private int individualDeletedMessagesSerializedSize;
class MarkDeleteEntry {
final PositionImpl newPosition;
@@ -827,6 +829,11 @@ public class ManagedCursorImpl implements ManagedCursor {
}
@Override
+ public int getNonContiguousDeletedMessagesRangeSerializedSize() {
+ return this.individualDeletedMessagesSerializedSize;
+ }
+
+ @Override
public long getEstimatedSizeSinceMarkDeletePosition() {
return ledger.estimateBacklogFromPosition(markDeletePosition);
}
@@ -2421,6 +2428,7 @@ public class ManagedCursorImpl implements ManagedCursor {
MLDataFormats.NestedPositionInfo.Builder nestedPositionBuilder = MLDataFormats.NestedPositionInfo
.newBuilder();
MLDataFormats.MessageRange.Builder messageRangeBuilder = MLDataFormats.MessageRange.newBuilder();
+ AtomicInteger acksSerializedSize = new AtomicInteger(0);
List<MessageRange> rangeList = Lists.newArrayList();
individualDeletedMessages.forEach((positionRange) -> {
PositionImpl p = positionRange.lowerEndpoint();
@@ -2431,9 +2439,12 @@ public class ManagedCursorImpl implements ManagedCursor {
nestedPositionBuilder.setLedgerId(p.getLedgerId());
nestedPositionBuilder.setEntryId(p.getEntryId());
messageRangeBuilder.setUpperEndpoint(nestedPositionBuilder.build());
- rangeList.add(messageRangeBuilder.build());
+ MessageRange messageRange = messageRangeBuilder.build();
+ acksSerializedSize.addAndGet(messageRange.getSerializedSize());
+ rangeList.add(messageRange);
return rangeList.size() <= config.getMaxUnackedRangesToPersist();
});
+ this.individualDeletedMessagesSerializedSize = acksSerializedSize.get();
return rangeList;
} finally {
lock.readLock().unlock();
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorContainerTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorContainerTest.java
index 32e55de..5733671 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorContainerTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorContainerTest.java
@@ -298,6 +298,11 @@ public class ManagedCursorContainerTest {
}
@Override
+ public int getNonContiguousDeletedMessagesRangeSerializedSize() {
+ return 0;
+ }
+
+ @Override
public long getEstimatedSizeSinceMarkDeletePosition() {
return 0L;
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
index 0a578d0..57b5bbc 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
@@ -941,6 +941,9 @@ public class PersistentSubscription implements Subscription {
});
}
}
+ subStats.nonContiguousDeletedMessagesRanges = cursor.getTotalNonContiguousDeletedMessagesRange();
+ subStats.nonContiguousDeletedMessagesRangesSerializedSize =
+ cursor.getNonContiguousDeletedMessagesRangeSerializedSize();
return subStats;
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index 3de20c5..3b91cf80 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -1609,6 +1609,9 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
stats.bytesOutCounter += subStats.bytesOutCounter;
stats.msgOutCounter += subStats.msgOutCounter;
stats.subscriptions.put(name, subStats);
+ stats.nonContiguousDeletedMessagesRanges += subStats.nonContiguousDeletedMessagesRanges;
+ stats.nonContiguousDeletedMessagesRangesSerializedSize +=
+ subStats.nonContiguousDeletedMessagesRangesSerializedSize;
});
replicators.forEach((cluster, replicator) -> {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/SubscriptionStatsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/SubscriptionStatsTest.java
index e441e19..25b0861 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/SubscriptionStatsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/SubscriptionStatsTest.java
@@ -22,17 +22,20 @@ import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.ProducerConsumerBase;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.common.policies.data.TopicStats;
+import org.awaitility.Awaitility;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
import java.util.UUID;
+import java.util.concurrent.TimeUnit;
@Slf4j
public class SubscriptionStatsTest extends ProducerConsumerBase {
@@ -91,4 +94,44 @@ public class SubscriptionStatsTest extends ProducerConsumerBase {
consumer2.close();
producer.close();
}
+
+ @Test
+ public void testNonContiguousDeletedMessagesRanges() throws Exception {
+ final String topicName = "persistent://my-property/my-ns/testNonContiguousDeletedMessagesRanges-"
+ + UUID.randomUUID().toString();
+ final String subName = "my-sub";
+
+ @Cleanup
+ Consumer<byte[]> consumer = pulsarClient.newConsumer()
+ .topic(topicName)
+ .subscriptionName(subName)
+ .subscribe();
+
+ @Cleanup
+ Producer<byte[]> producer = pulsarClient.newProducer()
+ .topic(topicName)
+ .create();
+
+ final int messages = 100;
+ for (int i = 0; i < messages; i++) {
+ producer.send(String.valueOf(i).getBytes());
+ }
+
+ for (int i = 0; i < messages; i++) {
+ Message<byte[]> received = consumer.receive();
+ if (i != 50) {
+ consumer.acknowledge(received);
+ }
+ }
+
+ Awaitility.await().atMost(3, TimeUnit.SECONDS).untilAsserted(() -> {
+ TopicStats stats = admin.topics().getStats(topicName);
+ Assert.assertEquals(stats.nonContiguousDeletedMessagesRanges, 1);
+ Assert.assertEquals(stats.subscriptions.size(), 1);
+ Assert.assertEquals(stats.subscriptions.get(subName).nonContiguousDeletedMessagesRanges, 1);
+ Assert.assertTrue(stats.nonContiguousDeletedMessagesRangesSerializedSize > 0);
+ Assert.assertTrue(stats.subscriptions.get(subName)
+ .nonContiguousDeletedMessagesRangesSerializedSize > 0);
+ });
+ }
}
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/SubscriptionStats.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/SubscriptionStats.java
index e5e6feb..c3a09ab 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/SubscriptionStats.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/SubscriptionStats.java
@@ -98,6 +98,12 @@ public class SubscriptionStats {
/** This is for Key_Shared subscription to get the recentJoinedConsumers in the Key_Shared subscription. */
public Map<String, String> consumersAfterMarkDeletePosition;
+ /** The number of non-contiguous deleted messages ranges. */
+ public int nonContiguousDeletedMessagesRanges;
+
+ /** The serialized size of non-contiguous deleted messages ranges. */
+ public int nonContiguousDeletedMessagesRangesSerializedSize;
+
public SubscriptionStats() {
this.consumers = Lists.newArrayList();
this.consumersAfterMarkDeletePosition = new LinkedHashMap<>();
@@ -115,6 +121,9 @@ public class SubscriptionStats {
msgRateExpired = 0;
lastExpireTimestamp = 0L;
consumers.clear();
+ consumersAfterMarkDeletePosition.clear();
+ nonContiguousDeletedMessagesRanges = 0;
+ nonContiguousDeletedMessagesRangesSerializedSize = 0;
}
// if the stats are added for the 1st time, we will need to make a copy of these stats and add it to the current
@@ -143,6 +152,8 @@ public class SubscriptionStats {
}
}
this.consumersAfterMarkDeletePosition.putAll(stats.consumersAfterMarkDeletePosition);
+ this.nonContiguousDeletedMessagesRanges += stats.nonContiguousDeletedMessagesRanges;
+ this.nonContiguousDeletedMessagesRangesSerializedSize += stats.nonContiguousDeletedMessagesRangesSerializedSize;
return this;
}
}
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/TopicStats.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/TopicStats.java
index 8d560a9..bc6716b 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/TopicStats.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/TopicStats.java
@@ -78,6 +78,12 @@ public class TopicStats {
public String deduplicationStatus;
+ /** The number of non-contiguous deleted messages ranges. */
+ public int nonContiguousDeletedMessagesRanges;
+
+ /** The serialized size of non-contiguous deleted messages ranges. */
+ public int nonContiguousDeletedMessagesRangesSerializedSize;
+
public TopicStats() {
this.publishers = Lists.newArrayList();
this.subscriptions = Maps.newHashMap();
@@ -101,6 +107,8 @@ public class TopicStats {
this.subscriptions.clear();
this.replication.clear();
this.deduplicationStatus = null;
+ this.nonContiguousDeletedMessagesRanges = 0;
+ this.nonContiguousDeletedMessagesRangesSerializedSize = 0;
}
// if the stats are added for the 1st time, we will need to make a copy of these stats and add it to the current
@@ -120,6 +128,8 @@ public class TopicStats {
this.averageMsgSize = newAverageMsgSize;
this.storageSize += stats.storageSize;
this.backlogSize += stats.backlogSize;
+ this.nonContiguousDeletedMessagesRanges += stats.nonContiguousDeletedMessagesRanges;
+ this.nonContiguousDeletedMessagesRangesSerializedSize += stats.nonContiguousDeletedMessagesRangesSerializedSize;
if (this.publishers.size() != stats.publishers.size()) {
for (int i = 0; i < stats.publishers.size(); i++) {
PublisherStats publisherStats = new PublisherStats();
[pulsar] 07/07: [C++] Add consumer's configs for reader (#8905)
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.7
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 4da0bafae1bb94a9856f78243db20ede1b60f85c
Author: Yunze Xu <xy...@163.com>
AuthorDate: Mon Dec 14 18:27:09 2020 -0600
[C++] Add consumer's configs for reader (#8905)
### Motivation
C++ reader cannot set some consumer's configs. Like `CryptoReader`, if the config is not supported for reader, then reader cannot decrypt encrypted messages. Also, reader cannot set other configs like `AckGroupingTimeMs` so that reader cannot control the behavior of `AckGroupingTracker`.
### Modifications
- Add other consumer's configs for reader, except:
- NegativeAckRedeliveryDelayMs: reader has no chance to call `negativeAcknowledge`
- BrokerConsumerStatsCacheTimeInMs: reader has no chance to call `getBrokerConsumerStatsAsync`
- KeySharedPolicy and ConsumerType: reader only supports Exclusive subscription now
- MaxTotalReceiverQueueSizeAcrossPartitions: reader doesn't support partitioned topics now
- PatternAutoDiscoveryPeriod: reader doesn't support regex subscription now
- Add some fields to get the consumer config before reader's internal consumer is created, then add some unit tests for the consumer config.
(cherry picked from commit a3ac12e8cd255bbacd308f6e2c3186e2d3f98850)
---
.../include/pulsar/ReaderConfiguration.h | 93 ++++++++++++++
pulsar-client-cpp/lib/ReaderConfiguration.cc | 76 +++++++++++
pulsar-client-cpp/lib/ReaderConfigurationImpl.h | 20 +--
pulsar-client-cpp/lib/ReaderImpl.cc | 18 +++
pulsar-client-cpp/lib/ReaderImpl.h | 10 +-
pulsar-client-cpp/tests/ReaderConfigurationTest.cc | 141 +++++++++++++++++++++
6 files changed, 347 insertions(+), 11 deletions(-)
diff --git a/pulsar-client-cpp/include/pulsar/ReaderConfiguration.h b/pulsar-client-cpp/include/pulsar/ReaderConfiguration.h
index 25b0ba3..96b36b6 100644
--- a/pulsar-client-cpp/include/pulsar/ReaderConfiguration.h
+++ b/pulsar-client-cpp/include/pulsar/ReaderConfiguration.h
@@ -25,6 +25,8 @@
#include <pulsar/Result.h>
#include <pulsar/Message.h>
#include <pulsar/Schema.h>
+#include <pulsar/CryptoKeyReader.h>
+#include <pulsar/ConsumerCryptoFailureAction.h>
namespace pulsar {
@@ -112,6 +114,97 @@ class PULSAR_PUBLIC ReaderConfiguration {
void setInternalSubscriptionName(std::string internalSubscriptionName);
const std::string& getInternalSubscriptionName() const;
+ /**
+ * Set the timeout in milliseconds for unacknowledged messages, the timeout needs to be greater than
+ * 10 seconds. An Exception is thrown if the given value is less than 10000 (10 seconds).
+ * If a successful acknowledgement is not sent within the timeout all the unacknowledged messages are
+ * redelivered.
+ * @param timeout in milliseconds
+ */
+ void setUnAckedMessagesTimeoutMs(const uint64_t milliSeconds);
+
+ /**
+ * @return the configured timeout in milliseconds for unacked messages.
+ */
+ long getUnAckedMessagesTimeoutMs() const;
+
+ void setTickDurationInMs(const uint64_t milliSeconds);
+
+ long getTickDurationInMs() const;
+
+ /**
+ * Set time window in milliseconds for grouping message ACK requests. An ACK request is not sent
+ * to broker until the time window reaches its end, or the number of grouped messages reaches
+ * limit. Default is 100 milliseconds. If it's set to a non-positive value, ACK requests will be
+ * directly sent to broker without grouping.
+ *
+ * @param ackGroupMillis time of ACK grouping window in milliseconds.
+ */
+ void setAckGroupingTimeMs(long ackGroupingMillis);
+
+ /**
+ * Get grouping time window in milliseconds.
+ *
+ * @return grouping time window in milliseconds.
+ */
+ long getAckGroupingTimeMs() const;
+
+ /**
+ * Set max number of grouped messages within one grouping time window. If it's set to a
+ * non-positive value, number of grouped messages is not limited. Default is 1000.
+ *
+ * @param maxGroupingSize max number of grouped messages with in one grouping time window.
+ */
+ void setAckGroupingMaxSize(long maxGroupingSize);
+
+ /**
+ * Get max number of grouped messages within one grouping time window.
+ *
+ * @return max number of grouped messages within one grouping time window.
+ */
+ long getAckGroupingMaxSize() const;
+
+ bool isEncryptionEnabled() const;
+ const CryptoKeyReaderPtr getCryptoKeyReader() const;
+ ReaderConfiguration& setCryptoKeyReader(CryptoKeyReaderPtr cryptoKeyReader);
+
+ ConsumerCryptoFailureAction getCryptoFailureAction() const;
+ ReaderConfiguration& setCryptoFailureAction(ConsumerCryptoFailureAction action);
+
+ /**
+ * Check whether the message has a specific property attached.
+ *
+ * @param name the name of the property to check
+ * @return true if the message has the specified property
+ * @return false if the property is not defined
+ */
+ bool hasProperty(const std::string& name) const;
+
+ /**
+ * Get the value of a specific property
+ *
+ * @param name the name of the property
+ * @return the value of the property or null if the property was not defined
+ */
+ const std::string& getProperty(const std::string& name) const;
+
+ /**
+ * Get all the properties attached to this producer.
+ */
+ std::map<std::string, std::string>& getProperties() const;
+
+ /**
+ * Sets a new property on a message.
+ * @param name the name of the property
+ * @param value the associated value
+ */
+ ReaderConfiguration& setProperty(const std::string& name, const std::string& value);
+
+ /**
+ * Add all the properties in the provided map
+ */
+ ReaderConfiguration& setProperties(const std::map<std::string, std::string>& properties);
+
private:
std::shared_ptr<ReaderConfigurationImpl> impl_;
};
diff --git a/pulsar-client-cpp/lib/ReaderConfiguration.cc b/pulsar-client-cpp/lib/ReaderConfiguration.cc
index eb18d11..0dfdbed 100644
--- a/pulsar-client-cpp/lib/ReaderConfiguration.cc
+++ b/pulsar-client-cpp/lib/ReaderConfiguration.cc
@@ -20,6 +20,8 @@
namespace pulsar {
+const static std::string emptyString;
+
ReaderConfiguration::ReaderConfiguration() : impl_(std::make_shared<ReaderConfigurationImpl>()) {}
ReaderConfiguration::~ReaderConfiguration() {}
@@ -76,4 +78,78 @@ const std::string& ReaderConfiguration::getInternalSubscriptionName() const {
return impl_->internalSubscriptionName;
}
+void ReaderConfiguration::setUnAckedMessagesTimeoutMs(const uint64_t milliSeconds) {
+ impl_->unAckedMessagesTimeoutMs = milliSeconds;
+}
+
+long ReaderConfiguration::getUnAckedMessagesTimeoutMs() const { return impl_->unAckedMessagesTimeoutMs; }
+
+void ReaderConfiguration::setTickDurationInMs(const uint64_t milliSeconds) {
+ impl_->tickDurationInMs = milliSeconds;
+}
+
+long ReaderConfiguration::getTickDurationInMs() const { return impl_->tickDurationInMs; }
+
+void ReaderConfiguration::setAckGroupingTimeMs(long ackGroupingMillis) {
+ impl_->ackGroupingTimeMs = ackGroupingMillis;
+}
+
+long ReaderConfiguration::getAckGroupingTimeMs() const { return impl_->ackGroupingTimeMs; }
+
+void ReaderConfiguration::setAckGroupingMaxSize(long maxGroupingSize) {
+ impl_->ackGroupingMaxSize = maxGroupingSize;
+}
+
+long ReaderConfiguration::getAckGroupingMaxSize() const { return impl_->ackGroupingMaxSize; }
+
+bool ReaderConfiguration::isEncryptionEnabled() const { return impl_->cryptoKeyReader != nullptr; }
+
+const CryptoKeyReaderPtr ReaderConfiguration::getCryptoKeyReader() const { return impl_->cryptoKeyReader; }
+
+ReaderConfiguration& ReaderConfiguration::setCryptoKeyReader(CryptoKeyReaderPtr cryptoKeyReader) {
+ impl_->cryptoKeyReader = cryptoKeyReader;
+ return *this;
+}
+
+ConsumerCryptoFailureAction ReaderConfiguration::getCryptoFailureAction() const {
+ return impl_->cryptoFailureAction;
+}
+
+ReaderConfiguration& ReaderConfiguration::setCryptoFailureAction(ConsumerCryptoFailureAction action) {
+ impl_->cryptoFailureAction = action;
+ return *this;
+}
+
+bool ReaderConfiguration::hasProperty(const std::string& name) const {
+ const auto& properties = impl_->properties;
+ return properties.find(name) != properties.cend();
+}
+
+const std::string& ReaderConfiguration::getProperty(const std::string& name) const {
+ const auto& properties = impl_->properties;
+ const auto it = properties.find(name);
+ return (it != properties.cend()) ? (it->second) : emptyString;
+}
+
+std::map<std::string, std::string>& ReaderConfiguration::getProperties() const { return impl_->properties; }
+
+ReaderConfiguration& ReaderConfiguration::setProperty(const std::string& name, const std::string& value) {
+ auto& properties = impl_->properties;
+ auto it = properties.find(name);
+ if (it != properties.end()) {
+ it->second = value;
+ } else {
+ properties.emplace(name, value);
+ }
+ return *this;
+}
+
+ReaderConfiguration& ReaderConfiguration::setProperties(
+ const std::map<std::string, std::string>& properties) {
+ for (const auto& kv : properties) {
+ setProperty(kv.first, kv.second);
+ }
+ return *this;
+}
+
} // namespace pulsar
diff --git a/pulsar-client-cpp/lib/ReaderConfigurationImpl.h b/pulsar-client-cpp/lib/ReaderConfigurationImpl.h
index 3b5cc8a..6f38c29 100644
--- a/pulsar-client-cpp/lib/ReaderConfigurationImpl.h
+++ b/pulsar-client-cpp/lib/ReaderConfigurationImpl.h
@@ -25,19 +25,19 @@ namespace pulsar {
struct ReaderConfigurationImpl {
SchemaInfo schemaInfo;
ReaderListener readerListener;
- bool hasReaderListener;
- int receiverQueueSize;
+ bool hasReaderListener{false};
+ int receiverQueueSize{1000};
std::string readerName;
std::string subscriptionRolePrefix;
- bool readCompacted;
+ bool readCompacted{false};
std::string internalSubscriptionName;
- ReaderConfigurationImpl()
- : schemaInfo(),
- hasReaderListener(false),
- receiverQueueSize(1000),
- readerName(),
- subscriptionRolePrefix(),
- readCompacted(false) {}
+ long unAckedMessagesTimeoutMs{0};
+ long tickDurationInMs{1000};
+ long ackGroupingTimeMs{100};
+ long ackGroupingMaxSize{1000};
+ CryptoKeyReaderPtr cryptoKeyReader;
+ ConsumerCryptoFailureAction cryptoFailureAction;
+ std::map<std::string, std::string> properties;
};
} // namespace pulsar
#endif /* LIB_READERCONFIGURATIONIMPL_H_ */
diff --git a/pulsar-client-cpp/lib/ReaderImpl.cc b/pulsar-client-cpp/lib/ReaderImpl.cc
index 2ea3430..bcf707d 100644
--- a/pulsar-client-cpp/lib/ReaderImpl.cc
+++ b/pulsar-client-cpp/lib/ReaderImpl.cc
@@ -23,6 +23,12 @@
namespace pulsar {
+namespace test {
+std::mutex readerConfigTestMutex;
+std::atomic_bool readerConfigTestEnabled{false};
+ConsumerConfiguration consumerConfigOfReader;
+} // namespace test
+
static ResultCallback emptyCallback;
ReaderImpl::ReaderImpl(const ClientImplPtr client, const std::string& topic, const ReaderConfiguration& conf,
@@ -35,6 +41,13 @@ void ReaderImpl::start(const MessageId& startMessageId) {
consumerConf.setReceiverQueueSize(readerConf_.getReceiverQueueSize());
consumerConf.setReadCompacted(readerConf_.isReadCompacted());
consumerConf.setSchema(readerConf_.getSchema());
+ consumerConf.setUnAckedMessagesTimeoutMs(readerConf_.getUnAckedMessagesTimeoutMs());
+ consumerConf.setTickDurationInMs(readerConf_.getTickDurationInMs());
+ consumerConf.setAckGroupingTimeMs(readerConf_.getAckGroupingTimeMs());
+ consumerConf.setAckGroupingMaxSize(readerConf_.getAckGroupingMaxSize());
+ consumerConf.setCryptoKeyReader(readerConf_.getCryptoKeyReader());
+ consumerConf.setCryptoFailureAction(readerConf_.getCryptoFailureAction());
+ consumerConf.setProperties(readerConf_.getProperties());
if (readerConf_.getReaderName().length() > 0) {
consumerConf.setConsumerName(readerConf_.getReaderName());
@@ -57,6 +70,11 @@ void ReaderImpl::start(const MessageId& startMessageId) {
}
}
+ // get the consumer's configuration before created
+ if (test::readerConfigTestEnabled) {
+ test::consumerConfigOfReader = consumerConf.clone();
+ }
+
consumer_ = std::make_shared<ConsumerImpl>(
client_.lock(), topic_, subscription, consumerConf, ExecutorServicePtr(), NonPartitioned,
Commands::SubscriptionModeNonDurable, Optional<MessageId>::of(startMessageId));
diff --git a/pulsar-client-cpp/lib/ReaderImpl.h b/pulsar-client-cpp/lib/ReaderImpl.h
index 4069247..f1ad387 100644
--- a/pulsar-client-cpp/lib/ReaderImpl.h
+++ b/pulsar-client-cpp/lib/ReaderImpl.h
@@ -29,7 +29,15 @@ class ReaderImpl;
typedef std::shared_ptr<ReaderImpl> ReaderImplPtr;
typedef std::weak_ptr<ReaderImpl> ReaderImplWeakPtr;
-class ReaderImpl : public std::enable_shared_from_this<ReaderImpl> {
+namespace test {
+
+extern std::mutex readerConfigTestMutex PULSAR_PUBLIC;
+extern std::atomic_bool readerConfigTestEnabled PULSAR_PUBLIC;
+extern ConsumerConfiguration consumerConfigOfReader PULSAR_PUBLIC;
+
+} // namespace test
+
+class PULSAR_PUBLIC ReaderImpl : public std::enable_shared_from_this<ReaderImpl> {
public:
ReaderImpl(const ClientImplPtr client, const std::string& topic, const ReaderConfiguration& conf,
const ExecutorServicePtr listenerExecutor, ReaderCallback readerCreatedCallback);
diff --git a/pulsar-client-cpp/tests/ReaderConfigurationTest.cc b/pulsar-client-cpp/tests/ReaderConfigurationTest.cc
new file mode 100644
index 0000000..6af4b4a
--- /dev/null
+++ b/pulsar-client-cpp/tests/ReaderConfigurationTest.cc
@@ -0,0 +1,141 @@
+/**
+ * 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.
+ */
+/**
+ * This test only tests the ConsumerConfiguration used for the Reader's internal consumer.
+ * Because the ReaderConfiguration for Reader itself is meaningless.
+ */
+#include <gtest/gtest.h>
+#include <pulsar/Client.h>
+#include <lib/LogUtils.h>
+#include <lib/ReaderImpl.h>
+
+DECLARE_LOG_OBJECT()
+
+using namespace pulsar;
+
+static const std::string lookupUrl = "pulsar://localhost:6650";
+
+class NoOpsCryptoKeyReader : public CryptoKeyReader {
+ public:
+ Result getPublicKey(const std::string& keyName, std::map<std::string, std::string>& metadata,
+ EncryptionKeyInfo& encKeyInfo) const override {
+ return ResultOk;
+ }
+
+ Result getPrivateKey(const std::string& keyName, std::map<std::string, std::string>& metadata,
+ EncryptionKeyInfo& encKeyInfo) const override {
+ return ResultOk;
+ }
+};
+
+TEST(ReaderConfigurationTest, testDefaultConfig) {
+ const std::string topic = "ReaderConfigurationTest-default-config";
+ Client client(lookupUrl);
+ ReaderConfiguration readerConf;
+ Reader reader;
+
+ std::unique_lock<std::mutex> lock(test::readerConfigTestMutex);
+ test::readerConfigTestEnabled = true;
+ ASSERT_EQ(ResultOk, client.createReader(topic, MessageId::earliest(), readerConf, reader));
+ const auto consumerConf = test::consumerConfigOfReader.clone();
+ test::readerConfigTestEnabled = false;
+ lock.unlock();
+
+ ASSERT_EQ(consumerConf.getConsumerType(), ConsumerExclusive);
+ ASSERT_EQ(consumerConf.getReceiverQueueSize(), 1000);
+ ASSERT_EQ(consumerConf.isReadCompacted(), false);
+ ASSERT_EQ(consumerConf.getSchema().getName(), "BYTES");
+ ASSERT_EQ(consumerConf.getUnAckedMessagesTimeoutMs(), 0);
+ ASSERT_EQ(consumerConf.getTickDurationInMs(), 1000);
+ ASSERT_EQ(consumerConf.getAckGroupingTimeMs(), 100);
+ ASSERT_EQ(consumerConf.getAckGroupingMaxSize(), 1000);
+ ASSERT_EQ(consumerConf.getCryptoKeyReader().get(), nullptr);
+ ASSERT_EQ(consumerConf.getCryptoFailureAction(), ConsumerCryptoFailureAction::FAIL);
+ ASSERT_TRUE(consumerConf.getProperties().empty());
+ ASSERT_TRUE(consumerConf.getConsumerName().empty());
+ ASSERT_FALSE(consumerConf.hasMessageListener());
+
+ client.close();
+}
+
+TEST(ReaderConfigurationTest, testCustomConfig) {
+ const std::string exampleSchema =
+ "{\"type\":\"record\",\"name\":\"Example\",\"namespace\":\"test\","
+ "\"fields\":[{\"name\":\"a\",\"type\":\"int\"},{\"name\":\"b\",\"type\":\"int\"}]}";
+
+ const std::string topic = "ReaderConfigurationTest-custom-config";
+ Client client(lookupUrl);
+
+ const SchemaInfo schema(AVRO, "Avro", exampleSchema, StringMap{{"schema-key", "schema-value"}});
+
+ ProducerConfiguration producerConf;
+ producerConf.setSchema(schema);
+ Producer producer;
+ ASSERT_EQ(ResultOk, client.createProducer(topic, producerConf, producer));
+ ASSERT_FALSE(producer.getSchemaVersion().empty());
+
+ ReaderConfiguration readerConf;
+ readerConf.setSchema(schema);
+ readerConf.setReaderListener([](Reader, const Message&) {});
+ readerConf.setReceiverQueueSize(2000);
+ readerConf.setReaderName("my-reader");
+ readerConf.setReadCompacted(true);
+ readerConf.setUnAckedMessagesTimeoutMs(11000);
+ readerConf.setTickDurationInMs(2000);
+ readerConf.setAckGroupingTimeMs(0);
+ readerConf.setAckGroupingMaxSize(4096);
+ const auto cryptoReader = std::make_shared<NoOpsCryptoKeyReader>();
+ readerConf.setCryptoKeyReader(cryptoReader);
+ readerConf.setCryptoFailureAction(ConsumerCryptoFailureAction::DISCARD);
+ const std::map<std::string, std::string> properties{{"key-1", "value-1"}, {"key-2", "value-2"}};
+ readerConf.setProperties(properties);
+
+ Reader reader;
+ std::unique_lock<std::mutex> lock(test::readerConfigTestMutex);
+ test::readerConfigTestEnabled = true;
+ ASSERT_EQ(ResultOk, client.createReader(topic, MessageId::earliest(), readerConf, reader));
+ const auto consumerConf = test::consumerConfigOfReader.clone();
+ test::readerConfigTestEnabled = false;
+ lock.unlock();
+
+ ASSERT_EQ(consumerConf.getSchema().getName(), schema.getName());
+ ASSERT_EQ(consumerConf.getSchema().getSchemaType(), schema.getSchemaType());
+ ASSERT_EQ(consumerConf.getSchema().getSchema(), schema.getSchema());
+ ASSERT_EQ(consumerConf.getSchema().getProperties(), schema.getProperties());
+
+ ASSERT_EQ(consumerConf.getConsumerType(), ConsumerExclusive);
+ ASSERT_TRUE(consumerConf.hasMessageListener());
+ ASSERT_EQ(consumerConf.getReceiverQueueSize(), 2000);
+ ASSERT_EQ(consumerConf.getConsumerName(), "my-reader");
+ ASSERT_EQ(consumerConf.isReadCompacted(), true);
+ ASSERT_EQ(consumerConf.getUnAckedMessagesTimeoutMs(), 11000);
+ ASSERT_EQ(consumerConf.getTickDurationInMs(), 2000);
+ ASSERT_EQ(consumerConf.getAckGroupingTimeMs(), 0);
+ ASSERT_EQ(consumerConf.getAckGroupingMaxSize(), 4096);
+ ASSERT_EQ(consumerConf.getCryptoKeyReader(), cryptoReader);
+ ASSERT_EQ(consumerConf.getCryptoFailureAction(), ConsumerCryptoFailureAction::DISCARD);
+ ASSERT_EQ(consumerConf.getProperties(), properties);
+ ASSERT_TRUE(consumerConf.hasProperty("key-1"));
+ ASSERT_EQ(consumerConf.getProperty("key-1"), "value-1");
+ ASSERT_TRUE(consumerConf.hasProperty("key-2"));
+ ASSERT_EQ(consumerConf.getProperty("key-2"), "value-2");
+ ASSERT_FALSE(consumerConf.hasProperty("key-3"));
+
+ client.close();
+}
[pulsar] 01/07: [ISSUE 8832] Avro custom schema not working in
consumer (#8939)
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.7
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 5a13be86a3a41532867addf6a89a498620aa3674
Author: Weijie Guo <re...@163.com>
AuthorDate: Mon Dec 14 12:22:47 2020 +0800
[ISSUE 8832] Avro custom schema not working in consumer (#8939)
Fix #8832
### Modifications
Make AvroSchema's clone method consider custom reader and writer.
(cherry picked from commit 3da236503b71131023f5de9928ed499c7d1201ea)
---
.../client/impl/BrokerClientIntegrationTest.java | 39 +++++++++++++++++++++-
.../pulsar/client/impl/schema/AvroSchema.java | 6 +++-
2 files changed, 43 insertions(+), 2 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java
index 9041134..2838293 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java
@@ -36,6 +36,7 @@ import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
import java.lang.reflect.Field;
+import java.nio.charset.StandardCharsets;
import java.security.GeneralSecurityException;
import java.util.ArrayList;
import java.util.List;
@@ -858,8 +859,42 @@ public class BrokerClientIntegrationTest extends ProducerConsumerBase {
consumer.close();
}
+
+ @Test
+ public void testAvroSchemaProducerConsumerWithSpecifiedReaderAndWriter() throws PulsarClientException {
+ final String topicName = "persistent://my-property/my-ns/my-topic1";
+ TestMessageObject object = new TestMessageObject();
+ SchemaReader<TestMessageObject> reader = Mockito.mock(SchemaReader.class);
+ SchemaWriter<TestMessageObject> writer = Mockito.mock(SchemaWriter.class);
+ Mockito.when(reader.read(Mockito.any(byte[].class), Mockito.any(byte[].class))).thenReturn(object);
+ Mockito.when(writer.write(Mockito.any(TestMessageObject.class))).thenReturn("fake data".getBytes(StandardCharsets.UTF_8));
+ SchemaDefinition<TestMessageObject> schemaDefinition = new SchemaDefinitionBuilderImpl<TestMessageObject>()
+ .withPojo(TestMessageObject.class)
+ .withSchemaReader(reader)
+ .withSchemaWriter(writer)
+ .build();
+ Schema<TestMessageObject> schema = Schema.AVRO(schemaDefinition);
+ PulsarClient client = PulsarClient.builder()
+ .serviceUrl(lookupUrl.toString())
+ .build();
+
+ try(Producer<TestMessageObject> producer = client.newProducer(schema).topic(topicName).create();
+ Consumer<TestMessageObject> consumer =
+ client.newConsumer(schema).topic(topicName).subscriptionName("my-subscriber-name").subscribe()) {
+ assertNotNull(producer);
+ assertNotNull(consumer);
+ producer.newMessage().value(object).send();
+ TestMessageObject testObject = consumer.receive().getValue();
+ Assert.assertEquals(object.getValue(), testObject.getValue());
+ Mockito.verify(writer, Mockito.times(1)).write(Mockito.any());
+ Mockito.verify(reader, Mockito.times(1)).read(Mockito.any(byte[].class), Mockito.any(byte[].class));
+ } finally {
+ client.close();
+ }
+ }
+
@Test
- public void testProducerConsumerWithSpecifiedReaderAndWriter() throws PulsarClientException {
+ public void testJsonSchemaProducerConsumerWithSpecifiedReaderAndWriter() throws PulsarClientException {
final String topicName = "persistent://my-property/my-ns/my-topic1";
ObjectMapper mapper = new ObjectMapper();
SchemaReader<TestMessageObject> reader = Mockito.spy(new JacksonJsonReader<>(mapper, TestMessageObject.class));
@@ -889,6 +924,8 @@ public class BrokerClientIntegrationTest extends ProducerConsumerBase {
Mockito.verify(writer, Mockito.times(1)).write(Mockito.any());
Mockito.verify(reader, Mockito.times(1)).read(Mockito.any(byte[].class));
+ } finally {
+ client.close();
}
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/AvroSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/AvroSchema.java
index 46227b8..d5a99f8 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/AvroSchema.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/AvroSchema.java
@@ -46,7 +46,7 @@ import static org.apache.pulsar.client.impl.schema.util.SchemaUtil.parseSchemaIn
@Slf4j
public class AvroSchema<T> extends AvroBaseStructSchema<T> {
private static final Logger LOG = LoggerFactory.getLogger(AvroSchema.class);
-
+ private boolean isCustomReaderAndWriter;
private ClassLoader pojoClassLoader;
private AvroSchema(SchemaInfo schemaInfo, ClassLoader pojoClassLoader) {
@@ -61,6 +61,7 @@ public class AvroSchema<T> extends AvroBaseStructSchema<T> {
super(schemaInfo);
setReader(reader);
setWriter(writer);
+ isCustomReaderAndWriter = true;
}
@Override
@@ -70,6 +71,9 @@ public class AvroSchema<T> extends AvroBaseStructSchema<T> {
@Override
public Schema<T> clone() {
+ if (isCustomReaderAndWriter) {
+ return new AvroSchema<>(reader, writer, schemaInfo);
+ }
Schema<T> schema = new AvroSchema<>(schemaInfo, pojoClassLoader);
if (schemaInfoProvider != null) {
schema.setSchemaInfoProvider(schemaInfoProvider);