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 584a376fac2bd..1d171f217e798 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 @@ -23,6 +23,8 @@ import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.ExecutionException; + import org.apache.bookkeeper.mledger.AsyncCallbacks.ClearBacklogCallback; import org.apache.bookkeeper.mledger.AsyncCallbacks.DeleteCallback; import org.apache.bookkeeper.mledger.AsyncCallbacks.FindEntryCallback; @@ -30,6 +32,7 @@ import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntriesCallback; import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntryCallback; import org.apache.bookkeeper.mledger.AsyncCallbacks.SkipEntriesCallback; +import org.apache.bookkeeper.mledger.impl.PositionImpl; /** * A ManangedCursor is a persisted cursor inside a ManagedLedger. @@ -75,6 +78,16 @@ enum IndividualDeletedEntries { */ Map getProperties(); + /** + * Return entry at the position. + */ + Entry readEntry(PositionImpl position) throws InterruptedException, ExecutionException; + + /** + * Return entry at the position async. + */ + void asyncReadEntry(PositionImpl position, ReadEntryCallback callback, Object ctx); + /** * Read entries from the ManagedLedger, up to the specified number. The returned list can be smaller. * 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 bab354762aff7..7cc8aff172eaa 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 @@ -44,7 +44,9 @@ import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; @@ -385,6 +387,27 @@ public void operationFailed(ManagedLedgerException exception) { }); } + @Override + public Entry readEntry(PositionImpl position) throws InterruptedException, ExecutionException { + final CompletableFuture readFuture = new CompletableFuture<>(); + ledger.asyncReadEntry(position, new ReadEntryCallback() { + @Override + public void readEntryComplete(Entry entry, Object ctx) { + readFuture.complete(entry); + } + @Override + public void readEntryFailed(ManagedLedgerException exception, Object ctx) { + readFuture.completeExceptionally(exception); + } + }, null); + return readFuture.get(); + } + + @Override + public void asyncReadEntry(PositionImpl position, ReadEntryCallback callback, Object ctx) { + ledger.asyncReadEntry(position, callback, ctx); + } + @Override public List readEntries(int numberOfEntriesToRead) throws InterruptedException, ManagedLedgerException { checkArgument(numberOfEntriesToRead > 0); 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 ecf6acfc5067c..694ffc317c503 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 @@ -27,6 +27,8 @@ import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.ExecutionException; + import org.apache.bookkeeper.mledger.AsyncCallbacks; import org.apache.bookkeeper.mledger.AsyncCallbacks.ClearBacklogCallback; import org.apache.bookkeeper.mledger.AsyncCallbacks.DeleteCallback; @@ -60,6 +62,17 @@ public Map getProperties() { return Collections.emptyMap(); } + @Override + public Entry readEntry(PositionImpl position) throws InterruptedException, ExecutionException { + return null; + } + + + @Override + public void asyncReadEntry(PositionImpl position, ReadEntryCallback callback, Object ctx) { + + } + @Override public boolean isDurable() { return true; diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java index f85c6175a7cb6..e0c4f8c816f8c 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java @@ -867,9 +867,8 @@ protected void internalCreateSubscription(String subscriptionName, MessageIdImpl if (topic.getSubscriptions().containsKey(subscriptionName)) { throw new RestException(Status.CONFLICT, "Subscription already exists for topic"); } - PersistentSubscription subscription = (PersistentSubscription) topic - .createSubscription(subscriptionName, InitialPosition.Latest).get(); + .createSubscription(subscriptionName, InitialPosition.Latest, 0, null).get(); subscription.resetCursor(PositionImpl.get(messageId.getLedgerId(), messageId.getEntryId())).get(); log.info("[{}][{}] Successfully created subscription {} at message id {}", clientAppId(), topicName, subscriptionName, messageId); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/RedeliveryTracker.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/RedeliveryTracker.java new file mode 100644 index 0000000000000..efbd89400cfc4 --- /dev/null +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/RedeliveryTracker.java @@ -0,0 +1,37 @@ +/** + * 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.service; + +import org.apache.bookkeeper.mledger.Position; + +import java.util.List; + +public interface RedeliveryTracker { + + int incrementAndGetRedeliveryCount(Position position); + + int getRedeliveryCount(Position position); + + void remove(Position position); + + void removeBatch(List positions); + + void clear(); + +} diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java index aacd7bcbf6f5d..97765a84fa8e3 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java @@ -523,6 +523,9 @@ protected void handleSubscribe(final CommandSubscribe subscribe) { final boolean readCompacted = subscribe.getReadCompacted(); final Map metadata = CommandUtils.metadataFromCommand(subscribe); final InitialPosition initialPosition = subscribe.getInitialPosition(); + final int maxRedeliveryCount = subscribe.getMaxRedeliveryCount(); + final String deadLetterTopic = subscribe.getDeadLetterTopic(); + final int maxUnackedMessagesPerConsumer = subscribe.getMaxUnackedMessagePerConsumer(); final SchemaData schema = subscribe.hasSchema() ? getSchema(subscribe.getSchema()) : null; CompletableFuture isProxyAuthorizedFuture; @@ -591,7 +594,8 @@ protected void handleSubscribe(final CommandSubscribe subscribe) { if (isCompatible) { return topic.subscribe(ServerCnx.this, subscriptionName, consumerId, subType, priorityLevel, consumerName, isDurable, - startMessageId, metadata, readCompacted, initialPosition); + startMessageId, metadata, readCompacted, initialPosition, + 0, null, maxUnackedMessagesPerConsumer); } else { return FutureUtil.failedFuture(new BrokerServiceException( "Trying to subscribe with incompatible schema" @@ -601,7 +605,8 @@ protected void handleSubscribe(final CommandSubscribe subscribe) { } else { return topic.subscribe(ServerCnx.this, subscriptionName, consumerId, subType, priorityLevel, consumerName, isDurable, - startMessageId, metadata, readCompacted, initialPosition); + startMessageId, metadata, readCompacted, initialPosition, + maxRedeliveryCount, deadLetterTopic, maxUnackedMessagesPerConsumer); } }) .thenAccept(consumer -> { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java index 7488204fcac11..334b942ae3d79 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java @@ -81,9 +81,11 @@ default long getOriginalSequenceId() { CompletableFuture subscribe(ServerCnx cnx, String subscriptionName, long consumerId, SubType subType, int priorityLevel, String consumerName, boolean isDurable, MessageId startMessageId, - Map metadata, boolean readCompacted, InitialPosition initialPosition); + Map metadata, boolean readCompacted, InitialPosition initialPosition, + int maxRedeliveryCount, String deadLetterTopic, int maxUnackedMessagesPerConsumer); - CompletableFuture createSubscription(String subscriptionName, InitialPosition initialPosition); + CompletableFuture createSubscription(String subscriptionName, InitialPosition initialPosition, + int maxRedeliveryCount, String deadLetterTopic); CompletableFuture unsubscribe(String subName); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentRedeliveryTracker.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentRedeliveryTracker.java new file mode 100644 index 0000000000000..ac5030446dbf2 --- /dev/null +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentRedeliveryTracker.java @@ -0,0 +1,60 @@ +/** + * 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.service.nonpersistent; + +import org.apache.bookkeeper.mledger.Position; +import org.apache.pulsar.broker.service.RedeliveryTracker; + +import java.util.List; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; + +public class NonPersistentRedeliveryTracker implements RedeliveryTracker { + + private ConcurrentHashMap trackerCache = new ConcurrentHashMap<>(16); + + @Override + public int incrementAndGetRedeliveryCount(Position position) { + trackerCache.putIfAbsent(position, new AtomicInteger(0)); + return trackerCache.get(position).incrementAndGet(); + } + + @Override + public int getRedeliveryCount(Position position) { + trackerCache.putIfAbsent(position, new AtomicInteger(0)); + return trackerCache.get(position).get(); + } + + @Override + public void remove(Position position) { + trackerCache.remove(position); + } + + @Override + public void removeBatch(List positions) { + if (positions != null) { + positions.forEach(this::remove); + } + } + + @Override + public void clear() { + trackerCache.clear(); + } +} diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java index 281d8438db89b..66fb9eb3895ea 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java @@ -314,7 +314,8 @@ public void removeProducer(Producer producer) { @Override public CompletableFuture subscribe(final ServerCnx cnx, String subscriptionName, long consumerId, SubType subType, int priorityLevel, String consumerName, boolean isDurable, MessageId startMessageId, - Map metadata, boolean readCompacted, InitialPosition initialPosition) { + Map metadata, boolean readCompacted, InitialPosition initialPosition, int maxRedeliveryCount, + String deadLetterTopic, int maxUnackedMessagesPerConsumer) { final CompletableFuture future = new CompletableFuture<>(); @@ -388,7 +389,8 @@ public CompletableFuture subscribe(final ServerCnx cnx, String subscri } @Override - public CompletableFuture createSubscription(String subscriptionName, InitialPosition initialPosition) { + public CompletableFuture createSubscription(String subscriptionName, InitialPosition initialPosition, + int maxRedeliveryCount, String deadLetterTopic) { return CompletableFuture.completedFuture(new NonPersistentSubscription(this, subscriptionName)); } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/CompactorSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/CompactorSubscription.java index 6fe74bfa93916..47217d68b9e09 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/CompactorSubscription.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/CompactorSubscription.java @@ -41,7 +41,7 @@ public class CompactorSubscription extends PersistentSubscription { public CompactorSubscription(PersistentTopic topic, CompactedTopic compactedTopic, String subscriptionName, ManagedCursor cursor) { - super(topic, subscriptionName, cursor); + super(topic, subscriptionName, cursor, 0, null); checkArgument(subscriptionName.equals(Compactor.COMPACTION_SUBSCRIPTION)); this.compactedTopic = compactedTopic; 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 17c5db7a6a819..b996e012e4da6 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 @@ -28,29 +28,39 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; -import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntriesCallback; -import org.apache.bookkeeper.mledger.Entry; +import io.netty.buffer.ByteBuf; import org.apache.bookkeeper.mledger.ManagedCursor; +import org.apache.bookkeeper.mledger.Position; +import org.apache.bookkeeper.mledger.Entry; import org.apache.bookkeeper.mledger.ManagedLedgerException; +import org.apache.bookkeeper.mledger.AsyncCallbacks; +import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntriesCallback; import org.apache.bookkeeper.mledger.ManagedLedgerException.NoMoreEntriesToReadException; import org.apache.bookkeeper.mledger.ManagedLedgerException.TooManyRequestsException; -import org.apache.bookkeeper.mledger.Position; import org.apache.bookkeeper.mledger.impl.PositionImpl; +import org.apache.commons.lang3.StringUtils; import org.apache.pulsar.broker.ServiceConfiguration; import org.apache.pulsar.broker.admin.AdminResource; import org.apache.pulsar.broker.service.AbstractDispatcherMultipleConsumers; +import org.apache.pulsar.broker.service.Dispatcher; +import org.apache.pulsar.broker.service.RedeliveryTracker; +import org.apache.pulsar.broker.service.Consumer; import org.apache.pulsar.broker.service.BrokerServiceException; import org.apache.pulsar.broker.service.BrokerServiceException.ConsumerBusyException; -import org.apache.pulsar.broker.service.Consumer; import org.apache.pulsar.broker.service.Consumer.SendMessageInfo; -import org.apache.pulsar.broker.service.Dispatcher; +import org.apache.pulsar.broker.service.nonpersistent.NonPersistentRedeliveryTracker; +import org.apache.pulsar.client.api.MessageId; +import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.impl.Backoff; +import org.apache.pulsar.client.impl.MessageImpl; +import org.apache.pulsar.client.impl.ProducerImpl; import org.apache.pulsar.common.api.proto.PulsarApi.CommandSubscribe.SubType; import org.apache.pulsar.common.naming.TopicName; import org.apache.pulsar.common.policies.data.Policies; import org.apache.pulsar.common.util.Codec; import org.apache.pulsar.common.util.collections.ConcurrentLongPairSet; import org.apache.pulsar.utils.CopyOnWriteArrayList; +import org.apache.pulsar.utils.Quorum; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -59,7 +69,7 @@ /** */ -public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMultipleConsumers implements Dispatcher, ReadEntriesCallback { +public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMultipleConsumers implements Dispatcher, ReadEntriesCallback { private static final int MaxReadBatchSize = 100; private static final int MaxRoundRobinBatchSize = 20; @@ -69,6 +79,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMu private CompletableFuture closeFuture = null; private ConcurrentLongPairSet messagesToReplay; + private ConcurrentLongPairSet messagesToDeadLetter; private boolean havePendingRead = false; private boolean havePendingReplayRead = false; @@ -88,20 +99,32 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMu private final ServiceConfiguration serviceConfig; private DispatchRateLimiter dispatchRateLimiter; + protected volatile int maxRedeliveryCount; + protected volatile String deadLetterTopic; + protected RedeliveryTracker redeliveryTracker; + private volatile ProducerImpl deadLetterTopicProducer; + enum ReadType { Normal, Replay } - public PersistentDispatcherMultipleConsumers(PersistentTopic topic, ManagedCursor cursor) { + public PersistentDispatcherMultipleConsumers(PersistentTopic topic, ManagedCursor cursor, int maxRedeliveryCount, + String deadLetterTopic) { this.cursor = cursor; this.name = topic.getName() + " / " + Codec.decode(cursor.getName()); this.topic = topic; this.messagesToReplay = new ConcurrentLongPairSet(512, 2); + this.messagesToDeadLetter = new ConcurrentLongPairSet(512, 2); this.readBatchSize = MaxReadBatchSize; this.maxUnackedMessages = topic.getBrokerService().pulsar().getConfiguration() .getMaxUnackedMessagesPerSubscription(); this.serviceConfig = topic.getBrokerService().pulsar().getConfiguration(); this.dispatchRateLimiter = null; + this.maxRedeliveryCount = maxRedeliveryCount; + this.deadLetterTopic = deadLetterTopic; + if (maxRedeliveryCount > 0) { + redeliveryTracker = new NonPersistentRedeliveryTracker(); + } } @Override @@ -132,11 +155,44 @@ public synchronized void addConsumer(Consumer consumer) throws BrokerServiceExce throw new ConsumerBusyException("Subscription reached max consumers limit"); } + deadLetterTopicProducer = newDeadLetterProducer(); + consumerList.add(consumer); consumerList.sort((c1, c2) -> c1.getPriorityLevel() - c2.getPriorityLevel()); consumerSet.add(consumer); } + private ProducerImpl newDeadLetterProducer() throws BrokerServiceException { + if (maxRedeliveryCount > 0 && deadLetterTopicProducer == null) { + try { + if (maxRedeliveryCount > 0 && StringUtils.isBlank(deadLetterTopic)) { + deadLetterTopic = String.format("%s-%s-DLQ", topic.getName(), Codec.decode(cursor.getName())); + } + return (ProducerImpl) topic.getBrokerService().pulsar().getClient().newProducer(Schema.BYTES) + .topic(deadLetterTopic) + .blockIfQueueFull(false) + .create(); + } catch (Throwable e) { + throw new BrokerServiceException(e); + } + } + return null; + } + + void reloadDeadLetterProducer() throws BrokerServiceException { + try { + if (deadLetterTopicProducer != null) { + deadLetterTopicProducer.closeAsync(); + } + deadLetterTopicProducer = newDeadLetterProducer(); + if (deadLetterTopicProducer == null) { + redeliveryTracker.clear(); + } + } catch (Throwable e) { + throw new BrokerServiceException(e); + } + } + private boolean isConsumersExceededOnTopic() { Policies policies; try { @@ -530,7 +586,7 @@ private boolean isAtleastOneConsumerAvailable() { // abort read if no consumers are connected or if disconnect is initiated return false; } - for(Consumer consumer : consumerList) { + for (Consumer consumer : consumerList) { if (isConsumerAvailable(consumer)) { return true; } @@ -556,11 +612,98 @@ public synchronized void redeliverUnacknowledgedMessages(Consumer consumer) { @Override public synchronized void redeliverUnacknowledgedMessages(Consumer consumer, List positions) { - positions.forEach(position -> messagesToReplay.add(position.getLedgerId(), position.getEntryId())); if (log.isDebugEnabled()) { log.debug("[{}-{}] Redelivering unacknowledged messages for consumer {}", name, consumer, positions); } - readMoreEntries(); + if (maxRedeliveryCount > 0 && redeliveryTracker != null) { + for (PositionImpl position : positions) { + if (redeliveryTracker.incrementAndGetRedeliveryCount(position) <= maxRedeliveryCount) { + messagesToReplay.add(position.getLedgerId(), position.getEntryId()); + } else { + messagesToDeadLetter.add(position.getLedgerId(), position.getEntryId()); + } + } + if (messagesToDeadLetter.size() > 0) { + + Quorum quorum = new Quorum(messagesToDeadLetter.size(), result -> { + readMoreEntries(); + }); + messagesToDeadLetter.forEach((ledgerId, entryId) -> { + PositionImpl position = PositionImpl.get(ledgerId, entryId); + cursor.asyncReadEntry(position, new AsyncCallbacks.ReadEntryCallback() { + @Override + public void readEntryComplete(Entry entry, Object ctx) { + if (entry == null) { + log.error("[{}-{}] Read an null entry from cursor {}", name, consumer, position); + quorum.succeed(); + } else { + try { + ByteBuf headersAndPayload = entry.getDataBuffer(); + MessageImpl msg = MessageImpl.deserialize(headersAndPayload); + headersAndPayload.retain(); + msg.setReplicatedFrom("DLQ"); + CompletableFuture future = deadLetterTopicProducer.sendAsync(msg); + future.whenCompleteAsync((messageId, error) -> { + if (error != null) { + log.error("[{}-{}] Fail to send message to dead letter topic {} {} {}", + name, consumer, deadLetterTopic, error.getMessage(), error); + messagesToReplay.add(position.getLedgerId(), position.getEntryId()); + quorum.succeed(); + } else { + cursor.asyncDelete(position, new AsyncCallbacks.DeleteCallback() { + @Override + public void deleteComplete(Object ctx) { + entry.release(); + redeliveryTracker.remove(position); + quorum.succeed(); + } + + @Override + public void deleteFailed(ManagedLedgerException exception, Object ctx) { + entry.release(); + messagesToReplay.add(position.getLedgerId(), position.getEntryId()); + quorum.succeed(); + } + }, position); + + } + }); + } catch (Throwable t) { + log.error("[{}-{}] Failed to deserialize message at {} {} {} {}", name, consumer, + entry.getPosition(), entry.getLedgerId(), t.getMessage(), t); + cursor.asyncDelete(position, new AsyncCallbacks.DeleteCallback() { + @Override + public void deleteComplete(Object ctx) { + entry.release(); + redeliveryTracker.remove(position); + quorum.succeed(); + } + + @Override + public void deleteFailed(ManagedLedgerException exception, Object ctx) { + entry.release(); + messagesToReplay.add(position.getLedgerId(), position.getEntryId()); + quorum.succeed(); + } + }, position); + + } + } + } + @Override + public void readEntryFailed(ManagedLedgerException exception, Object ctx) { + log.error("[{}-{}] Read entries failed {} {}", name, consumer, exception.getMessage(), exception); + messagesToReplay.add(position.getLedgerId(), position.getEntryId()); + quorum.succeed(); + } + }, null); + }); + messagesToDeadLetter.clear(); + } + } else { + positions.forEach(position -> messagesToReplay.add(position.getLedgerId(), position.getEntryId())); + readMoreEntries(); + } } @Override 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 dac9f055d704c..5bfa32ce7685e 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 @@ -27,7 +27,6 @@ import org.apache.bookkeeper.mledger.AsyncCallbacks; import org.apache.bookkeeper.mledger.AsyncCallbacks.ClearBacklogCallback; -import org.apache.bookkeeper.mledger.AsyncCallbacks.CloseCallback; import org.apache.bookkeeper.mledger.AsyncCallbacks.DeleteCallback; import org.apache.bookkeeper.mledger.AsyncCallbacks.MarkDeleteCallback; import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntryCallback; @@ -39,8 +38,8 @@ import org.apache.bookkeeper.mledger.ManagedLedgerException.InvalidCursorPositionException; import org.apache.bookkeeper.mledger.Position; import org.apache.bookkeeper.mledger.impl.PositionImpl; +import org.apache.commons.lang3.StringUtils; import org.apache.pulsar.broker.service.BrokerServiceException; -import org.apache.pulsar.broker.service.BrokerServiceException.PersistenceException; import org.apache.pulsar.broker.service.BrokerServiceException.ServerMetadataException; import org.apache.pulsar.broker.service.BrokerServiceException.SubscriptionBusyException; import org.apache.pulsar.broker.service.BrokerServiceException.SubscriptionFencedException; @@ -76,12 +75,18 @@ public class PersistentSubscription implements Subscription { // for connected subscriptions, message expiry will be checked if the backlog is greater than this threshold private static final int MINIMUM_BACKLOG_FOR_EXPIRY_CHECK = 1000; - public PersistentSubscription(PersistentTopic topic, String subscriptionName, ManagedCursor cursor) { + protected int maxRedeliveryCount = 0; + protected String deadLetterTopic = null; + + public PersistentSubscription(PersistentTopic topic, String subscriptionName, ManagedCursor cursor, + int maxRedeliveryCount, String deadLetterTopic) { this.topic = topic; this.cursor = cursor; this.topicName = topic.getName(); this.subName = subscriptionName; this.expiryMonitor = new PersistentMessageExpiryMonitor(topicName, subscriptionName, cursor); + this.maxRedeliveryCount = maxRedeliveryCount; + this.deadLetterTopic = deadLetterTopic; IS_FENCED_UPDATER.set(this, FALSE); } @@ -112,7 +117,7 @@ public synchronized void addConsumer(Consumer consumer) throws BrokerServiceExce break; case Shared: if (dispatcher == null || dispatcher.getType() != SubType.Shared) { - dispatcher = new PersistentDispatcherMultipleConsumers(topic, cursor); + dispatcher = new PersistentDispatcherMultipleConsumers(topic, cursor, maxRedeliveryCount, deadLetterTopic); } break; case Failover: @@ -153,7 +158,7 @@ public synchronized void removeConsumer(Consumer consumer) throws BrokerServiceE close(); // when topic closes: it iterates through concurrent-subscription map to close each subscription. so, // topic.remove again try to access same map which creates deadlock. so, execute it in different thread. - topic.getBrokerService().pulsar().getExecutor().submit(() ->{ + topic.getBrokerService().pulsar().getExecutor().submit(() -> { topic.removeSubscription(subName); }); } @@ -178,7 +183,7 @@ public void consumerFlow(Consumer consumer, int additionalNumberOfMessages) { } @Override - public void acknowledgeMessage(List positions, AckType ackType, Map properties) { + public void acknowledgeMessage(List positions, AckType ackType, Map properties) { if (ackType == AckType.Cumulative) { if (positions.size() != 1) { log.warn("[{}][{}] Invalid cumulative ack received with multiple message ids", topicName, subName); @@ -195,6 +200,11 @@ public void acknowledgeMessage(List positions, AckType ackType, Map subscribe(final ServerCnx cnx, String subscriptionName, long consumerId, SubType subType, int priorityLevel, String consumerName, boolean isDurable, MessageId startMessageId, - Map metadata, boolean readCompacted, InitialPosition initialPosition) { + Map metadata, boolean readCompacted, InitialPosition initialPosition, + int maxRedeliveryCount, String deadLetterTopic, int maxUnackedMessagesPerConsumer) { final CompletableFuture future = new CompletableFuture<>(); @@ -505,10 +509,13 @@ public CompletableFuture subscribe(final ServerCnx cnx, String subscri } CompletableFuture subscriptionFuture = isDurable ? // - getDurableSubscription(subscriptionName, initialPosition) // + getDurableSubscription(subscriptionName, initialPosition, maxRedeliveryCount, deadLetterTopic) // : getNonDurableSubscription(subscriptionName, startMessageId); - int maxUnackedMessages = isDurable ? brokerService.pulsar().getConfiguration().getMaxUnackedMessagesPerConsumer() :0; + final int maxUackedMessaegesPerConsumerInBroker = brokerService.pulsar().getConfiguration().getMaxUnackedMessagesPerConsumer(); + int maxUnackedMessages = isDurable ? ( + maxUnackedMessagesPerConsumer > 0 && maxUnackedMessagesPerConsumer < maxUackedMessaegesPerConsumerInBroker + ? maxUnackedMessagesPerConsumer : maxUackedMessaegesPerConsumerInBroker) : 0; subscriptionFuture.thenAccept(subscription -> { try { @@ -548,7 +555,8 @@ public CompletableFuture subscribe(final ServerCnx cnx, String subscri return future; } - private CompletableFuture getDurableSubscription(String subscriptionName, InitialPosition initialPosition) { + private CompletableFuture getDurableSubscription(String subscriptionName, InitialPosition initialPosition, + int maxRedeliveryCount, String deadLetterTopic) { CompletableFuture subscriptionFuture = new CompletableFuture<>(); ledger.asyncOpenCursor(Codec.encode(subscriptionName), initialPosition, new OpenCursorCallback() { @Override @@ -556,9 +564,28 @@ public void openCursorComplete(ManagedCursor cursor, Object ctx) { if (log.isDebugEnabled()) { log.debug("[{}][{}] Opened cursor", topic, subscriptionName); } - - subscriptionFuture.complete(subscriptions.computeIfAbsent(subscriptionName, - name -> createPersistentSubscription(subscriptionName, cursor))); + subscriptions.computeIfAbsent(subscriptionName, + name -> createPersistentSubscription(subscriptionName, cursor, maxRedeliveryCount, deadLetterTopic)); + + PersistentSubscription subscription = subscriptions.get(subscriptionName); + if (subscription.maxRedeliveryCount != maxRedeliveryCount) { + subscription.maxRedeliveryCount = maxRedeliveryCount; + if (subscription.dispatcher instanceof PersistentDispatcherMultipleConsumers) { + ((PersistentDispatcherMultipleConsumers) subscription.dispatcher).maxRedeliveryCount = maxRedeliveryCount; + } + } + if (!StringUtils.equals(subscription.deadLetterTopic, deadLetterTopic)) { + subscription.deadLetterTopic = deadLetterTopic; + if (subscription.dispatcher instanceof PersistentDispatcherMultipleConsumers) { + ((PersistentDispatcherMultipleConsumers) subscription.dispatcher).deadLetterTopic = deadLetterTopic; + try { + ((PersistentDispatcherMultipleConsumers) subscription.dispatcher).reloadDeadLetterProducer(); + } catch (Throwable e) { + subscriptionFuture.completeExceptionally(e); + } + } + } + subscriptionFuture.complete(subscription); } @Override @@ -602,7 +629,7 @@ private CompletableFuture getNonDurableSubscription(Stri subscriptionFuture.completeExceptionally(e); } - return new PersistentSubscription(this, subscriptionName, cursor); + return new PersistentSubscription(this, subscriptionName, cursor, 0, null); }); if (!subscriptionFuture.isDone()) { @@ -617,8 +644,9 @@ private CompletableFuture getNonDurableSubscription(Stri @SuppressWarnings("unchecked") @Override - public CompletableFuture createSubscription(String subscriptionName, InitialPosition initialPosition) { - return getDurableSubscription(subscriptionName, initialPosition); + public CompletableFuture createSubscription(String subscriptionName, InitialPosition initialPosition, + int maxRedeliveryCount, String deadLetterTopic) { + return getDurableSubscription(subscriptionName, initialPosition, maxRedeliveryCount, deadLetterTopic); } /** diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/utils/Quorum.java b/pulsar-broker/src/main/java/org/apache/pulsar/utils/Quorum.java new file mode 100644 index 0000000000000..8473a873d0f95 --- /dev/null +++ b/pulsar-broker/src/main/java/org/apache/pulsar/utils/Quorum.java @@ -0,0 +1,68 @@ +/** + * 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.utils; + +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Consumer; + +/** + * Quorum helper. Completes and invokes a callback when the number of {@link #succeed()} or + * {@link #fail()} calls equal the expected quorum count. + */ +public class Quorum { + + private final long quorum; + + private final Consumer callback; + + private AtomicLong succeeded = new AtomicLong(0); + + private AtomicLong failed = new AtomicLong(0); + + public Quorum(long quorum, Consumer callback) { + this.quorum = quorum; + this.callback = callback; + } + + private void complete(boolean result) { + if (callback != null) { + callback.accept(result); + } + } + + /** + * Indicates that a call in the quorum succeeded. + */ + public Quorum succeed() { + if (succeeded.incrementAndGet() >= quorum) { + complete(true); + } + return this; + } + + /** + * Indicates that a call in the quorum failed. + */ + public Quorum fail() { + if (failed.incrementAndGet() >= quorum) { + complete(false); + } + return this; + } +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java index c1be8d1b2feb2..17110a849aa1e 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java @@ -257,7 +257,7 @@ private void verifyActiveConsumerChange(CommandActiveConsumerChange change, @Test public void testConsumerGroupChangesWithOldNewConsumers() throws Exception { PersistentTopic topic = new PersistentTopic(successTopicName, ledgerMock, brokerService); - PersistentSubscription sub = new PersistentSubscription(topic, "sub-1", cursorMock); + PersistentSubscription sub = new PersistentSubscription(topic, "sub-1", cursorMock, 0, null); int partitionIndex = 0; PersistentDispatcherSingleActiveConsumer pdfc = new PersistentDispatcherSingleActiveConsumer(cursorMock, @@ -296,7 +296,7 @@ public void testAddRemoveConsumer() throws Exception { log.info("--- Starting PersistentDispatcherFailoverConsumerTest::testAddConsumer ---"); PersistentTopic topic = new PersistentTopic(successTopicName, ledgerMock, brokerService); - PersistentSubscription sub = new PersistentSubscription(topic, "sub-1", cursorMock); + PersistentSubscription sub = new PersistentSubscription(topic, "sub-1", cursorMock, 0, null); int partitionIndex = 0; PersistentDispatcherSingleActiveConsumer pdfc = new PersistentDispatcherSingleActiveConsumer(cursorMock, @@ -415,7 +415,7 @@ public void testAddRemoveConsumer() throws Exception { public void testMultipleDispatcherGetNextConsumerWithDifferentPriorityLevel() throws Exception { PersistentTopic topic = new PersistentTopic(successTopicName, ledgerMock, brokerService); - PersistentDispatcherMultipleConsumers dispatcher = new PersistentDispatcherMultipleConsumers(topic, cursorMock); + PersistentDispatcherMultipleConsumers dispatcher = new PersistentDispatcherMultipleConsumers(topic, cursorMock, 0, null); Consumer consumer1 = createConsumer(0, 2, false, 1); Consumer consumer2 = createConsumer(0, 2, false, 2); Consumer consumer3 = createConsumer(0, 2, false, 3); @@ -459,7 +459,7 @@ public void testMultipleDispatcherGetNextConsumerWithDifferentPriorityLevel() th @Test public void testFewBlockedConsumerSamePriority() throws Exception{ PersistentTopic topic = new PersistentTopic(successTopicName, ledgerMock, brokerService); - PersistentDispatcherMultipleConsumers dispatcher = new PersistentDispatcherMultipleConsumers(topic, cursorMock); + PersistentDispatcherMultipleConsumers dispatcher = new PersistentDispatcherMultipleConsumers(topic, cursorMock, 0, null); Consumer consumer1 = createConsumer(0, 2, false, 1); Consumer consumer2 = createConsumer(0, 2, false, 2); Consumer consumer3 = createConsumer(0, 2, false, 3); @@ -486,7 +486,7 @@ public void testFewBlockedConsumerSamePriority() throws Exception{ @Test public void testFewBlockedConsumerDifferentPriority() throws Exception { PersistentTopic topic = new PersistentTopic(successTopicName, ledgerMock, brokerService); - PersistentDispatcherMultipleConsumers dispatcher = new PersistentDispatcherMultipleConsumers(topic, cursorMock); + PersistentDispatcherMultipleConsumers dispatcher = new PersistentDispatcherMultipleConsumers(topic, cursorMock, 0, null); Consumer consumer1 = createConsumer(0, 2, false, 1); Consumer consumer2 = createConsumer(0, 2, false, 2); Consumer consumer3 = createConsumer(0, 2, false, 3); @@ -540,7 +540,7 @@ public void testFewBlockedConsumerDifferentPriority() throws Exception { @Test public void testFewBlockedConsumerDifferentPriority2() throws Exception { PersistentTopic topic = new PersistentTopic(successTopicName, ledgerMock, brokerService); - PersistentDispatcherMultipleConsumers dispatcher = new PersistentDispatcherMultipleConsumers(topic, cursorMock); + PersistentDispatcherMultipleConsumers dispatcher = new PersistentDispatcherMultipleConsumers(topic, cursorMock, 0, null); Consumer consumer1 = createConsumer(0, 2, true, 1); Consumer consumer2 = createConsumer(0, 2, true, 2); Consumer consumer3 = createConsumer(0, 2, true, 3); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicConcurrentTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicConcurrentTest.java index 7c5ca2077da5e..4ac755898ff3d 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicConcurrentTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicConcurrentTest.java @@ -123,7 +123,7 @@ public void testConcurrentTopicAndSubscriptionDelete() throws Exception { .setSubType(PulsarApi.CommandSubscribe.SubType.Exclusive).build(); Future f1 = topic.subscribe(serverCnx, cmd.getSubscription(), cmd.getConsumerId(), cmd.getSubType(), - 0, cmd.getConsumerName(), cmd.getDurable(), null, Collections.emptyMap(), cmd.getReadCompacted(), InitialPosition.Latest); + 0, cmd.getConsumerName(), cmd.getDurable(), null, Collections.emptyMap(), cmd.getReadCompacted(), InitialPosition.Latest, 0, null, 0); f1.get(); final CyclicBarrier barrier = new CyclicBarrier(2); @@ -181,7 +181,7 @@ public void testConcurrentTopicGCAndSubscriptionDelete() throws Exception { .setSubType(PulsarApi.CommandSubscribe.SubType.Exclusive).build(); Future f1 = topic.subscribe(serverCnx, cmd.getSubscription(), cmd.getConsumerId(), cmd.getSubType(), - 0, cmd.getConsumerName(), cmd.getDurable(), null, Collections.emptyMap(), cmd.getReadCompacted(), InitialPosition.Latest); + 0, cmd.getConsumerName(), cmd.getDurable(), null, Collections.emptyMap(), cmd.getReadCompacted(), InitialPosition.Latest, 0, null, 0); f1.get(); final CyclicBarrier barrier = new CyclicBarrier(2); @@ -243,7 +243,7 @@ public void testConcurrentTopicDeleteAndUnsubscribe() throws Exception { .setSubType(PulsarApi.CommandSubscribe.SubType.Exclusive).build(); Future f1 = topic.subscribe(serverCnx, cmd.getSubscription(), cmd.getConsumerId(), cmd.getSubType(), - 0, cmd.getConsumerName(), cmd.getDurable(), null, Collections.emptyMap(), cmd.getReadCompacted(), InitialPosition.Latest); + 0, cmd.getConsumerName(), cmd.getDurable(), null, Collections.emptyMap(), cmd.getReadCompacted(), InitialPosition.Latest, 0, null, 0); f1.get(); final CyclicBarrier barrier = new CyclicBarrier(2); @@ -301,7 +301,7 @@ public void testConcurrentTopicDeleteAndSubsUnsubscribe() throws Exception { .setSubType(PulsarApi.CommandSubscribe.SubType.Exclusive).build(); Future f1 = topic.subscribe(serverCnx, cmd.getSubscription(), cmd.getConsumerId(), cmd.getSubType(), - 0, cmd.getConsumerName(), cmd.getDurable(), null, Collections.emptyMap(), cmd.getReadCompacted(), InitialPosition.Latest); + 0, cmd.getConsumerName(), cmd.getDurable(), null, Collections.emptyMap(), cmd.getReadCompacted(), InitialPosition.Latest, 0, null, 0); f1.get(); final CyclicBarrier barrier = new CyclicBarrier(2); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java index f8ac40a945f80..5675a1a3fd9de 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java @@ -280,7 +280,7 @@ public void testDispatcherMultiConsumerReadFailed() throws Exception { PersistentTopic topic = spy(new PersistentTopic(successTopicName, ledgerMock, brokerService)); ManagedCursor cursor = mock(ManagedCursor.class); when(cursor.getName()).thenReturn("cursor"); - PersistentDispatcherMultipleConsumers dispatcher = new PersistentDispatcherMultipleConsumers(topic, cursor); + PersistentDispatcherMultipleConsumers dispatcher = new PersistentDispatcherMultipleConsumers(topic, cursor, 0, null); dispatcher.readEntriesFailed(new ManagedLedgerException.InvalidCursorPositionException("failed"), null); verify(topic, atLeast(1)).getBrokerService(); } @@ -428,7 +428,7 @@ public void testSubscribeFail() throws Exception { .setSubscription("").setRequestId(1).setSubType(SubType.Exclusive).build(); Future f1 = topic.subscribe(serverCnx, cmd.getSubscription(), cmd.getConsumerId(), cmd.getSubType(), - 0, cmd.getConsumerName(), cmd.getDurable(), null, Collections.emptyMap(), cmd.getReadCompacted(), InitialPosition.Latest); + 0, cmd.getConsumerName(), cmd.getDurable(), null, Collections.emptyMap(), cmd.getReadCompacted(), InitialPosition.Latest, 0, null, 0); try { f1.get(); fail("should fail with exception"); @@ -447,12 +447,12 @@ public void testSubscribeUnsubscribe() throws Exception { // 1. simple subscribe Future f1 = topic.subscribe(serverCnx, cmd.getSubscription(), cmd.getConsumerId(), cmd.getSubType(), - 0, cmd.getConsumerName(), cmd.getDurable(), null, Collections.emptyMap(), cmd.getReadCompacted(), InitialPosition.Latest); + 0, cmd.getConsumerName(), cmd.getDurable(), null, Collections.emptyMap(), cmd.getReadCompacted(), InitialPosition.Latest, 0, null, 0); f1.get(); // 2. duplicate subscribe Future f2 = topic.subscribe(serverCnx, cmd.getSubscription(), cmd.getConsumerId(), cmd.getSubType(), - 0, cmd.getConsumerName(), cmd.getDurable(), null, Collections.emptyMap(), cmd.getReadCompacted(), InitialPosition.Latest); + 0, cmd.getConsumerName(), cmd.getDurable(), null, Collections.emptyMap(), cmd.getReadCompacted(), InitialPosition.Latest, 0, null, 0); try { f2.get(); @@ -472,7 +472,7 @@ public void testSubscribeUnsubscribe() throws Exception { @Test public void testAddRemoveConsumer() throws Exception { PersistentTopic topic = new PersistentTopic(successTopicName, ledgerMock, brokerService); - PersistentSubscription sub = new PersistentSubscription(topic, "sub-1", cursorMock); + PersistentSubscription sub = new PersistentSubscription(topic, "sub-1", cursorMock, 0, null); // 1. simple add consumer Consumer consumer = new Consumer(sub, SubType.Exclusive, topic.getName(), 1 /* consumer id */, 0, "Cons1"/* consumer name */, @@ -503,8 +503,8 @@ public void testAddRemoveConsumer() throws Exception { public void testMaxConsumersShared() throws Exception { PersistentTopic topic = new PersistentTopic(successTopicName, ledgerMock, brokerService); - PersistentSubscription sub = new PersistentSubscription(topic, "sub-1", cursorMock); - PersistentSubscription sub2 = new PersistentSubscription(topic, "sub-2", cursorMock); + PersistentSubscription sub = new PersistentSubscription(topic, "sub-1", cursorMock, 0, null); + PersistentSubscription sub2 = new PersistentSubscription(topic, "sub-2", cursorMock, 0, null); // for count consumers on topic ConcurrentOpenHashMap subscriptions = new ConcurrentOpenHashMap<>(16, 1); @@ -594,8 +594,8 @@ public void testMaxConsumersSharedForNamespace() throws Exception { public void testMaxConsumersFailover() throws Exception { PersistentTopic topic = new PersistentTopic(successTopicName, ledgerMock, brokerService); - PersistentSubscription sub = new PersistentSubscription(topic, "sub-1", cursorMock); - PersistentSubscription sub2 = new PersistentSubscription(topic, "sub-2", cursorMock); + PersistentSubscription sub = new PersistentSubscription(topic, "sub-1", cursorMock, 0, null); + PersistentSubscription sub2 = new PersistentSubscription(topic, "sub-2", cursorMock, 0, null); // for count consumers on topic ConcurrentOpenHashMap subscriptions = new ConcurrentOpenHashMap<>(16, 1); @@ -685,7 +685,7 @@ public void testMaxConsumersFailoverForNamespace() throws Exception { @Test public void testUbsubscribeRaceConditions() throws Exception { PersistentTopic topic = new PersistentTopic(successTopicName, ledgerMock, brokerService); - PersistentSubscription sub = new PersistentSubscription(topic, "sub-1", cursorMock); + PersistentSubscription sub = new PersistentSubscription(topic, "sub-1", cursorMock, 0, null); Consumer consumer1 = new Consumer(sub, SubType.Exclusive, topic.getName(), 1 /* consumer id */, 0, "Cons1"/* consumer name */, 50000, serverCnx, "myrole-1", Collections.emptyMap(), false /* read compacted */, InitialPosition.Latest); sub.addConsumer(consumer1); @@ -739,7 +739,7 @@ public void testDeleteTopic() throws Exception { .setSubscription(successSubName).setRequestId(1).setSubType(SubType.Exclusive).build(); Future f1 = topic.subscribe(serverCnx, cmd.getSubscription(), cmd.getConsumerId(), cmd.getSubType(), - 0, cmd.getConsumerName(), cmd.getDurable(), null, Collections.emptyMap(), false /* read compacted */, InitialPosition.Latest); + 0, cmd.getConsumerName(), cmd.getDurable(), null, Collections.emptyMap(), false /* read compacted */, InitialPosition.Latest, 0, null, 0); f1.get(); assertTrue(topic.delete().isCompletedExceptionally()); @@ -754,7 +754,7 @@ public void testDeleteAndUnsubscribeTopic() throws Exception { .setSubscription(successSubName).setRequestId(1).setSubType(SubType.Exclusive).build(); Future f1 = topic.subscribe(serverCnx, cmd.getSubscription(), cmd.getConsumerId(), cmd.getSubType(), - 0, cmd.getConsumerName(), cmd.getDurable(), null, Collections.emptyMap(), cmd.getReadCompacted(), InitialPosition.Latest); + 0, cmd.getConsumerName(), cmd.getDurable(), null, Collections.emptyMap(), cmd.getReadCompacted(), InitialPosition.Latest, 0, null, 0); f1.get(); final CyclicBarrier barrier = new CyclicBarrier(2); @@ -808,7 +808,7 @@ public void testConcurrentTopicAndSubscriptionDelete() throws Exception { .setSubscription(successSubName).setRequestId(1).setSubType(SubType.Exclusive).build(); Future f1 = topic.subscribe(serverCnx, cmd.getSubscription(), cmd.getConsumerId(), cmd.getSubType(), - 0, cmd.getConsumerName(), cmd.getDurable(), null, Collections.emptyMap(), cmd.getReadCompacted(), InitialPosition.Latest); + 0, cmd.getConsumerName(), cmd.getDurable(), null, Collections.emptyMap(), cmd.getReadCompacted(), InitialPosition.Latest, 0, null, 0); f1.get(); final CyclicBarrier barrier = new CyclicBarrier(2); @@ -895,7 +895,7 @@ public Object answer(InvocationOnMock invocationOnMock) throws Throwable { .setSubscription(successSubName).setRequestId(1).setSubType(SubType.Exclusive).build(); Future f = topic.subscribe(serverCnx, cmd.getSubscription(), cmd.getConsumerId(), cmd.getSubType(), - 0, cmd.getConsumerName(), cmd.getDurable(), null, Collections.emptyMap(), cmd.getReadCompacted(), InitialPosition.Latest); + 0, cmd.getConsumerName(), cmd.getDurable(), null, Collections.emptyMap(), cmd.getReadCompacted(), InitialPosition.Latest, 0, null, 0); try { f.get(); @@ -1013,7 +1013,7 @@ public void testFailoverSubscription() throws Exception { // 1. Subscribe with non partition topic Future f1 = topic1.subscribe(serverCnx, cmd1.getSubscription(), cmd1.getConsumerId(), cmd1.getSubType(), 0, cmd1.getConsumerName(), cmd1.getDurable(), null, Collections.emptyMap(), - cmd1.getReadCompacted(), InitialPosition.Latest); + cmd1.getReadCompacted(), InitialPosition.Latest, 0, null, 0); f1.get(); // 2. Subscribe with partition topic @@ -1025,7 +1025,7 @@ public void testFailoverSubscription() throws Exception { Future f2 = topic2.subscribe(serverCnx, cmd2.getSubscription(), cmd2.getConsumerId(), cmd2.getSubType(), 0, cmd2.getConsumerName(), cmd2.getDurable(), null, Collections.emptyMap(), - cmd2.getReadCompacted(), InitialPosition.Latest); + cmd2.getReadCompacted(), InitialPosition.Latest, 0, null, 0); f2.get(); // 3. Subscribe and create second consumer @@ -1035,7 +1035,7 @@ public void testFailoverSubscription() throws Exception { Future f3 = topic2.subscribe(serverCnx, cmd3.getSubscription(), cmd3.getConsumerId(), cmd3.getSubType(), 0, cmd3.getConsumerName(), cmd3.getDurable(), null, Collections.emptyMap(), - cmd3.getReadCompacted(), InitialPosition.Latest); + cmd3.getReadCompacted(), InitialPosition.Latest, 0, null, 0); f3.get(); assertEquals( @@ -1056,7 +1056,7 @@ public void testFailoverSubscription() throws Exception { Future f4 = topic2.subscribe(serverCnx, cmd4.getSubscription(), cmd4.getConsumerId(), cmd4.getSubType(), 0, cmd4.getConsumerName(), cmd4.getDurable(), null, Collections.emptyMap(), - cmd4.getReadCompacted(), InitialPosition.Latest); + cmd4.getReadCompacted(), InitialPosition.Latest, 0, null, 0); f4.get(); assertEquals( @@ -1082,7 +1082,7 @@ public void testFailoverSubscription() throws Exception { Future f5 = topic2.subscribe(serverCnx, cmd5.getSubscription(), cmd5.getConsumerId(), cmd5.getSubType(), 0, cmd5.getConsumerName(), cmd5.getDurable(), null, Collections.emptyMap(), - cmd5.getReadCompacted(), InitialPosition.Latest); + cmd5.getReadCompacted(), InitialPosition.Latest, 0, null, 0); try { f5.get(); @@ -1099,7 +1099,7 @@ public void testFailoverSubscription() throws Exception { Future f6 = topic2.subscribe(serverCnx, cmd6.getSubscription(), cmd6.getConsumerId(), cmd6.getSubType(), 0, cmd6.getConsumerName(), cmd6.getDurable(), null, Collections.emptyMap(), - cmd6.getReadCompacted(), InitialPosition.Latest); + cmd6.getReadCompacted(), InitialPosition.Latest, 0, null, 0); f6.get(); // 7. unsubscribe exclusive sub diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java index 61bdad034ad1f..365f8e2517bf2 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java @@ -22,11 +22,7 @@ import static org.mockito.Mockito.atLeastOnce; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.verify; -import static org.testng.Assert.assertEquals; -import static org.testng.Assert.assertFalse; -import static org.testng.Assert.assertNotEquals; -import static org.testng.Assert.assertTrue; -import static org.testng.Assert.fail; +import static org.testng.Assert.*; import java.io.IOException; import java.lang.reflect.Field; @@ -2662,4 +2658,142 @@ public void received(Consumer consumer, Message message) assertEquals(latch.getCount(), 1); consumer.close(); } + + @Test + public void testDeadLetterTopic() throws Exception { + + final String topic = "persistent://my-property/my-ns/dead-letter-topic"; + + final int maxRedeliveryCount = 2; + + final int sendMessages = 100; + + Producer producer = pulsarClient.newProducer(Schema.BYTES) + .topic(topic) + .create(); + + for (int i = 0; i < sendMessages; i++) { + producer.send(String.format("Hello Pulsar [%d]", i).getBytes()); + } + + producer.close(); + + Consumer consumer = pulsarClient.newConsumer(Schema.BYTES) + .topic(topic) + .subscriptionName("my-subscription") + .subscriptionType(SubscriptionType.Shared) + .ackTimeout(3, TimeUnit.SECONDS) + .maxRedeliveryCount(maxRedeliveryCount) + .receiverQueueSize(100) + .maxUnackedMessagesPerConsumer(100) + .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest) + .subscribe(); + + Consumer deadLetterConsumer = pulsarClient.newConsumer(Schema.BYTES) + .topic("persistent://my-property/my-ns/dead-letter-topic-my-subscription-DLQ") + .subscriptionName("my-subscription") + .subscribe(); + + int totalReceived = 0; + do { + Message message = consumer.receive(); + log.info("consumer received message : {} {}", message.getMessageId(), new String(message.getData())); + totalReceived++; + } while (totalReceived < sendMessages * (maxRedeliveryCount + 1)); + + int totalInDeadLetter = 0; + do { + Message message = deadLetterConsumer.receive(); + log.info("dead letter consumer received message : {} {}", message.getMessageId(), new String(message.getData())); + deadLetterConsumer.acknowledge(message); + totalInDeadLetter++; + } while (totalInDeadLetter < sendMessages); + + deadLetterConsumer.close(); + consumer.close(); + + Consumer checkConsumer = pulsarClient.newConsumer(Schema.BYTES) + .topic(topic) + .subscriptionName("my-subscription") + .subscriptionType(SubscriptionType.Shared) + .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest) + .subscribe(); + + Message checkMessage = checkConsumer.receive(3, TimeUnit.SECONDS); + if (checkMessage != null) { + log.info("check consumer received message : {} {}", checkMessage.getMessageId(), new String(checkMessage.getData())); + } + assertNull(checkMessage); + + checkConsumer.close(); + } + + @Test + public void testDeadLetterTopicByCustomTopicName() throws Exception { + final String topic = "persistent://my-property/my-ns/dead-letter-topic"; + + final int maxRedeliveryCount = 2; + + final int sendMessages = 100; + + Producer producer = pulsarClient.newProducer(Schema.BYTES) + .topic(topic) + .create(); + + for (int i = 0; i < sendMessages; i++) { + producer.send(String.format("Hello Pulsar [%d]", i).getBytes()); + } + + producer.close(); + + Consumer consumer = pulsarClient.newConsumer(Schema.BYTES) + .topic(topic) + .subscriptionName("my-subscription") + .subscriptionType(SubscriptionType.Shared) + .ackTimeout(3, TimeUnit.SECONDS) + .maxRedeliveryCount(maxRedeliveryCount) + .receiverQueueSize(100) + .maxUnackedMessagesPerConsumer(100) + .deadLetterTopic("persistent://my-property/my-ns/dead-letter-custom-topic-my-subscription-custom-DLQ") + .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest) + .subscribe(); + + Consumer deadLetterConsumer = pulsarClient.newConsumer(Schema.BYTES) + .topic("persistent://my-property/my-ns/dead-letter-custom-topic-my-subscription-custom-DLQ") + .subscriptionName("my-subscription") + .subscribe(); + + int totalReceived = 0; + do { + Message message = consumer.receive(); + log.info("consumer received message : {} {}", message.getMessageId(), new String(message.getData())); + totalReceived++; + } while (totalReceived < sendMessages * (maxRedeliveryCount + 1)); + + int totalInDeadLetter = 0; + do { + Message message = deadLetterConsumer.receive(); + log.info("dead letter consumer received message : {} {}", message.getMessageId(), new String(message.getData())); + deadLetterConsumer.acknowledge(message); + totalInDeadLetter++; + } while (totalInDeadLetter < sendMessages); + + deadLetterConsumer.close(); + consumer.close(); + + Consumer checkConsumer = pulsarClient.newConsumer(Schema.BYTES) + .topic(topic) + .subscriptionName("my-subscription") + .subscriptionType(SubscriptionType.Shared) + .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest) + .subscribe(); + + Message checkMessage = checkConsumer.receive(3, TimeUnit.SECONDS); + if (checkMessage != null) { + log.info("check consumer received message : {} {}", checkMessage.getMessageId(), new String(checkMessage.getData())); + } + assertNull(checkMessage); + + checkConsumer.close(); + } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/utils/QuorumTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/utils/QuorumTest.java new file mode 100644 index 0000000000000..ec3413c056ff7 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/utils/QuorumTest.java @@ -0,0 +1,42 @@ +/** + * 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.utils; + + +import org.junit.Assert; +import org.testng.annotations.Test; + +public class QuorumTest { + + @Test + public void testSucceed() { + Quorum quorum = new Quorum(3, Assert::assertTrue); + for (int i = 0; i < 3; i++) { + quorum.succeed(); + } + } + + @Test + public void testFail() { + Quorum quorum = new Quorum(3, Assert::assertFalse); + for (int i = 0; i < 3; i++) { + quorum.fail(); + } + } +} diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/api/ConsumerBuilder.java b/pulsar-client/src/main/java/org/apache/pulsar/client/api/ConsumerBuilder.java index 8657859103c51..5958bd5def08c 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/api/ConsumerBuilder.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/api/ConsumerBuilder.java @@ -330,4 +330,24 @@ public interface ConsumerBuilder extends Cloneable { * Set subscriptionInitialPosition for the consumer */ ConsumerBuilder subscriptionInitialPosition(SubscriptionInitialPosition subscriptionInitialPosition); + + /** + * Set maximum number of redelivery. + * Message exceeding the maximum number of redelivery will send to Dead Letter Topic and acknowledged automatic. + * @param maxRedeliveryCount maximum number of redelivery + */ + ConsumerBuilder maxRedeliveryCount(int maxRedeliveryCount); + + /** + * Set name of Dead Letter Topic. + * Before set name of Dead Letter Topic, ensure that maxRedeliveryCount > 0 + * @param deadLetterTopic name of dead letter topic + */ + ConsumerBuilder deadLetterTopic(String deadLetterTopic); + + /** + * Set max un-acked messages per consumer. + * This config should less than broker config, if not, config will not enable. 0 is not limit, default is 0 + */ + ConsumerBuilder maxUnackedMessagesPerConsumer(int maxUnackedMessagesPerConsumer); } 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 f0067f7f10f2f..085d0c3fbb397 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 @@ -242,7 +242,25 @@ public ConsumerBuilder subscriptionInitialPosition(SubscriptionInitialPositio return this; } - public ConsumerConfigurationData getConf() { + @Override + public ConsumerBuilder maxRedeliveryCount(int maxRedeliveryCount) { + conf.setMaxRedeliveryCount(maxRedeliveryCount); + return this; + } + + @Override + public ConsumerBuilder deadLetterTopic(String deadLetterTopic) { + conf.setDeadLetterTopic(deadLetterTopic); + return this; + } + + @Override + public ConsumerBuilder maxUnackedMessagesPerConsumer(int maxUnackedMessagesPerConsumer) { + conf.setMaxUnackedMessagesPerConsumer(maxUnackedMessagesPerConsumer); + return this; + } + + public ConsumerConfigurationData getConf() { return conf; } } 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 1c69c754a84c5..18fa97baa2612 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 @@ -131,6 +131,10 @@ public class ConsumerImpl extends ConsumerBase implements ConnectionHandle private final SubscriptionInitialPosition subscriptionInitialPosition; private final ConnectionHandler connectionHandler; + private final int maxRedeliveryCount; + private final String deadLetterTopic; + private final int maxUnackedMessagesPerConsumer; + enum SubscriptionMode { // Make the subscription to be backed by a durable cursor that will retain messages and persist the current // position @@ -159,6 +163,9 @@ enum SubscriptionMode { this.priorityLevel = conf.getPriorityLevel(); this.readCompacted = conf.isReadCompacted(); this.subscriptionInitialPosition = conf.getSubscriptionInitialPosition(); + this.maxRedeliveryCount = conf.getMaxRedeliveryCount(); + this.deadLetterTopic = conf.getDeadLetterTopic(); + this.maxUnackedMessagesPerConsumer = conf.getMaxUnackedMessagesPerConsumer(); TopicName topicName = TopicName.get(topic); if (topicName.isPersistent()) { @@ -486,7 +493,9 @@ public void connectionOpened(final ClientCnx cnx) { } ByteBuf request = Commands.newSubscribe(topic, subscription, consumerId, requestId, getSubType(), priorityLevel, - consumerName, isDurable, startMessageIdData, metadata, readCompacted, InitialPosition.valueOf(subscriptionInitialPosition.getValue()), schema.getSchemaInfo()); + consumerName, isDurable, startMessageIdData, metadata, readCompacted, + InitialPosition.valueOf(subscriptionInitialPosition.getValue()), schema.getSchemaInfo(), + maxRedeliveryCount, deadLetterTopic, maxUnackedMessagesPerConsumer); if (startMessageIdData != null) { startMessageIdData.recycle(); } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConsumerConfigurationData.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConsumerConfigurationData.java index 47ede41de6000..6751dc0608832 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConsumerConfigurationData.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConsumerConfigurationData.java @@ -82,6 +82,12 @@ public class ConsumerConfigurationData implements Serializable, Cloneable { private int patternAutoDiscoveryPeriod = 1; + private int maxRedeliveryCount = 0; + + private String deadLetterTopic; + + private int maxUnackedMessagesPerConsumer = 0; + @JsonIgnore public String getSingleTopic() { checkArgument(topicNames.size() == 1); diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/tutorial/SubscriptionWithDeadLetter.java b/pulsar-client/src/test/java/org/apache/pulsar/client/tutorial/SubscriptionWithDeadLetter.java new file mode 100644 index 0000000000000..573899f77892f --- /dev/null +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/tutorial/SubscriptionWithDeadLetter.java @@ -0,0 +1,119 @@ +/** + * 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.tutorial; + +import org.apache.pulsar.client.api.*; + +import java.util.concurrent.TimeUnit; + +public class SubscriptionWithDeadLetter { + + private static final String topic = "persistent://public/default/my-topic"; + + private static final int maxRedeliveryCount = 2; + + private static final int sendMessages = 500; + + public static void main(String[] args) throws PulsarClientException { + + PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://127.0.0.1:6650").build(); + + Producer producer = client.newProducer(Schema.BYTES) + .topic(topic) + .create(); + + for (int i = 0; i < sendMessages; i++) { + producer.send("Hello Pulsar!".getBytes()); + } + + Consumer consumer = client.newConsumer(Schema.BYTES) + .topic(topic) + .subscriptionName("my-subscription") + .subscriptionType(SubscriptionType.Shared) + .ackTimeout(10, TimeUnit.SECONDS) + .maxRedeliveryCount(maxRedeliveryCount) + .receiverQueueSize(100) + .maxUnackedMessagesPerConsumer(100) + .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest) + .subscribe(); + + Consumer deadLetterConsumer = client.newConsumer(Schema.BYTES) + .topic("persistent://public/default/my-topic-my-subscription-DLQ") + .subscriptionName("my-subscription") + .subscribe(); + + int totalReceived = 0; + do { + Message msg = consumer.receive(); + totalReceived++; + System.out.println(new String(msg.getData())); + } while (totalReceived < sendMessages * (maxRedeliveryCount + 1)); + + int totalInDeadLetter = 0; + do { + Message msg = deadLetterConsumer.receive(); + totalInDeadLetter++; + System.out.println(new String(msg.getData())); + } while (totalInDeadLetter < sendMessages); + + deadLetterConsumer.close(); + consumer.close(); + + // test subscribe changed + for (int i = 0; i < sendMessages; i++) { + producer.send("Hello Pulsar!".getBytes()); + } + + Consumer consumer2 = client.newConsumer(Schema.BYTES) + .topic(topic) + .subscriptionName("my-subscription") + .subscriptionType(SubscriptionType.Shared) + .ackTimeout(10, TimeUnit.SECONDS) + .maxRedeliveryCount(maxRedeliveryCount) + .receiverQueueSize(100) + .maxUnackedMessagesPerConsumer(100) + .deadLetterTopic("persistent://public/default/my-topic-my-subscription-custom-DLQ") + .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest) + .subscribe(); + + Consumer deadLetterConsumer2 = client.newConsumer(Schema.BYTES) + .topic("persistent://public/default/my-topic-my-subscription-custom-DLQ") + .subscriptionName("my-subscription") + .subscribe(); + + int totalReceived2 = 0; + do { + Message msg = consumer2.receive(); + totalReceived2++; + System.out.println(new String(msg.getData())); + } while (totalReceived2 < sendMessages * (maxRedeliveryCount + 1)); + + int totalInDeadLetter2 = 0; + do { + Message msg = deadLetterConsumer2.receive(); + totalInDeadLetter2++; + System.out.println(new String(msg.getData())); + } while (totalInDeadLetter2 < sendMessages); + + deadLetterConsumer2.close(); + consumer2.close(); + producer.close(); + client.close(); + } +} diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/api/Commands.java b/pulsar-common/src/main/java/org/apache/pulsar/common/api/Commands.java index b5e2406c9ea27..0658fe1ce52a8 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/api/Commands.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/api/Commands.java @@ -318,12 +318,14 @@ public static ByteBufPair newSend(long producerId, long sequenceId, int numMessa public static ByteBuf newSubscribe(String topic, String subscription, long consumerId, long requestId, SubType subType, int priorityLevel, String consumerName) { return newSubscribe(topic, subscription, consumerId, requestId, subType, priorityLevel, consumerName, - true /* isDurable */, null /* startMessageId */, Collections.emptyMap(), false, InitialPosition.Earliest, null); + true /* isDurable */, null /* startMessageId */, Collections.emptyMap(), false, InitialPosition.Earliest, + null, 0, null, 0); } public static ByteBuf newSubscribe(String topic, String subscription, long consumerId, long requestId, SubType subType, int priorityLevel, String consumerName, boolean isDurable, MessageIdData startMessageId, - Map metadata, boolean readCompacted, InitialPosition subscriptionInitialPosition, SchemaInfo schemaInfo) { + Map metadata, boolean readCompacted, InitialPosition subscriptionInitialPosition, + SchemaInfo schemaInfo, int maxRedeliveryCount, String deadLetterTopic, int maxUnackedMessagesPerConsumer) { CommandSubscribe.Builder subscribeBuilder = CommandSubscribe.newBuilder(); subscribeBuilder.setTopic(topic); subscribeBuilder.setSubscription(subscription); @@ -335,6 +337,11 @@ public static ByteBuf newSubscribe(String topic, String subscription, long consu subscribeBuilder.setDurable(isDurable); subscribeBuilder.setReadCompacted(readCompacted); subscribeBuilder.setInitialPosition(subscriptionInitialPosition); + subscribeBuilder.setMaxRedeliveryCount(maxRedeliveryCount); + subscribeBuilder.setMaxUnackedMessagePerConsumer(maxUnackedMessagesPerConsumer); + if (deadLetterTopic != null) { + subscribeBuilder.setDeadLetterTopic(deadLetterTopic); + } if (startMessageId != null) { subscribeBuilder.setStartMessageId(startMessageId); } diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/api/proto/PulsarApi.java b/pulsar-common/src/main/java/org/apache/pulsar/common/api/proto/PulsarApi.java index edc7e9b118468..720d4da6d5456 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/api/proto/PulsarApi.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/api/proto/PulsarApi.java @@ -6722,6 +6722,18 @@ public interface CommandSubscribeOrBuilder // optional .pulsar.proto.CommandSubscribe.InitialPosition initialPosition = 13 [default = Latest]; boolean hasInitialPosition(); org.apache.pulsar.common.api.proto.PulsarApi.CommandSubscribe.InitialPosition getInitialPosition(); + + // optional int32 maxRedeliveryCount = 14; + boolean hasMaxRedeliveryCount(); + int getMaxRedeliveryCount(); + + // optional string deadLetterTopic = 15; + boolean hasDeadLetterTopic(); + String getDeadLetterTopic(); + + // optional int32 maxUnackedMessagePerConsumer = 16; + boolean hasMaxUnackedMessagePerConsumer(); + int getMaxUnackedMessagePerConsumer(); } public static final class CommandSubscribe extends org.apache.pulsar.shaded.com.google.protobuf.v241.GeneratedMessageLite @@ -7050,6 +7062,58 @@ public org.apache.pulsar.common.api.proto.PulsarApi.CommandSubscribe.InitialPosi return initialPosition_; } + // optional int32 maxRedeliveryCount = 14; + public static final int MAXREDELIVERYCOUNT_FIELD_NUMBER = 14; + private int maxRedeliveryCount_; + public boolean hasMaxRedeliveryCount() { + return ((bitField0_ & 0x00001000) == 0x00001000); + } + public int getMaxRedeliveryCount() { + return maxRedeliveryCount_; + } + + // optional string deadLetterTopic = 15; + public static final int DEADLETTERTOPIC_FIELD_NUMBER = 15; + private java.lang.Object deadLetterTopic_; + public boolean hasDeadLetterTopic() { + return ((bitField0_ & 0x00002000) == 0x00002000); + } + public String getDeadLetterTopic() { + java.lang.Object ref = deadLetterTopic_; + if (ref instanceof String) { + return (String) ref; + } else { + org.apache.pulsar.shaded.com.google.protobuf.v241.ByteString bs = + (org.apache.pulsar.shaded.com.google.protobuf.v241.ByteString) ref; + String s = bs.toStringUtf8(); + if (org.apache.pulsar.shaded.com.google.protobuf.v241.Internal.isValidUtf8(bs)) { + deadLetterTopic_ = s; + } + return s; + } + } + private org.apache.pulsar.shaded.com.google.protobuf.v241.ByteString getDeadLetterTopicBytes() { + java.lang.Object ref = deadLetterTopic_; + if (ref instanceof String) { + org.apache.pulsar.shaded.com.google.protobuf.v241.ByteString b = + org.apache.pulsar.shaded.com.google.protobuf.v241.ByteString.copyFromUtf8((String) ref); + deadLetterTopic_ = b; + return b; + } else { + return (org.apache.pulsar.shaded.com.google.protobuf.v241.ByteString) ref; + } + } + + // optional int32 maxUnackedMessagePerConsumer = 16; + public static final int MAXUNACKEDMESSAGEPERCONSUMER_FIELD_NUMBER = 16; + private int maxUnackedMessagePerConsumer_; + public boolean hasMaxUnackedMessagePerConsumer() { + return ((bitField0_ & 0x00004000) == 0x00004000); + } + public int getMaxUnackedMessagePerConsumer() { + return maxUnackedMessagePerConsumer_; + } + private void initFields() { topic_ = ""; subscription_ = ""; @@ -7064,6 +7128,9 @@ private void initFields() { readCompacted_ = false; schema_ = org.apache.pulsar.common.api.proto.PulsarApi.Schema.getDefaultInstance(); initialPosition_ = org.apache.pulsar.common.api.proto.PulsarApi.CommandSubscribe.InitialPosition.Latest; + maxRedeliveryCount_ = 0; + deadLetterTopic_ = ""; + maxUnackedMessagePerConsumer_ = 0; } private byte memoizedIsInitialized = -1; public final boolean isInitialized() { @@ -7159,6 +7226,15 @@ public void writeTo(org.apache.pulsar.common.util.protobuf.ByteBufCodedOutputStr if (((bitField0_ & 0x00000800) == 0x00000800)) { output.writeEnum(13, initialPosition_.getNumber()); } + if (((bitField0_ & 0x00001000) == 0x00001000)) { + output.writeInt32(14, maxRedeliveryCount_); + } + if (((bitField0_ & 0x00002000) == 0x00002000)) { + output.writeBytes(15, getDeadLetterTopicBytes()); + } + if (((bitField0_ & 0x00004000) == 0x00004000)) { + output.writeInt32(16, maxUnackedMessagePerConsumer_); + } } private int memoizedSerializedSize = -1; @@ -7219,6 +7295,18 @@ public int getSerializedSize() { size += org.apache.pulsar.shaded.com.google.protobuf.v241.CodedOutputStream .computeEnumSize(13, initialPosition_.getNumber()); } + if (((bitField0_ & 0x00001000) == 0x00001000)) { + size += org.apache.pulsar.shaded.com.google.protobuf.v241.CodedOutputStream + .computeInt32Size(14, maxRedeliveryCount_); + } + if (((bitField0_ & 0x00002000) == 0x00002000)) { + size += org.apache.pulsar.shaded.com.google.protobuf.v241.CodedOutputStream + .computeBytesSize(15, getDeadLetterTopicBytes()); + } + if (((bitField0_ & 0x00004000) == 0x00004000)) { + size += org.apache.pulsar.shaded.com.google.protobuf.v241.CodedOutputStream + .computeInt32Size(16, maxUnackedMessagePerConsumer_); + } memoizedSerializedSize = size; return size; } @@ -7358,6 +7446,12 @@ public Builder clear() { bitField0_ = (bitField0_ & ~0x00000800); initialPosition_ = org.apache.pulsar.common.api.proto.PulsarApi.CommandSubscribe.InitialPosition.Latest; bitField0_ = (bitField0_ & ~0x00001000); + maxRedeliveryCount_ = 0; + bitField0_ = (bitField0_ & ~0x00002000); + deadLetterTopic_ = ""; + bitField0_ = (bitField0_ & ~0x00004000); + maxUnackedMessagePerConsumer_ = 0; + bitField0_ = (bitField0_ & ~0x00008000); return this; } @@ -7444,6 +7538,18 @@ public org.apache.pulsar.common.api.proto.PulsarApi.CommandSubscribe buildPartia to_bitField0_ |= 0x00000800; } result.initialPosition_ = initialPosition_; + if (((from_bitField0_ & 0x00002000) == 0x00002000)) { + to_bitField0_ |= 0x00001000; + } + result.maxRedeliveryCount_ = maxRedeliveryCount_; + if (((from_bitField0_ & 0x00004000) == 0x00004000)) { + to_bitField0_ |= 0x00002000; + } + result.deadLetterTopic_ = deadLetterTopic_; + if (((from_bitField0_ & 0x00008000) == 0x00008000)) { + to_bitField0_ |= 0x00004000; + } + result.maxUnackedMessagePerConsumer_ = maxUnackedMessagePerConsumer_; result.bitField0_ = to_bitField0_; return result; } @@ -7496,6 +7602,15 @@ public Builder mergeFrom(org.apache.pulsar.common.api.proto.PulsarApi.CommandSub if (other.hasInitialPosition()) { setInitialPosition(other.getInitialPosition()); } + if (other.hasMaxRedeliveryCount()) { + setMaxRedeliveryCount(other.getMaxRedeliveryCount()); + } + if (other.hasDeadLetterTopic()) { + setDeadLetterTopic(other.getDeadLetterTopic()); + } + if (other.hasMaxUnackedMessagePerConsumer()) { + setMaxUnackedMessagePerConsumer(other.getMaxUnackedMessagePerConsumer()); + } return this; } @@ -7647,6 +7762,21 @@ public Builder mergeFrom( } break; } + case 112: { + bitField0_ |= 0x00002000; + maxRedeliveryCount_ = input.readInt32(); + break; + } + case 122: { + bitField0_ |= 0x00004000; + deadLetterTopic_ = input.readBytes(); + break; + } + case 128: { + bitField0_ |= 0x00008000; + maxUnackedMessagePerConsumer_ = input.readInt32(); + break; + } } } } @@ -8089,6 +8219,84 @@ public Builder clearInitialPosition() { return this; } + // optional int32 maxRedeliveryCount = 14; + private int maxRedeliveryCount_ ; + public boolean hasMaxRedeliveryCount() { + return ((bitField0_ & 0x00002000) == 0x00002000); + } + public int getMaxRedeliveryCount() { + return maxRedeliveryCount_; + } + public Builder setMaxRedeliveryCount(int value) { + bitField0_ |= 0x00002000; + maxRedeliveryCount_ = value; + + return this; + } + public Builder clearMaxRedeliveryCount() { + bitField0_ = (bitField0_ & ~0x00002000); + maxRedeliveryCount_ = 0; + + return this; + } + + // optional string deadLetterTopic = 15; + private java.lang.Object deadLetterTopic_ = ""; + public boolean hasDeadLetterTopic() { + return ((bitField0_ & 0x00004000) == 0x00004000); + } + public String getDeadLetterTopic() { + java.lang.Object ref = deadLetterTopic_; + if (!(ref instanceof String)) { + String s = ((org.apache.pulsar.shaded.com.google.protobuf.v241.ByteString) ref).toStringUtf8(); + deadLetterTopic_ = s; + return s; + } else { + return (String) ref; + } + } + public Builder setDeadLetterTopic(String value) { + if (value == null) { + throw new NullPointerException(); + } + bitField0_ |= 0x00004000; + deadLetterTopic_ = value; + + return this; + } + public Builder clearDeadLetterTopic() { + bitField0_ = (bitField0_ & ~0x00004000); + deadLetterTopic_ = getDefaultInstance().getDeadLetterTopic(); + + return this; + } + void setDeadLetterTopic(org.apache.pulsar.shaded.com.google.protobuf.v241.ByteString value) { + bitField0_ |= 0x00004000; + deadLetterTopic_ = value; + + } + + // optional int32 maxUnackedMessagePerConsumer = 16; + private int maxUnackedMessagePerConsumer_ ; + public boolean hasMaxUnackedMessagePerConsumer() { + return ((bitField0_ & 0x00008000) == 0x00008000); + } + public int getMaxUnackedMessagePerConsumer() { + return maxUnackedMessagePerConsumer_; + } + public Builder setMaxUnackedMessagePerConsumer(int value) { + bitField0_ |= 0x00008000; + maxUnackedMessagePerConsumer_ = value; + + return this; + } + public Builder clearMaxUnackedMessagePerConsumer() { + bitField0_ = (bitField0_ & ~0x00008000); + maxUnackedMessagePerConsumer_ = 0; + + return this; + } + // @@protoc_insertion_point(builder_scope:pulsar.proto.CommandSubscribe) } diff --git a/pulsar-common/src/main/proto/PulsarApi.proto b/pulsar-common/src/main/proto/PulsarApi.proto index 779a27bc1f027..7eea11ad56ee6 100644 --- a/pulsar-common/src/main/proto/PulsarApi.proto +++ b/pulsar-common/src/main/proto/PulsarApi.proto @@ -229,6 +229,16 @@ message CommandSubscribe { // Signal wthether the subscription will initialize on latest // or not -- earliest optional InitialPosition initialPosition = 13 [default = Latest]; + // Maximum number of redeliveries. + // Message exceeding the maximum number of redeliveries should send to Dead Letter Topic and acknowledged automatic. + // Enable this feature by set maxRedeliveryCount > 0 + optional int32 maxRedeliveryCount = 14; + // Name of Dead Letter Topic. + // If not set, pulsar broker will generate with topic name and subscription name and suffix with -DLQ + optional string deadLetterTopic = 15; + //Maximum number of un-acked messages count per consumer. + //This config should less than broker config, if not, config will not enable. + optional int32 maxUnackedMessagePerConsumer = 16; } message CommandPartitionedTopicMetadata {