sijie closed pull request #1994: Issue 1967: Non-persistent topic drop too many
messages
URL: https://github.com/apache/incubator-pulsar/pull/1994
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/AcknowledgmentsGroupingTracker.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/AcknowledgmentsGroupingTracker.java
index 68ab1fb191..195a9bb005 100644
---
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/AcknowledgmentsGroupingTracker.java
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/AcknowledgmentsGroupingTracker.java
@@ -18,193 +18,22 @@
*/
package org.apache.pulsar.client.impl;
-import io.netty.buffer.ByteBuf;
-import io.netty.channel.EventLoopGroup;
-
-import java.io.Closeable;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
import java.util.Map;
-import java.util.concurrent.ConcurrentSkipListSet;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
-
-import lombok.extern.slf4j.Slf4j;
-
-import org.apache.commons.lang3.tuple.Pair;
import org.apache.pulsar.client.api.MessageId;
-import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
-import org.apache.pulsar.common.api.Commands;
import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
/**
- * Group the acknowledgments for a certain time and then sends them out in a
single protobuf command.
+ * Acknowledgments grouping tracker.
*/
-@Slf4j
-public class AcknowledgmentsGroupingTracker implements Closeable {
-
- /**
- * When reaching the max group size, an ack command is sent out immediately
- */
- private static final int MAX_ACK_GROUP_SIZE = 1000;
-
- private final ConsumerImpl<?> consumer;
-
- private final long acknowledgementGroupTimeMicros;
-
- /**
- * Latest cumulative ack sent to broker
- */
- private volatile MessageIdImpl lastCumulativeAck = (MessageIdImpl)
MessageId.earliest;
-
- private static final
AtomicReferenceFieldUpdater<AcknowledgmentsGroupingTracker, MessageIdImpl>
LAST_CUMULATIVE_ACK_UPDATER = AtomicReferenceFieldUpdater
- .newUpdater(AcknowledgmentsGroupingTracker.class,
MessageIdImpl.class, "lastCumulativeAck");
-
- /**
- * This is a set of all the individual acks that the application has
issued and that were not already sent to
- * broker.
- */
- private final ConcurrentSkipListSet<MessageIdImpl> pendingIndividualAcks;
-
- private final ScheduledFuture<?> scheduledTask;
-
- public AcknowledgmentsGroupingTracker(ConsumerImpl<?> consumer,
ConsumerConfigurationData<?> conf,
- EventLoopGroup eventLoopGroup) {
- this.consumer = consumer;
- this.pendingIndividualAcks = new ConcurrentSkipListSet<>();
- this.acknowledgementGroupTimeMicros =
conf.getAcknowledgementsGroupTimeMicros();
-
- if (acknowledgementGroupTimeMicros > 0) {
- scheduledTask =
eventLoopGroup.next().scheduleWithFixedDelay(this::flush,
acknowledgementGroupTimeMicros,
- acknowledgementGroupTimeMicros, TimeUnit.MICROSECONDS);
- } else {
- scheduledTask = null;
- }
- }
-
- /**
- * Since the ack are delayed, we need to do some best-effort duplicate
check to discard messages that are being
- * resent after a disconnection and for which the user has already sent an
acknowlowdgement.
- */
- public boolean isDuplicate(MessageId messageId) {
- if (messageId.compareTo(lastCumulativeAck) <= 0) {
- // Already included in a cumulative ack
- return true;
- } else {
- return pendingIndividualAcks.contains(messageId);
- }
- }
+public interface AcknowledgmentsGroupingTracker extends AutoCloseable {
- public void addAcknowledgment(MessageIdImpl msgId, AckType ackType,
Map<String, Long> properties) {
- if (acknowledgementGroupTimeMicros == 0 || !properties.isEmpty()) {
- // We cannot group acks if the delay is 0 or when there are
properties attached to it. Fortunately that's an
- // uncommon condition since it's only used for the compaction
subscription.
- doImmediateAck(msgId, ackType, properties);
- } else if (ackType == AckType.Cumulative) {
- doCumulativeAck(msgId);
- } else {
- // Individual ack
- pendingIndividualAcks.add(msgId);
- if (pendingIndividualAcks.size() >= MAX_ACK_GROUP_SIZE) {
- flush();
- }
- }
- }
+ boolean isDuplicate(MessageId messageId);
- private void doCumulativeAck(MessageIdImpl msgId) {
- // Handle concurrent updates from different threads
- while (true) {
- MessageIdImpl lastCumlativeAck = this.lastCumulativeAck;
- if (msgId.compareTo(lastCumlativeAck) > 0) {
- if (LAST_CUMULATIVE_ACK_UPDATER.compareAndSet(this,
lastCumlativeAck, msgId)) {
- // Successfully updated the last cumlative ack. Next flush
iteration will send this to broker.
- return;
- }
- } else {
- // message id acknowledging an before the current last
cumulative ack
- return;
- }
- }
- }
+ void addAcknowledgment(MessageIdImpl msgId, AckType ackType, Map<String,
Long> properties);
- private boolean doImmediateAck(MessageIdImpl msgId, AckType ackType,
Map<String, Long> properties) {
- ClientCnx cnx = consumer.getClientCnx();
-
- if (cnx == null) {
- return false;
- }
-
- final ByteBuf cmd = Commands.newAck(consumer.consumerId,
msgId.getLedgerId(), msgId.getEntryId(), ackType, null,
- properties);
-
- cnx.ctx().writeAndFlush(cmd, cnx.ctx().voidPromise());
- return true;
- }
-
- /**
- * Flush all the pending acks and send them to the broker
- */
- public void flush() {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Flushing pending acks to broker:
last-cumulative-ack: {} -- individual-acks: {}", consumer,
- lastCumulativeAck, pendingIndividualAcks);
- }
-
- ClientCnx cnx = consumer.getClientCnx();
-
- if (cnx == null) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Cannot flush pending acks since we're not
connected to broker", consumer);
- }
- return;
- }
-
- if (!lastCumulativeAck.equals(MessageId.earliest)) {
- ByteBuf cmd = Commands.newAck(consumer.consumerId,
lastCumulativeAck.ledgerId, lastCumulativeAck.entryId,
- AckType.Cumulative, null, Collections.emptyMap());
- cnx.ctx().write(cmd, cnx.ctx().voidPromise());
- }
-
- // Flush all individual acks
- if (!pendingIndividualAcks.isEmpty()) {
- if
(Commands.peerSupportsMultiMessageAcknowledgment(cnx.getRemoteEndpointProtocolVersion()))
{
- // We can send 1 single protobuf command with all individual
acks
- List<Pair<Long, Long>> entriesToAck = new
ArrayList<>(pendingIndividualAcks.size());
- while (true) {
- MessageIdImpl msgId = pendingIndividualAcks.pollFirst();
- if (msgId == null) {
- break;
- }
-
- entriesToAck.add(Pair.of(msgId.getLedgerId(),
msgId.getEntryId()));
- }
-
-
cnx.ctx().write(Commands.newMultiMessageAck(consumer.consumerId, entriesToAck),
- cnx.ctx().voidPromise());
- } else {
- // When talking to older brokers, send the acknowledgments
individually
- while (true) {
- MessageIdImpl msgId = pendingIndividualAcks.pollFirst();
- if (msgId == null) {
- break;
- }
-
- cnx.ctx().write(Commands.newAck(consumer.consumerId,
msgId.getLedgerId(), msgId.getEntryId(),
- AckType.Individual, null, Collections.emptyMap()),
cnx.ctx().voidPromise());
- }
- }
- }
-
- cnx.ctx().flush();
- }
+ void flush();
@Override
- public void close() {
- flush();
- if (scheduledTask != null) {
- scheduledTask.cancel(true);
- }
- }
+ void close();
+
}
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
index dc98dc2e15..dd3b5c4675 100644
---
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
@@ -72,6 +72,7 @@
import org.apache.pulsar.common.api.proto.PulsarApi.ProtocolVersion;
import org.apache.pulsar.common.compression.CompressionCodec;
import org.apache.pulsar.common.compression.CompressionCodecProvider;
+import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.util.FutureUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -152,7 +153,15 @@
this.priorityLevel = conf.getPriorityLevel();
this.readCompacted = conf.isReadCompacted();
this.subscriptionInitialPosition =
conf.getSubscriptionInitialPosition();
- this.acknowledgmentsGroupingTracker = new
AcknowledgmentsGroupingTracker(this, conf, client.eventLoopGroup());
+
+ TopicName topicName = TopicName.get(topic);
+ if (topicName.isPersistent()) {
+ this.acknowledgmentsGroupingTracker =
+ new PersistentAcknowledgmentsGroupingTracker(this, conf,
client.eventLoopGroup());
+ } else {
+ this.acknowledgmentsGroupingTracker =
+ NonPersistentAcknowledgmentGroupingTracker.of();
+ }
if (client.getConfiguration().getStatsIntervalSeconds() > 0) {
stats = new ConsumerStatsRecorderImpl(client, conf, this);
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/NonPersistentAcknowledgmentGroupingTracker.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/NonPersistentAcknowledgmentGroupingTracker.java
new file mode 100644
index 0000000000..fbf6d0386e
--- /dev/null
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/NonPersistentAcknowledgmentGroupingTracker.java
@@ -0,0 +1,57 @@
+/**
+ * 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.impl;
+
+import java.util.Map;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
+
+/**
+ * A no-op acknowledgment grouping tracker.
+ */
+public class NonPersistentAcknowledgmentGroupingTracker implements
AcknowledgmentsGroupingTracker {
+
+ public static NonPersistentAcknowledgmentGroupingTracker of() {
+ return INSTANCE;
+ }
+
+ private static final NonPersistentAcknowledgmentGroupingTracker INSTANCE =
new NonPersistentAcknowledgmentGroupingTracker();
+
+ private NonPersistentAcknowledgmentGroupingTracker() {}
+
+ @Override
+ public boolean isDuplicate(MessageId messageId) {
+ return false;
+ }
+
+ @Override
+ public void addAcknowledgment(MessageIdImpl msgId, AckType ackType,
Map<String, Long> properties) {
+ // no-op
+ }
+
+ @Override
+ public void flush() {
+ // no-op
+ }
+
+ @Override
+ public void close() {
+ // no-op
+ }
+}
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java
new file mode 100644
index 0000000000..0fefe9c182
--- /dev/null
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java
@@ -0,0 +1,209 @@
+/**
+ * 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.impl;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.EventLoopGroup;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentSkipListSet;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+import org.apache.pulsar.common.api.Commands;
+import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
+
+/**
+ * Group the acknowledgments for a certain time and then sends them out in a
single protobuf command.
+ */
+@Slf4j
+public class PersistentAcknowledgmentsGroupingTracker implements
AcknowledgmentsGroupingTracker {
+
+ /**
+ * When reaching the max group size, an ack command is sent out immediately
+ */
+ private static final int MAX_ACK_GROUP_SIZE = 1000;
+
+ private final ConsumerImpl<?> consumer;
+
+ private final long acknowledgementGroupTimeMicros;
+
+ /**
+ * Latest cumulative ack sent to broker
+ */
+ private volatile MessageIdImpl lastCumulativeAck = (MessageIdImpl)
MessageId.earliest;
+
+ private static final
AtomicReferenceFieldUpdater<PersistentAcknowledgmentsGroupingTracker,
MessageIdImpl> LAST_CUMULATIVE_ACK_UPDATER = AtomicReferenceFieldUpdater
+ .newUpdater(PersistentAcknowledgmentsGroupingTracker.class,
MessageIdImpl.class, "lastCumulativeAck");
+
+ /**
+ * This is a set of all the individual acks that the application has
issued and that were not already sent to
+ * broker.
+ */
+ private final ConcurrentSkipListSet<MessageIdImpl> pendingIndividualAcks;
+
+ private final ScheduledFuture<?> scheduledTask;
+
+ public PersistentAcknowledgmentsGroupingTracker(ConsumerImpl<?> consumer,
ConsumerConfigurationData<?> conf,
+ EventLoopGroup
eventLoopGroup) {
+ this.consumer = consumer;
+ this.pendingIndividualAcks = new ConcurrentSkipListSet<>();
+ this.acknowledgementGroupTimeMicros =
conf.getAcknowledgementsGroupTimeMicros();
+
+ if (acknowledgementGroupTimeMicros > 0) {
+ scheduledTask =
eventLoopGroup.next().scheduleWithFixedDelay(this::flush,
acknowledgementGroupTimeMicros,
+ acknowledgementGroupTimeMicros, TimeUnit.MICROSECONDS);
+ } else {
+ scheduledTask = null;
+ }
+ }
+
+ /**
+ * Since the ack are delayed, we need to do some best-effort duplicate
check to discard messages that are being
+ * resent after a disconnection and for which the user has already sent an
acknowlowdgement.
+ */
+ public boolean isDuplicate(MessageId messageId) {
+ if (messageId.compareTo(lastCumulativeAck) <= 0) {
+ // Already included in a cumulative ack
+ return true;
+ } else {
+ return pendingIndividualAcks.contains(messageId);
+ }
+ }
+
+ public void addAcknowledgment(MessageIdImpl msgId, AckType ackType,
Map<String, Long> properties) {
+ if (acknowledgementGroupTimeMicros == 0 || !properties.isEmpty()) {
+ // We cannot group acks if the delay is 0 or when there are
properties attached to it. Fortunately that's an
+ // uncommon condition since it's only used for the compaction
subscription.
+ doImmediateAck(msgId, ackType, properties);
+ } else if (ackType == AckType.Cumulative) {
+ doCumulativeAck(msgId);
+ } else {
+ // Individual ack
+ pendingIndividualAcks.add(msgId);
+ if (pendingIndividualAcks.size() >= MAX_ACK_GROUP_SIZE) {
+ flush();
+ }
+ }
+ }
+
+ private void doCumulativeAck(MessageIdImpl msgId) {
+ // Handle concurrent updates from different threads
+ while (true) {
+ MessageIdImpl lastCumlativeAck = this.lastCumulativeAck;
+ if (msgId.compareTo(lastCumlativeAck) > 0) {
+ if (LAST_CUMULATIVE_ACK_UPDATER.compareAndSet(this,
lastCumlativeAck, msgId)) {
+ // Successfully updated the last cumlative ack. Next flush
iteration will send this to broker.
+ return;
+ }
+ } else {
+ // message id acknowledging an before the current last
cumulative ack
+ return;
+ }
+ }
+ }
+
+ private boolean doImmediateAck(MessageIdImpl msgId, AckType ackType,
Map<String, Long> properties) {
+ ClientCnx cnx = consumer.getClientCnx();
+
+ if (cnx == null) {
+ return false;
+ }
+
+ final ByteBuf cmd = Commands.newAck(consumer.consumerId,
msgId.getLedgerId(), msgId.getEntryId(), ackType, null,
+ properties);
+
+ cnx.ctx().writeAndFlush(cmd, cnx.ctx().voidPromise());
+ return true;
+ }
+
+ /**
+ * Flush all the pending acks and send them to the broker
+ */
+ public void flush() {
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Flushing pending acks to broker:
last-cumulative-ack: {} -- individual-acks: {}", consumer,
+ lastCumulativeAck, pendingIndividualAcks);
+ }
+
+ ClientCnx cnx = consumer.getClientCnx();
+
+ if (cnx == null) {
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Cannot flush pending acks since we're not
connected to broker", consumer);
+ }
+ return;
+ }
+
+ if (!lastCumulativeAck.equals(MessageId.earliest)) {
+ ByteBuf cmd = Commands.newAck(consumer.consumerId,
lastCumulativeAck.ledgerId, lastCumulativeAck.entryId,
+ AckType.Cumulative, null, Collections.emptyMap());
+ cnx.ctx().write(cmd, cnx.ctx().voidPromise());
+ }
+
+ // Flush all individual acks
+ if (!pendingIndividualAcks.isEmpty()) {
+ if
(Commands.peerSupportsMultiMessageAcknowledgment(cnx.getRemoteEndpointProtocolVersion()))
{
+ // We can send 1 single protobuf command with all individual
acks
+ List<Pair<Long, Long>> entriesToAck = new
ArrayList<>(pendingIndividualAcks.size());
+ while (true) {
+ MessageIdImpl msgId = pendingIndividualAcks.pollFirst();
+ if (msgId == null) {
+ break;
+ }
+
+ entriesToAck.add(Pair.of(msgId.getLedgerId(),
msgId.getEntryId()));
+ }
+
+
cnx.ctx().write(Commands.newMultiMessageAck(consumer.consumerId, entriesToAck),
+ cnx.ctx().voidPromise());
+ } else {
+ // When talking to older brokers, send the acknowledgments
individually
+ while (true) {
+ MessageIdImpl msgId = pendingIndividualAcks.pollFirst();
+ if (msgId == null) {
+ break;
+ }
+
+ cnx.ctx().write(Commands.newAck(consumer.consumerId,
msgId.getLedgerId(), msgId.getEntryId(),
+ AckType.Individual, null, Collections.emptyMap()),
cnx.ctx().voidPromise());
+ }
+ }
+ }
+
+ cnx.ctx().flush();
+ }
+
+ @Override
+ public void close() {
+ flush();
+ if (scheduledTask != null) {
+ scheduledTask.cancel(true);
+ }
+ }
+}
diff --git
a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/AcknowledgementsGroupingTrackerTest.java
b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/AcknowledgementsGroupingTrackerTest.java
index b6a3d2a001..fc6df91afc 100644
---
a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/AcknowledgementsGroupingTrackerTest.java
+++
b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/AcknowledgementsGroupingTrackerTest.java
@@ -61,7 +61,7 @@ public void teardown() {
public void testAckTracker() throws Exception {
ConsumerConfigurationData<?> conf = new ConsumerConfigurationData<>();
conf.setAcknowledgementsGroupTimeMicros(TimeUnit.SECONDS.toMicros(10));
- AcknowledgmentsGroupingTracker tracker = new
AcknowledgmentsGroupingTracker(consumer, conf, eventLoopGroup);
+ PersistentAcknowledgmentsGroupingTracker tracker = new
PersistentAcknowledgmentsGroupingTracker(consumer, conf, eventLoopGroup);
MessageIdImpl msg1 = new MessageIdImpl(5, 1, 0);
MessageIdImpl msg2 = new MessageIdImpl(5, 2, 0);
@@ -119,7 +119,7 @@ public void testAckTracker() throws Exception {
public void testImmediateAckingTracker() throws Exception {
ConsumerConfigurationData<?> conf = new ConsumerConfigurationData<>();
conf.setAcknowledgementsGroupTimeMicros(0);
- AcknowledgmentsGroupingTracker tracker = new
AcknowledgmentsGroupingTracker(consumer, conf, eventLoopGroup);
+ PersistentAcknowledgmentsGroupingTracker tracker = new
PersistentAcknowledgmentsGroupingTracker(consumer, conf, eventLoopGroup);
MessageIdImpl msg1 = new MessageIdImpl(5, 1, 0);
MessageIdImpl msg2 = new MessageIdImpl(5, 2, 0);
@@ -146,7 +146,7 @@ public void testImmediateAckingTracker() throws Exception {
public void testAckTrackerMultiAck() throws Exception {
ConsumerConfigurationData<?> conf = new ConsumerConfigurationData<>();
conf.setAcknowledgementsGroupTimeMicros(TimeUnit.SECONDS.toMicros(10));
- AcknowledgmentsGroupingTracker tracker = new
AcknowledgmentsGroupingTracker(consumer, conf, eventLoopGroup);
+ PersistentAcknowledgmentsGroupingTracker tracker = new
PersistentAcknowledgmentsGroupingTracker(consumer, conf, eventLoopGroup);
when(cnx.getRemoteEndpointProtocolVersion()).thenReturn(ProtocolVersion.v12_VALUE);
diff --git
a/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java
b/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java
index ba5da90aa5..ef455986c3 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java
@@ -170,6 +170,10 @@ private TopicName(String completeTopicName) {
}
+ public boolean isPersistent() {
+ return TopicDomain.persistent == domain;
+ }
+
/**
* Extract the namespace portion out of a completeTopicName name.
*
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
