From a60156f6d31cd5ed62f41c9ab08fe049af8c3f13 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Mon, 1 Jun 2020 10:53:44 -0700 Subject: [PATCH 1/6] PIP 68: Exclusive Producer --- .../pulsar/broker/service/AbstractTopic.java | 144 ++++++++++- .../service/BrokerServiceException.java | 8 + .../pulsar/broker/service/Producer.java | 19 +- .../broker/service/PulsarCommandSender.java | 5 +- .../service/PulsarCommandSenderImpl.java | 8 +- .../pulsar/broker/service/ServerCnx.java | 64 ++--- .../apache/pulsar/broker/service/Topic.java | 8 +- .../NonPersistentSubscription.java | 4 +- .../nonpersistent/NonPersistentTopic.java | 63 ++--- .../persistent/PersistentSubscription.java | 4 +- .../service/persistent/PersistentTopic.java | 106 ++++----- .../pulsar/broker/service/BrokerTestBase.java | 8 + .../broker/service/ExclusiveProducerTest.java | 190 +++++++++++++++ .../broker/service/PersistentTopicTest.java | 63 ++--- .../pulsar/client/api/ProducerAccessMode.java | 41 ++++ .../pulsar/client/api/ProducerBuilder.java | 19 ++ .../client/api/PulsarClientException.java | 18 ++ .../apache/pulsar/client/impl/ClientCnx.java | 8 +- .../pulsar/client/impl/ConnectionHandler.java | 1 + .../pulsar/client/impl/HandlerState.java | 3 +- .../client/impl/PartitionedProducerImpl.java | 2 + .../client/impl/ProducerBuilderImpl.java | 7 + .../pulsar/client/impl/ProducerImpl.java | 19 +- .../pulsar/client/impl/ProducerResponse.java | 4 + .../impl/conf/ProducerConfigurationData.java | 3 + .../pulsar/common/api/proto/PulsarApi.java | 225 ++++++++++++++++++ .../pulsar/common/protocol/Commands.java | 34 ++- pulsar-common/src/main/proto/PulsarApi.proto | 26 ++ .../testclient/PerformanceProducer.java | 5 + 29 files changed, 927 insertions(+), 182 deletions(-) create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ExclusiveProducerTest.java create mode 100644 pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerAccessMode.java diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java index 341193e4f44e5..7018bb80dcca0 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java @@ -18,22 +18,30 @@ */ package org.apache.pulsar.broker.service; +import static com.google.common.base.Preconditions.checkArgument; import static org.apache.bookkeeper.mledger.impl.ManagedLedgerMBeanImpl.ENTRY_LATENCY_BUCKETS_USEC; import static org.apache.pulsar.broker.cache.ConfigurationCacheService.POLICIES; import com.google.common.base.MoreObjects; + import java.util.Map; import java.util.Objects; +import java.util.Optional; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLongFieldUpdater; import java.util.concurrent.atomic.LongAdder; import java.util.concurrent.locks.ReentrantLock; import java.util.concurrent.locks.ReentrantReadWriteLock; + import org.apache.bookkeeper.mledger.util.StatsBuckets; import org.apache.pulsar.broker.ServiceConfiguration; import org.apache.pulsar.broker.admin.AdminResource; import org.apache.pulsar.broker.service.BrokerServiceException.ConsumerBusyException; +import org.apache.pulsar.broker.service.BrokerServiceException.ProducerBusyException; +import org.apache.pulsar.broker.service.BrokerServiceException.ProducerFencedException; +import org.apache.pulsar.broker.service.BrokerServiceException.TopicTerminatedException; import org.apache.pulsar.broker.service.schema.SchemaRegistryService; import org.apache.pulsar.broker.service.schema.exceptions.IncompatibleSchemaException; import org.apache.pulsar.broker.stats.prometheus.metrics.Summary; @@ -101,6 +109,13 @@ public abstract class AbstractTopic implements Topic { protected CompletableFuture transactionBuffer; protected ReentrantLock transactionBufferLock = new ReentrantLock(); + protected volatile Optional topicEpoch = Optional.empty(); + private volatile boolean hasExclusiveProducer; + + private static final AtomicLongFieldUpdater USAGE_COUNT_UPDATER = + AtomicLongFieldUpdater.newUpdater(AbstractTopic.class, "usageCount"); + private volatile long usageCount = 0; + public AbstractTopic(String topic, BrokerService brokerService) { this.topic = topic; this.brokerService = brokerService; @@ -316,6 +331,94 @@ public CompletableFuture checkSchemaCompatibleForConsumer(SchemaData schem .checkConsumerCompatibility(id, schema, schemaCompatibilityStrategy); } + @Override + public CompletableFuture> addProducer(Producer producer) { + checkArgument(producer.getTopic() == this); + + CompletableFuture> future = new CompletableFuture<>(); + + incrementTopicEpochIfNeeded(producer) + .thenAccept(epoch -> { + lock.readLock().lock(); + try { + brokerService.checkTopicNsOwnership(getName()); + checkTopicFenced(); + if (isTerminated()) { + log.warn("[{}] Attempting to add producer to a terminated topic", topic); + throw new TopicTerminatedException("Topic was already terminated"); + } + internalAddProducer(producer); + + USAGE_COUNT_UPDATER.incrementAndGet(this); + if (log.isDebugEnabled()) { + log.debug("[{}] [{}] Added producer -- count: {}", topic, producer.getProducerName(), + USAGE_COUNT_UPDATER.get(this)); + } + + future.complete(epoch); + } catch (Throwable e) { + future.completeExceptionally(e); + } finally { + lock.readLock().unlock(); + } + }).exceptionally(ex -> { + future.completeExceptionally(ex); + return null; + }); + + return future; + } + + protected CompletableFuture> incrementTopicEpochIfNeeded(Producer producer) { + lock.writeLock().lock(); + try { + switch (producer.getAccessMode()) { + case Shared: + if (hasExclusiveProducer) { + return FutureUtil.failedFuture(new ProducerBusyException("Topic has an existing exclusive producer")); + } else { + // Normal producer getting added, we don't need a new epoch + return CompletableFuture.completedFuture(topicEpoch); + } + + case Exclusive: + if (hasExclusiveProducer || !producers.isEmpty()) { + return FutureUtil.failedFuture(new ProducerFencedException("Topic has existing producers")); + } else if (producer.getTopicEpoch().isPresent() && producer.getTopicEpoch().get() < topicEpoch.orElse(-1L)){ + // If a producer reconnects, but all the topic epoch has already moved forward, this producer needs to + // be fenced, because a new producer had been present in between. + return FutureUtil.failedFuture(new ProducerFencedException("Topic epoch has already moved")); + } else { + // There are currently no existing producers + hasExclusiveProducer = true; + + CompletableFuture> future = incrementTopicEpoch(topicEpoch).thenApply(epoch -> { + topicEpoch = Optional.of(epoch); + return topicEpoch; + }); + + future.exceptionally(ex -> { + hasExclusiveProducer = false; + return null; + }); + return future; + } + + // case WaitForExclusive: + // TODO: Implementation + + default: + return FutureUtil.failedFuture( + new BrokerServiceException("Invalid producer access mode: " + producer.getAccessMode())); + } + + } finally { + lock.writeLock().unlock(); + } + } + + protected abstract CompletableFuture incrementTopicEpoch(Optional currentEpoch); + @Override public void recordAddLatency(long latency, TimeUnit unit) { addEntryLatencyStatsUsec.addValue(unit.toMicros(latency)); @@ -450,7 +553,44 @@ private boolean isUserProvidedProducerName(Producer producer){ return producer.isUserProvidedProducerName() && !producer.getProducerName().startsWith(replicatorPrefix); } - protected abstract void handleProducerRemoved(Producer producer); + + @Override + public void removeProducer(Producer producer) { + checkArgument(producer.getTopic() == this); + + if (producers.remove(producer.getProducerName(), producer)) { + handleProducerRemoved(producer); + } + } + + protected void handleProducerRemoved(Producer producer) { + // decrement usage only if this was a valid producer close + long newCount = USAGE_COUNT_UPDATER.decrementAndGet(this); + if (newCount == 0) { + hasExclusiveProducer = false; + } + if (log.isDebugEnabled()) { + log.debug("[{}] [{}] Removed producer -- count: {}", topic, producer.getProducerName(), + USAGE_COUNT_UPDATER.get(this)); + } + lastActive = System.nanoTime(); + } + + public void handleConsumerAdded(String subscriptionName, String consumerName) { + USAGE_COUNT_UPDATER.incrementAndGet(this); + if (log.isDebugEnabled()) { + log.debug("[{}] [{}] [{}] Added consumer -- count: {}", topic, subscriptionName, + consumerName, USAGE_COUNT_UPDATER.get(this)); + } + } + + public void decrementUsageCount() { + USAGE_COUNT_UPDATER.decrementAndGet(this); + } + + public long currentUsageCount() { + return usageCount; + } @Override public boolean isPublishRateExceeded() { @@ -536,6 +676,8 @@ public void setDeleteWhileInactive(boolean deleteWhileInactive) { this.inactiveTopicPolicies.setDeleteWhileInactive(deleteWhileInactive); } + protected abstract boolean isTerminated(); + private static final Logger log = LoggerFactory.getLogger(AbstractTopic.class); public InactiveTopicPolicies getInactiveTopicPolicies() { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerServiceException.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerServiceException.java index 9c6b2d1db9019..75d8b39fdc11b 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerServiceException.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerServiceException.java @@ -56,6 +56,12 @@ public ProducerBusyException(String msg) { } } + public static class ProducerFencedException extends BrokerServiceException { + public ProducerFencedException(String msg) { + super(msg); + } + } + public static class ServiceUnitNotReadyException extends BrokerServiceException { public ServiceUnitNotReadyException(String msg) { super(msg); @@ -217,6 +223,8 @@ private static PulsarApi.ServerError getClientErrorCode(Throwable t, boolean che return ServerError.InvalidTxnStatus; } else if (t instanceof NotAllowedException) { return ServerError.NotAllowedError; + } else if (t instanceof ProducerFencedException) { + return ServerError.ProducerFenced; } else if (t instanceof TransactionConflictException) { return ServerError.TransactionConflict; } else if (t instanceof CoordinatorException.TransactionNotFoundException) { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java index 012e7f7edd8b0..6602814bcf339 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java @@ -31,6 +31,7 @@ import java.util.Collections; import java.util.Map; import java.util.Objects; +import java.util.Optional; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLongFieldUpdater; @@ -42,6 +43,7 @@ import org.apache.pulsar.broker.service.persistent.PersistentTopic; import org.apache.pulsar.client.api.transaction.TxnID; import org.apache.pulsar.common.api.proto.PulsarApi.MessageMetadata; +import org.apache.pulsar.common.api.proto.PulsarApi.ProducerAccessMode; import org.apache.pulsar.common.api.proto.PulsarApi.ServerError; import org.apache.pulsar.common.naming.TopicName; import org.apache.pulsar.common.policies.data.NonPersistentPublisherStats; @@ -83,13 +85,18 @@ public class Producer { private final boolean isNonPersistentTopic; private final boolean isEncrypted; + private final ProducerAccessMode accessMode; + private Optional topicEpoch; + private final Map metadata; private final SchemaVersion schemaVersion; public Producer(Topic topic, TransportCnx cnx, long producerId, String producerName, String appId, boolean isEncrypted, Map metadata, SchemaVersion schemaVersion, long epoch, - boolean userProvidedProducerName) { + boolean userProvidedProducerName, + ProducerAccessMode accessMode, + Optional topicEpoch) { this.topic = topic; this.cnx = cnx; this.producerId = producerId; @@ -124,6 +131,8 @@ public Producer(Topic topic, TransportCnx cnx, long producerId, String producerN this.isEncrypted = isEncrypted; this.schemaVersion = schemaVersion; + this.accessMode = accessMode; + this.topicEpoch = topicEpoch; } @Override @@ -635,6 +644,14 @@ public SchemaVersion getSchemaVersion() { return schemaVersion; } + public ProducerAccessMode getAccessMode() { + return accessMode; + } + + public Optional getTopicEpoch() { + return topicEpoch; + } + private static final Logger log = LoggerFactory.getLogger(Producer.class); } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarCommandSender.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarCommandSender.java index 0a06a7fdec843..d1ad3d39e0a62 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarCommandSender.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarCommandSender.java @@ -19,7 +19,10 @@ package org.apache.pulsar.broker.service; import io.netty.util.concurrent.Future; + import java.util.List; +import java.util.Optional; + import org.apache.bookkeeper.mledger.Entry; import org.apache.pulsar.common.api.proto.PulsarApi; import org.apache.pulsar.common.protocol.schema.SchemaVersion; @@ -38,7 +41,7 @@ public interface PulsarCommandSender { void sendProducerSuccessResponse(long requestId, String producerName, SchemaVersion schemaVersion); void sendProducerSuccessResponse(long requestId, String producerName, long lastSequenceId, - SchemaVersion schemaVersion); + SchemaVersion schemaVersion, Optional topicEpoch); void sendSendReceiptResponse(long producerId, long sequenceId, long highestId, long ledgerId, long entryId); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarCommandSenderImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarCommandSenderImpl.java index c2797d43fc63a..ad0b828f2d358 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarCommandSenderImpl.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarCommandSenderImpl.java @@ -22,8 +22,12 @@ import io.netty.buffer.Unpooled; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelPromise; + import java.util.List; +import java.util.Optional; + import lombok.extern.slf4j.Slf4j; + import org.apache.bookkeeper.mledger.Entry; import org.apache.bookkeeper.mledger.impl.PositionImpl; import org.apache.pulsar.broker.intercept.BrokerInterceptor; @@ -97,9 +101,9 @@ public void sendProducerSuccessResponse(long requestId, String producerName, Sch @Override public void sendProducerSuccessResponse(long requestId, String producerName, long lastSequenceId, - SchemaVersion schemaVersion) { + SchemaVersion schemaVersion, Optional topicEpoch) { PulsarApi.BaseCommand command = Commands.newProducerSuccessCommand(requestId, producerName, lastSequenceId, - schemaVersion); + schemaVersion, topicEpoch); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); command.getProducerSuccess().recycle(); 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 b5bbb847b9ccc..ffdfe8176ed6d 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 @@ -37,6 +37,7 @@ import java.util.List; import java.util.Map; import java.util.NoSuchElementException; +import java.util.Optional; import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.Semaphore; @@ -103,6 +104,7 @@ import org.apache.pulsar.common.api.proto.PulsarApi.FeatureFlags; import org.apache.pulsar.common.api.proto.PulsarApi.MessageIdData; import org.apache.pulsar.common.api.proto.PulsarApi.MessageMetadata; +import org.apache.pulsar.common.api.proto.PulsarApi.ProducerAccessMode; import org.apache.pulsar.common.api.proto.PulsarApi.ProtocolVersion; import org.apache.pulsar.common.api.proto.PulsarApi.ServerError; import org.apache.pulsar.common.intercept.InterceptException; @@ -1022,6 +1024,10 @@ protected void handleProducer(final CommandProducer cmdProducer) { final boolean isEncrypted = cmdProducer.getEncrypted(); final Map metadata = CommandUtils.metadataFromCommand(cmdProducer); final SchemaData schema = cmdProducer.hasSchema() ? getSchema(cmdProducer.getSchema()) : null; + + final ProducerAccessMode producerAccessMode = cmdProducer.getProducerAccessMode(); + final Optional topicEpoch = cmdProducer.hasTopicEpoch() ? Optional.of(cmdProducer.getTopicEpoch()) : Optional.empty(); + TopicName topicName = validateTopicName(cmdProducer.getTopic(), requestId, cmdProducer); if (topicName == null) { return; @@ -1124,40 +1130,40 @@ protected void handleProducer(final CommandProducer cmdProducer) { schemaVersionFuture.thenAccept(schemaVersion -> { Producer producer = new Producer(topic, ServerCnx.this, producerId, producerName, getPrincipal(), - isEncrypted, metadata, schemaVersion, epoch, userProvidedProducerName); - - try { - topic.addProducer(producer); - - if (isActive()) { - if (producerFuture.complete(producer)) { - log.info("[{}] Created new producer: {}", remoteAddress, producer); - commandSender.sendProducerSuccessResponse(requestId, producerName, - producer.getLastSequenceId(), producer.getSchemaVersion()); - return; + isEncrypted, metadata, schemaVersion, epoch, userProvidedProducerName, producerAccessMode, topicEpoch); + + topic.addProducer(producer).thenAccept(newTopicEpoch -> { + if (isActive()) { + if (producerFuture.complete(producer)) { + log.info("[{}] Created new producer: {}", remoteAddress, producer); + commandSender.sendProducerSuccessResponse(requestId, producerName, + producer.getLastSequenceId(), producer.getSchemaVersion(), newTopicEpoch); + return; + } else { + // The producer's future was completed before by + // a close command + producer.closeNow(true); + log.info("[{}] Cleared producer created after timeout on client side {}", + remoteAddress, producer); + } } else { - // The producer's future was completed before by - // a close command producer.closeNow(true); - log.info("[{}] Cleared producer created after timeout on client side {}", + log.info("[{}] Cleared producer created after connection was closed: {}", remoteAddress, producer); + producerFuture.completeExceptionally( + new IllegalStateException("Producer created after connection was closed")); } - } else { - producer.closeNow(true); - log.info("[{}] Cleared producer created after connection was closed: {}", - remoteAddress, producer); - producerFuture.completeExceptionally( - new IllegalStateException("Producer created after connection was closed")); - } - } catch (Exception ise) { - log.error("[{}] Failed to add producer to topic {}: {}", remoteAddress, topicName, - ise.getMessage()); - commandSender.sendErrorResponse(requestId, - BrokerServiceException.getClientErrorCode(ise), ise.getMessage()); - producerFuture.completeExceptionally(ise); - } - producers.remove(producerId, producerFuture); + producers.remove(producerId, producerFuture); + }).exceptionally(ex -> { + log.warn("[{}] Failed to add producer {}: {}", remoteAddress, producer, ex.getMessage()); + producer.closeNow(true); + if (producerFuture.completeExceptionally(ex)) { + commandSender.sendErrorResponse(requestId, + BrokerServiceException.getClientErrorCode(ex), ex.getMessage()); + } + return null; + }); }); }).exceptionally(exception -> { Throwable cause = exception.getCause(); 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 c2d88f7b266cc..8f86c242c238a 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 @@ -94,7 +94,13 @@ default long getOriginalHighestSequenceId() { void publishMessage(ByteBuf headersAndPayload, PublishContext callback); - void addProducer(Producer producer) throws BrokerServiceException; + /** + * Tries to add a producer to the topic. Several validations will be performed. + * + * @param producer + * @return the "topic epoch" if there is one or empty + */ + CompletableFuture> addProducer(Producer producer); void removeProducer(Producer producer); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java index 8a84f53e1bfca..a4d2d6f7ce9a5 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java @@ -184,10 +184,10 @@ public synchronized void removeConsumer(Consumer consumer, boolean isResetCursor // invalid consumer remove will throw an exception // decrement usage is triggered only for valid consumer close - NonPersistentTopic.USAGE_COUNT_UPDATER.decrementAndGet(topic); + topic.decrementUsageCount(); if (log.isDebugEnabled()) { log.debug("[{}] [{}] [{}] Removed consumer -- count: {}", topic.getName(), subName, consumer.consumerName(), - NonPersistentTopic.USAGE_COUNT_UPDATER.get(topic)); + topic.currentUsageCount()); } } 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 551a9cc0779f2..56875a7892584 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 @@ -29,6 +29,7 @@ import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; @@ -92,10 +93,6 @@ public class NonPersistentTopic extends AbstractTopic implements Topic { private final ConcurrentOpenHashMap replicators; - protected static final AtomicLongFieldUpdater USAGE_COUNT_UPDATER = AtomicLongFieldUpdater - .newUpdater(NonPersistentTopic.class, "usageCount"); - private volatile long usageCount = 0; - // Ever increasing counter of entries added private static final AtomicLongFieldUpdater ENTRIES_ADDED_COUNTER_UPDATER = AtomicLongFieldUpdater .newUpdater(NonPersistentTopic.class, "entriesAddedCounter"); @@ -136,7 +133,6 @@ public NonPersistentTopic(String topic, BrokerService brokerService) { this.subscriptions = new ConcurrentOpenHashMap<>(16, 1); this.replicators = new ConcurrentOpenHashMap<>(16, 1); this.isFenced = false; - USAGE_COUNT_UPDATER.set(this, 0); try { Policies policies = brokerService.pulsar().getConfigurationCache().policiesCache() @@ -193,27 +189,10 @@ public void publishMessage(ByteBuf data, PublishContext callback) { } } - @Override - public void addProducer(Producer producer) throws BrokerServiceException { - checkArgument(producer.getTopic() == this); - - lock.readLock().lock(); - try { - brokerService.checkTopicNsOwnership(getName()); - - checkTopicFenced(); - - internalAddProducer(producer); - - USAGE_COUNT_UPDATER.incrementAndGet(this); - if (log.isDebugEnabled()) { - log.debug("[{}] [{}] Added producer -- count: {}", topic, producer.getProducerName(), - USAGE_COUNT_UPDATER.get(this)); - } - - } finally { - lock.readLock().unlock(); - } + protected CompletableFuture incrementTopicEpoch(Optional currentEpoch) { + // Non-persistent topic does not have any durable metadata, so we're just + // keeping the epoch in memory + return CompletableFuture.completedFuture(currentEpoch.orElse(-1L) + 1); } @Override @@ -229,17 +208,6 @@ public void removeProducer(Producer producer) { } } - @Override - public void handleProducerRemoved(Producer producer) { - // decrement usage only if this was a valid producer close - USAGE_COUNT_UPDATER.decrementAndGet(this); - if (log.isDebugEnabled()) { - log.debug("[{}] [{}] Removed producer -- count: {}", topic, producer.getProducerName(), - USAGE_COUNT_UPDATER.get(this)); - } - lastActive = System.nanoTime(); - } - @Override public CompletableFuture subscribe(final TransportCnx cnx, String subscriptionName, long consumerId, SubType subType, int priorityLevel, String consumerName, boolean isDurable, MessageId startMessageId, @@ -281,11 +249,8 @@ public CompletableFuture subscribe(final TransportCnx cnx, String subs future.completeExceptionally(new TopicFencedException("Topic is temporarily unavailable")); return future; } - USAGE_COUNT_UPDATER.incrementAndGet(this); - if (log.isDebugEnabled()) { - log.debug("[{}] [{}] [{}] Added consumer -- count: {}", topic, subscriptionName, consumerName, - USAGE_COUNT_UPDATER.get(this)); - } + + handleConsumerAdded(subscriptionName, consumerName); } finally { lock.readLock().unlock(); } @@ -301,7 +266,7 @@ public CompletableFuture subscribe(final TransportCnx cnx, String subs consumer.close(); if (log.isDebugEnabled()) { log.debug("[{}] [{}] [{}] Subscribe failed -- count: {}", topic, subscriptionName, - consumer.consumerName(), USAGE_COUNT_UPDATER.get(NonPersistentTopic.this)); + consumer.consumerName(), currentUsageCount()); } future.completeExceptionally( new BrokerServiceException("Connection was closed while the opening the cursor ")); @@ -317,7 +282,7 @@ public CompletableFuture subscribe(final TransportCnx cnx, String subs log.warn("[{}][{}] {}", topic, subscriptionName, e.getMessage()); } - USAGE_COUNT_UPDATER.decrementAndGet(NonPersistentTopic.this); + decrementUsageCount(); future.completeExceptionally(e); } @@ -377,7 +342,7 @@ private CompletableFuture delete(boolean failIfHasSubscriptions, boolean c closeClientFuture.thenAccept(delete -> { - if (USAGE_COUNT_UPDATER.get(this) == 0) { + if (currentUsageCount() == 0) { isFenced = true; List> futures = Lists.newArrayList(); @@ -411,7 +376,7 @@ private CompletableFuture delete(boolean failIfHasSubscriptions, boolean c }); } else { deleteFuture.completeExceptionally(new TopicBusyException( - "Topic has " + USAGE_COUNT_UPDATER.get(this) + " connected producers/consumers")); + "Topic has " + currentUsageCount() + " connected producers/consumers")); } }).exceptionally(ex -> { deleteFuture.completeExceptionally( @@ -836,7 +801,7 @@ public boolean isActive() { // No local consumers and no local producers return !subscriptions.isEmpty() || hasLocalProducers(); } - return USAGE_COUNT_UPDATER.get(this) != 0 || !subscriptions.isEmpty(); + return currentUsageCount() != 0 || !subscriptions.isEmpty(); } @Override @@ -1019,4 +984,8 @@ public CompletableFuture endTxn(TxnID txnID, int txnAction, List replicators; - protected static final AtomicLongFieldUpdater USAGE_COUNT_UPDATER = - AtomicLongFieldUpdater.newUpdater(PersistentTopic.class, "usageCount"); - @SuppressWarnings("unused") - private volatile long usageCount = 0; - static final String DEDUPLICATION_CURSOR_NAME = "pulsar.dedup"; private static final double MESSAGE_EXPIRY_THRESHOLD = 1.5; @@ -226,7 +223,6 @@ public PersistentTopic(String topic, ManagedLedger ledger, BrokerService brokerS this.ledger = ledger; this.subscriptions = new ConcurrentOpenHashMap<>(16, 1); this.replicators = new ConcurrentOpenHashMap<>(16, 1); - USAGE_COUNT_UPDATER.set(this, 0); this.delayedDeliveryEnabled = brokerService.pulsar().getConfiguration().isDelayedDeliveryEnabled(); this.delayedDeliveryTickTimeMillis = brokerService.pulsar().getConfiguration().getDelayedDeliveryTickTimeMillis(); this.backloggedCursorThresholdEntries = brokerService.pulsar().getConfiguration().getManagedLedgerCursorBackloggedThreshold(); @@ -281,6 +277,10 @@ public PersistentTopic(String topic, ManagedLedger ledger, BrokerService brokerS isEncryptionRequired = false; } + if (ledger.getProperties().containsKey(TOPIC_EPOCH_PROPERTY_NAME)) { + topicEpoch = Optional.of(Long.parseLong(ledger.getProperties().get(TOPIC_EPOCH_PROPERTY_NAME))); + } + checkReplicatedSubscriptionControllerState(); } // for testing purposes @@ -465,33 +465,37 @@ public synchronized void addFailed(ManagedLedgerException exception, Object ctx) } @Override - public void addProducer(Producer producer) throws BrokerServiceException { - checkArgument(producer.getTopic() == this); + public CompletableFuture> addProducer(Producer producer) { + return super.addProducer(producer).thenApply(epoch -> { + messageDeduplication.producerAdded(producer.getProducerName()); - lock.readLock().lock(); - try { - brokerService.checkTopicNsOwnership(getName()); + // Start replication producers if not already + startReplProducers(); + return epoch; + }); + } - checkTopicFenced(); + private static final String TOPIC_EPOCH_PROPERTY_NAME = "pulsar.topic.epoch"; - if (ledger.isTerminated()) { - log.warn("[{}] Attempting to add producer to a terminated topic", topic); - throw new TopicTerminatedException("Topic was already terminated"); - } + protected CompletableFuture incrementTopicEpoch(Optional currentEpoch) { + long newEpoch = currentEpoch.orElse(-1L) + 1; - internalAddProducer(producer); + CompletableFuture future = new CompletableFuture<>(); + ledger.asyncSetProperty(TOPIC_EPOCH_PROPERTY_NAME, String.valueOf(newEpoch), new UpdatePropertiesCallback() { + @Override + public void updatePropertiesComplete(Map properties, Object ctx) { + log.info("[{}] Updated topic epoch to {}", getName(), newEpoch); + future.complete(newEpoch); + } - USAGE_COUNT_UPDATER.incrementAndGet(this); - if (log.isDebugEnabled()) { - log.debug("[{}] [{}] Added producer -- count: {}", topic, producer.getProducerName(), USAGE_COUNT_UPDATER.get(this)); + @Override + public void updatePropertiesFailed(ManagedLedgerException exception, Object ctx) { + log.warn("[{}] Failed to update topic epoch to {}: {}", getName(), newEpoch, exception.getMessage()); + future.completeExceptionally(exception); } - messageDeduplication.producerAdded(producer.getProducerName()); + }, null); - // Start replication producers if not already - startReplProducers(); - } finally { - lock.readLock().unlock(); - } + return future; } private boolean hasRemoteProducers() { @@ -539,25 +543,9 @@ private synchronized CompletableFuture closeReplProducersIfNoBacklog() { return FutureUtil.waitForAll(closeFutures); } - @Override - public void removeProducer(Producer producer) { - checkArgument(producer.getTopic() == this); - - if (producers.remove(producer.getProducerName(), producer)) { - handleProducerRemoved(producer); - } - } - @Override protected void handleProducerRemoved(Producer producer) { - // decrement usage only if this was a valid producer close - USAGE_COUNT_UPDATER.decrementAndGet(this); - if (log.isDebugEnabled()) { - log.debug("[{}] [{}] Removed producer -- count: {}", topic, producer.getProducerName(), - USAGE_COUNT_UPDATER.get(this)); - } - lastActive = System.nanoTime(); - + super.handleProducerRemoved(producer); messageDeduplication.producerRemoved(producer.getProducerName()); } @@ -639,11 +627,7 @@ public CompletableFuture subscribe(final TransportCnx cnx, String subs future.completeExceptionally(new TopicFencedException("Topic is temporarily unavailable")); return future; } - USAGE_COUNT_UPDATER.incrementAndGet(this); - if (log.isDebugEnabled()) { - log.debug("[{}] [{}] [{}] Added consumer -- count: {}", topic, subscriptionName, consumerName, - USAGE_COUNT_UPDATER.get(this)); - } + handleConsumerAdded(subscriptionName, consumerName); } finally { lock.readLock().unlock(); } @@ -668,9 +652,10 @@ public CompletableFuture subscribe(final TransportCnx cnx, String subs consumer.close(); if (log.isDebugEnabled()) { log.debug("[{}] [{}] [{}] Subscribe failed -- count: {}", topic, subscriptionName, - consumer.consumerName(), USAGE_COUNT_UPDATER.get(PersistentTopic.this)); + consumer.consumerName(), currentUsageCount()); } - USAGE_COUNT_UPDATER.decrementAndGet(PersistentTopic.this); + + decrementUsageCount(); future.completeExceptionally( new BrokerServiceException("Connection was closed while the opening the cursor ")); } else { @@ -686,12 +671,12 @@ public CompletableFuture subscribe(final TransportCnx cnx, String subs log.warn("[{}][{}] {}", topic, subscriptionName, e.getMessage()); } - USAGE_COUNT_UPDATER.decrementAndGet(PersistentTopic.this); + decrementUsageCount(); future.completeExceptionally(e); } }).exceptionally(ex -> { log.error("[{}] Failed to create subscription: {} error: {}", topic, subscriptionName, ex); - USAGE_COUNT_UPDATER.decrementAndGet(PersistentTopic.this); + decrementUsageCount(); future.completeExceptionally(new PersistenceException(ex)); return null; }); @@ -747,7 +732,7 @@ public void openCursorComplete(ManagedCursor cursor, Object ctx) { @Override public void openCursorFailed(ManagedLedgerException exception, Object ctx) { log.warn("[{}] Failed to create subscription for {}: {}", topic, subscriptionName, exception.getMessage()); - USAGE_COUNT_UPDATER.decrementAndGet(PersistentTopic.this); + decrementUsageCount(); subscriptionFuture.completeExceptionally(new PersistenceException(exception)); if (exception instanceof ManagedLedgerFencedException) { // If the managed ledger has been fenced, we cannot continue using it. We need to close and reopen @@ -948,7 +933,7 @@ private CompletableFuture delete(boolean failIfHasSubscriptions, // 1. No one is connected // 2. We want to kick out everyone and forcefully delete the topic. // In this case, we shouldn't care if the usageCount is 0 or not, just proceed - if (USAGE_COUNT_UPDATER.get(this) == 0 || (closeIfClientsConnected && !failIfHasSubscriptions)) { + if (currentUsageCount() == 0 || (closeIfClientsConnected && !failIfHasSubscriptions)) { CompletableFuture deleteSchemaFuture = deleteSchema ? deleteSchema() : CompletableFuture.completedFuture(null); @@ -990,7 +975,7 @@ public void deleteLedgerFailed(ManagedLedgerException exception, Object ctx) { } else { unfenceTopicToResume(); deleteFuture.completeExceptionally(new TopicBusyException( - "Topic has " + USAGE_COUNT_UPDATER.get(this) + " connected producers/consumers")); + "Topic has " + currentUsageCount() + " connected producers/consumers")); } }).exceptionally(ex->{ unfenceTopicToResume(); @@ -1748,7 +1733,7 @@ public boolean isActive(InactiveTopicDeleteMode deleteMode) { // no local producers return hasLocalProducers(); } else { - return USAGE_COUNT_UPDATER.get(this) != 0; + return currentUsageCount() != 0; } } @@ -2668,4 +2653,9 @@ private boolean checkMaxSubscriptionsPerTopicExceed() { return false; } + + @Override + protected boolean isTerminated() { + return ledger.isTerminated(); + } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerTestBase.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerTestBase.java index bbaafed0b4cdb..ea7ddf8b11f05 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerTestBase.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerTestBase.java @@ -28,6 +28,8 @@ import com.google.common.collect.Sets; +import java.util.Random; + /** */ public abstract class BrokerTestBase extends MockedPulsarServiceBaseTest { @@ -86,5 +88,11 @@ void runMessageExpiryCheck() { } } + private static final Random random = new Random(); + + protected String newTopicName() { + return "prop/ns-abc/topic-" + Long.toHexString(random.nextLong()); + } + private static final Logger LOG = LoggerFactory.getLogger(BrokerTestBase.class); } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ExclusiveProducerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ExclusiveProducerTest.java new file mode 100644 index 0000000000000..4157fedc54064 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ExclusiveProducerTest.java @@ -0,0 +1,190 @@ +/** + * 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 static org.testng.Assert.fail; + +import java.util.Optional; +import java.util.concurrent.CompletableFuture; + +import lombok.Cleanup; + +import org.apache.pulsar.client.api.Producer; +import org.apache.pulsar.client.api.ProducerAccessMode; +import org.apache.pulsar.client.api.PulsarClientException.ProducerBusyException; +import org.apache.pulsar.client.api.PulsarClientException.ProducerFencedException; +import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.common.naming.TopicName; +import org.powermock.reflect.Whitebox; +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.DataProvider; +import org.testng.annotations.Test; + +public class ExclusiveProducerTest extends BrokerTestBase { + + @BeforeClass + protected void setup() throws Exception { + baseSetup(); + } + + @AfterClass(alwaysRun = true) + protected void cleanup() throws Exception { + internalCleanup(); + } + + @DataProvider(name = "topics") + public static Object[][] topics() { + return new Object[][] { + { "persistent", Boolean.TRUE }, + { "persistent", Boolean.FALSE }, + { "non-persistent", Boolean.TRUE }, + { "non-persistent", Boolean.FALSE }, + }; + } + + @DataProvider(name = "partitioned") + public static Object[][] partitioned() { + return new Object[][] { { Boolean.TRUE }, { Boolean.FALSE } }; + } + + @Test(dataProvider = "topics") + public void simpleTest(String type, boolean partitioned) throws Exception { + String topic = newTopic(type, partitioned); + + Producer p1 = pulsarClient.newProducer(Schema.STRING) + .topic(topic) + .accessMode(ProducerAccessMode.Exclusive) + .create(); + + try { + pulsarClient.newProducer(Schema.STRING) + .topic(topic) + .accessMode(ProducerAccessMode.Exclusive) + .create(); + fail("Should have failed"); + } catch (ProducerFencedException e) { + // Expected + } + + try { + pulsarClient.newProducer(Schema.STRING) + .topic(topic) + .accessMode(ProducerAccessMode.Shared) + .create(); + fail("Should have failed"); + } catch (ProducerBusyException e) { + // Expected + } + + p1.close(); + + // Now producer should be allowed to get in + Producer p2 = pulsarClient.newProducer(Schema.STRING) + .topic(topic) + .accessMode(ProducerAccessMode.Exclusive) + .create(); + p2.close(); + } + + @Test(dataProvider = "topics") + public void existingSharedProducer(String type, boolean partitioned) throws Exception { + String topic = newTopic(type, partitioned); + + @Cleanup + Producer p1 = pulsarClient.newProducer(Schema.STRING) + .topic(topic) + .accessMode(ProducerAccessMode.Shared) + .create(); + + try { + pulsarClient.newProducer(Schema.STRING) + .topic(topic) + .accessMode(ProducerAccessMode.Exclusive) + .create(); + fail("Should have failed"); + } catch (ProducerFencedException e) { + // Expected + } + } + + @Test(dataProvider = "topics") + public void producerReconnection(String type, boolean partitioned) throws Exception { + String topic = newTopic(type, partitioned); + + Producer p1 = pulsarClient.newProducer(Schema.STRING) + .topic(topic) + .accessMode(ProducerAccessMode.Exclusive) + .create(); + + p1.send("msg-1"); + + admin.topics().unload(topic); + + p1.send("msg-2"); + } + + @Test(dataProvider = "partitioned") + public void producerFenced(boolean partitioned) throws Exception { + String topic = newTopic("persistent", partitioned); + + Producer p1 = pulsarClient.newProducer(Schema.STRING) + .topic(topic) + .accessMode(ProducerAccessMode.Exclusive) + .create(); + + p1.send("msg-1"); + + // Simulate a producer that takes over and fences p1 through the topic epoch + if (!partitioned) { + Topic t = pulsar.getBrokerService().getTopic(topic, false).get().get(); + CompletableFuture f = (CompletableFuture) Whitebox + .getMethod(AbstractTopic.class, "incrementTopicEpoch", Optional.class) + .invoke(t, Optional.of(0L)); + f.get(); + } else { + for (int i = 0; i < 3; i++) { + String name = TopicName.get(topic).getPartition(i).toString(); + Topic t = pulsar.getBrokerService().getTopic(name, false).get().get(); + CompletableFuture f = (CompletableFuture) Whitebox + .getMethod(AbstractTopic.class, "incrementTopicEpoch", Optional.class) + .invoke(t, Optional.of(0L)); + f.get(); + } + } + + admin.topics().unload(topic); + + try { + p1.send("msg-2"); + fail("Should have failed"); + } catch (ProducerFencedException e) { + // Expected + } + } + + private String newTopic(String type, boolean isPartitioned) throws Exception { + String topic = type + "://" + newTopicName(); + if (isPartitioned) { + admin.topics().createPartitionedTopic(topic, 3); + } + + return topic; + } +} 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 d2f58fc0e27bb..6117c8f6e27d1 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 @@ -107,6 +107,7 @@ import org.apache.pulsar.common.api.proto.PulsarApi.CommandSubscribe.InitialPosition; import org.apache.pulsar.common.api.proto.PulsarApi.CommandSubscribe.SubType; import org.apache.pulsar.common.api.proto.PulsarApi.MessageMetadata; +import org.apache.pulsar.common.api.proto.PulsarApi.ProducerAccessMode; import org.apache.pulsar.common.naming.TopicName; import org.apache.pulsar.common.policies.data.Policies; import org.apache.pulsar.common.protocol.ByteBufPair; @@ -371,23 +372,25 @@ public void testAddRemoveProducer() throws Exception { String role = "appid1"; // 1. simple add producer Producer producer = new Producer(topic, serverCnx, 1 /* producer id */, "prod-name", - role, false, null, SchemaVersion.Latest, 0, false); + role, false, null, SchemaVersion.Latest, 0, false, + ProducerAccessMode.Shared, Optional.empty()); topic.addProducer(producer); assertEquals(topic.getProducers().size(), 1); // 2. duplicate add try { - topic.addProducer(producer); + topic.addProducer(producer).join(); fail("Should have failed with naming exception because producer 'null' is already connected to the topic"); - } catch (BrokerServiceException e) { - assertTrue(e instanceof BrokerServiceException.NamingException); + } catch (Exception e) { + assertEquals(e.getCause().getClass(), BrokerServiceException.NamingException.class); } assertEquals(topic.getProducers().size(), 1); // 3. add producer for a different topic PersistentTopic failTopic = new PersistentTopic(failTopicName, ledgerMock, brokerService); Producer failProducer = new Producer(failTopic, serverCnx, 2 /* producer id */, "prod-name", - role, false, null, SchemaVersion.Latest,0, false); + role, false, null, SchemaVersion.Latest,0, false, + ProducerAccessMode.Shared, Optional.empty()); try { topic.addProducer(failProducer); fail("should have failed"); @@ -408,27 +411,29 @@ public void testProducerOverwrite() throws Exception { PersistentTopic topic = new PersistentTopic(successTopicName, ledgerMock, brokerService); String role = "appid1"; Producer producer1 = new Producer(topic, serverCnx, 1 /* producer id */, "prod-name", - role, false, null, SchemaVersion.Latest, 0, true); + role, false, null, SchemaVersion.Latest, 0, true, ProducerAccessMode.Shared, Optional.empty()); Producer producer2 = new Producer(topic, serverCnx, 2 /* producer id */, "prod-name", - role, false, null, SchemaVersion.Latest, 0, true); + role, false, null, SchemaVersion.Latest, 0, true, ProducerAccessMode.Shared, Optional.empty()); try { - topic.addProducer(producer1); - topic.addProducer(producer2); + topic.addProducer(producer1).join(); + topic.addProducer(producer2).join(); fail("should have failed"); - } catch (BrokerServiceException.NamingException e) { + } catch (Exception e) { // OK + assertEquals(e.getCause().getClass(), BrokerServiceException.NamingException.class); } Assert.assertEquals(topic.getProducers().size(), 1); Producer producer3 = new Producer(topic, serverCnx, 2 /* producer id */, "prod-name", - role, false, null, SchemaVersion.Latest, 1, false); + role, false, null, SchemaVersion.Latest, 1, false, ProducerAccessMode.Shared, Optional.empty()); try { - topic.addProducer(producer3); + topic.addProducer(producer3).join(); fail("should have failed"); - } catch (BrokerServiceException.NamingException e) { + } catch (Exception e) { // OK + assertEquals(e.getCause().getClass(), BrokerServiceException.NamingException.class); } Assert.assertEquals(topic.getProducers().size(), 1); @@ -437,7 +442,7 @@ public void testProducerOverwrite() throws Exception { Assert.assertEquals(topic.getProducers().size(), 0); Producer producer4 = new Producer(topic, serverCnx, 2 /* producer id */, "prod-name", - role, false, null, SchemaVersion.Latest, 2, false); + role, false, null, SchemaVersion.Latest, 2, false, ProducerAccessMode.Shared, Optional.empty()); topic.addProducer(producer3); topic.addProducer(producer4); @@ -450,13 +455,13 @@ public void testProducerOverwrite() throws Exception { Assert.assertEquals(topic.getProducers().size(), 0); Producer producer5 = new Producer(topic, serverCnx, 2 /* producer id */, "pulsar.repl.cluster1", - role, false, null, SchemaVersion.Latest, 1, false); + role, false, null, SchemaVersion.Latest, 1, false, ProducerAccessMode.Shared, Optional.empty()); topic.addProducer(producer5); Assert.assertEquals(topic.getProducers().size(), 1); Producer producer6 = new Producer(topic, serverCnx, 2 /* producer id */, "pulsar.repl.cluster1", - role, false, null, SchemaVersion.Latest, 2, false); + role, false, null, SchemaVersion.Latest, 2, false, ProducerAccessMode.Shared, Optional.empty()); topic.addProducer(producer6); Assert.assertEquals(topic.getProducers().size(), 1); @@ -464,7 +469,7 @@ public void testProducerOverwrite() throws Exception { topic.getProducers().values().forEach(producer -> Assert.assertEquals(producer.getEpoch(), 2)); Producer producer7 = new Producer(topic, serverCnx, 2 /* producer id */, "pulsar.repl.cluster1", - role, false, null, SchemaVersion.Latest, 3, true); + role, false, null, SchemaVersion.Latest, 3, true, ProducerAccessMode.Shared, Optional.empty()); topic.addProducer(producer7); Assert.assertEquals(topic.getProducers().size(), 1); @@ -476,24 +481,24 @@ public void testMaxProducers() throws Exception { String role = "appid1"; // 1. add producer1 Producer producer = new Producer(topic, serverCnx, 1 /* producer id */, "prod-name1", role, - false, null, SchemaVersion.Latest,0, false); + false, null, SchemaVersion.Latest,0, false, ProducerAccessMode.Shared, Optional.empty()); topic.addProducer(producer); assertEquals(topic.getProducers().size(), 1); // 2. add producer2 Producer producer2 = new Producer(topic, serverCnx, 2 /* producer id */, "prod-name2", role, - false, null, SchemaVersion.Latest,0, false); + false, null, SchemaVersion.Latest,0, false, ProducerAccessMode.Shared, Optional.empty()); topic.addProducer(producer2); assertEquals(topic.getProducers().size(), 2); // 3. add producer3 but reached maxProducersPerTopic try { Producer producer3 = new Producer(topic, serverCnx, 3 /* producer id */, "prod-name3", role, - false, null, SchemaVersion.Latest,0, false); - topic.addProducer(producer3); + false, null, SchemaVersion.Latest,0, false, ProducerAccessMode.Shared, Optional.empty()); + topic.addProducer(producer3).join(); fail("should have failed"); - } catch (BrokerServiceException e) { - assertTrue(e instanceof BrokerServiceException.ProducerBusyException); + } catch (Exception e) { + assertEquals(e.getCause().getClass(), BrokerServiceException.ProducerBusyException.class); } } @@ -947,8 +952,8 @@ public void testDeleteTopic() throws Exception { // 3. delete topic with producer topic = (PersistentTopic) brokerService.getOrCreateTopic(successTopicName).get(); Producer producer = new Producer(topic, serverCnx, 1 /* producer id */, "prod-name", - role, false, null, SchemaVersion.Latest, 0, false); - topic.addProducer(producer); + role, false, null, SchemaVersion.Latest, 0, false, ProducerAccessMode.Shared, Optional.empty()); + topic.addProducer(producer).join(); assertTrue(topic.delete().isCompletedExceptionally()); assertFalse((boolean) isFencedField.get(topic)); @@ -1110,11 +1115,11 @@ public Object answer(InvocationOnMock invocationOnMock) throws Throwable { String role = "appid1"; Thread.sleep(10); /* delay to ensure that the delete gets executed first */ Producer producer = new Producer(topic, serverCnx, 1 /* producer id */, "prod-name", - role, false, null, SchemaVersion.Latest, 0, false); - topic.addProducer(producer); + role, false, null, SchemaVersion.Latest, 0, false, ProducerAccessMode.Shared, Optional.empty()); + topic.addProducer(producer).join(); fail("Should have failed"); - } catch (BrokerServiceException e) { - assertTrue(e instanceof BrokerServiceException.TopicFencedException); + } catch (Exception e) { + assertEquals(e.getCause().getClass(), BrokerServiceException.TopicFencedException.class); } CommandSubscribe cmd = CommandSubscribe.newBuilder().setConsumerId(1).setTopic(successTopicName) diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerAccessMode.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerAccessMode.java new file mode 100644 index 0000000000000..1a69da80f1387 --- /dev/null +++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerAccessMode.java @@ -0,0 +1,41 @@ +/** + * 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.api; + +/** + * The type of access to the topic that the producer requires. + */ +public enum ProducerAccessMode { + /** + * By default multiple producers can publish on a topic. + */ + Shared, + + /** + * Require exclusive access for producer. Fail immediately if there's already a producer connected. + */ + Exclusive, + +// TODO +// /** +// * Producer creation is pending until it can acquire exclusive access. +// */ +// WaitForExclusive, +} diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java index 8c40343770ccc..5abf28b07b771 100644 --- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java +++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java @@ -125,6 +125,25 @@ public interface ProducerBuilder extends Cloneable { */ ProducerBuilder producerName(String producerName); + /** + * Configure the type of access mode that the producer requires on the topic. + * + *

Possible values are: + *

    + *
  • {@link ProducerAccessMode#Shared}: By default multiple producers can publish on a topic + *
  • {@link ProducerAccessMode#Exclusive}: Require exclusive access for producer. Fail immediately if there's + * already a producer connected. + *
  • {@link ProducerAccessMode#WaitForExclusive}: Producer creation is pending until it can acquire exclusive + * access + *
+ * + * @see ProducerAccessMode + * @param accessMode + * The type of access to the topic that the producer requires + * @return the producer builder instance + */ + ProducerBuilder accessMode(ProducerAccessMode accessMode); + /** * Set the send timeout (default: 30 seconds). * diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientException.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientException.java index f49ea52171526..a11966874cc32 100644 --- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientException.java +++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientException.java @@ -416,6 +416,22 @@ public TopicTerminatedException(String msg, long sequenceId) { } } + /** + * Producer fenced exception thrown by Pulsar client. + */ + public static class ProducerFencedException extends PulsarClientException { + /** + * Constructs a {@code ProducerFencedException} with the specified detail message. + * + * @param msg + * The detail message (which is saved for later retrieval + * by the {@link #getMessage()} method) + */ + public ProducerFencedException(String msg) { + super(msg); + } + } + /** * Authentication exception thrown by Pulsar client. */ @@ -972,6 +988,8 @@ public static PulsarClientException unwrap(Throwable t) { return new TransactionConflictException(msg); } else if (cause instanceof TopicDoesNotExistException) { return new TopicDoesNotExistException(msg); + } else if (cause instanceof ProducerFencedException) { + return new ProducerFencedException(msg); } else { return new PulsarClientException(t); } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java index 42c3facf0f4f6..212292eee7f5f 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java @@ -462,7 +462,11 @@ protected void handleProducerSuccess(CommandProducerSuccess success) { long requestId = success.getRequestId(); CompletableFuture requestFuture = (CompletableFuture) pendingRequests.remove(requestId); if (requestFuture != null) { - requestFuture.complete(new ProducerResponse(success.getProducerName(), success.getLastSequenceId(), success.getSchemaVersion().toByteArray())); + ProducerResponse pr = new ProducerResponse(success.getProducerName(), + success.getLastSequenceId(), + success.getSchemaVersion().toByteArray(), + success.hasTopicEpoch() ? Optional.of(success.getTopicEpoch()) : Optional.empty()); + requestFuture.complete(pr); } else { log.warn("{} Received unknown request id from server: {}", ctx.channel(), success.getRequestId()); } @@ -1029,6 +1033,8 @@ private PulsarClientException getPulsarClientException(ServerError error, String return new PulsarClientException.NotAllowedException(errorMsg); case TransactionConflict: return new PulsarClientException.TransactionConflictException(errorMsg); + case ProducerFenced: + return new PulsarClientException.ProducerFencedException(errorMsg); case UnknownError: default: return new PulsarClientException(errorMsg); diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java index a9a8f7cb195b7..6591199129ca3 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java @@ -155,6 +155,7 @@ private boolean isValidStateForReconnection() { case Closing: case Closed: case Failed: + case ProducerFenced: case Terminated: return false; } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HandlerState.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HandlerState.java index 187d28d92c0b0..e72c97fadadcc 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HandlerState.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HandlerState.java @@ -39,7 +39,8 @@ enum State { Terminated, // Topic associated with this handler // has been terminated Failed, // Handler is failed - RegisteringSchema // Handler is registering schema + RegisteringSchema, // Handler is registering schema + ProducerFenced, // The producer has been fenced by the broker }; public HandlerState(PulsarClientImpl client, String topic) { diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PartitionedProducerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PartitionedProducerImpl.java index 3795a7568a44e..35c55aef759aa 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PartitionedProducerImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PartitionedProducerImpl.java @@ -178,6 +178,8 @@ CompletableFuture internalSendWithTxnAsync(Message message, Transa case Closing: case Closed: return FutureUtil.failedFuture(new PulsarClientException.AlreadyClosedException("Producer already closed")); + case ProducerFenced: + return FutureUtil.failedFuture(new PulsarClientException.ProducerFencedException("Producer was fenced")); case Terminated: return FutureUtil.failedFuture(new PulsarClientException.TopicTerminatedException("Topic was terminated")); case Failed: diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java index 64a22d7b6d2d1..a1a17bcfd1b71 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java @@ -38,6 +38,7 @@ import org.apache.pulsar.client.api.MessageRouter; import org.apache.pulsar.client.api.MessageRoutingMode; import org.apache.pulsar.client.api.Producer; +import org.apache.pulsar.client.api.ProducerAccessMode; import org.apache.pulsar.client.api.ProducerBuilder; import org.apache.pulsar.client.api.ProducerCryptoFailureAction; import org.apache.pulsar.client.api.PulsarClientException; @@ -153,6 +154,12 @@ public ProducerBuilder maxPendingMessagesAcrossPartitions(int maxPendingMessa return this; } + @Override + public ProducerBuilder accessMode(ProducerAccessMode accessMode) { + conf.setAccessMode(accessMode); + return this; + } + @Override public ProducerBuilder blockIfQueueFull(boolean blockIfQueueFull) { conf.setBlockIfQueueFull(blockIfQueueFull); diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java index 3ab4207544c55..9ae41681c2d4e 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java @@ -60,6 +60,7 @@ import org.apache.pulsar.client.api.MessageCrypto; import org.apache.pulsar.client.api.MessageId; import org.apache.pulsar.client.api.Producer; +import org.apache.pulsar.client.api.ProducerAccessMode; import org.apache.pulsar.client.api.ProducerCryptoFailureAction; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.PulsarClientException.CryptoException; @@ -136,6 +137,8 @@ public class ProducerImpl extends ProducerBase implements TimerTask, Conne private ScheduledFuture batchTimerTask; + private Optional topicEpoch = Optional.empty(); + @SuppressWarnings("rawtypes") private static final AtomicLongFieldUpdater msgIdGeneratorUpdater = AtomicLongFieldUpdater .newUpdater(ProducerImpl.class, "msgIdGenerator"); @@ -708,6 +711,9 @@ private boolean isValidProducerState(SendCallback callback, long sequenceId) { case Closed: callback.sendComplete(new PulsarClientException.AlreadyClosedException("Producer already closed", sequenceId)); return false; + case ProducerFenced: + callback.sendComplete(new PulsarClientException.ProducerFencedException("Producer was fenced")); + return false; case Terminated: callback.sendComplete(new PulsarClientException.TopicTerminatedException("Topic was terminated", sequenceId)); return false; @@ -1229,7 +1235,8 @@ public void connectionOpened(final ClientCnx cnx) { cnx.sendRequestWithId( Commands.newProducer(topic, producerId, requestId, producerName, conf.isEncryptionEnabled(), metadata, - schemaInfo, connectionHandler.epoch, userProvidedProducerName), + schemaInfo, connectionHandler.epoch, userProvidedProducerName, + conf.getAccessMode(), topicEpoch), requestId).thenAccept(response -> { String producerName = response.getProducerName(); long lastSequenceId = response.getLastSequenceId(); @@ -1251,6 +1258,11 @@ public void connectionOpened(final ClientCnx cnx) { log.info("[{}] [{}] Created producer on cnx {}", topic, producerName, cnx.ctx().channel()); connectionId = cnx.ctx().channel().toString(); connectedSince = DateFormatter.now(); + if (conf.getAccessMode() != ProducerAccessMode.Shared && !topicEpoch.isPresent()) { + log.info("[{}] [{}] Producer epoch is {}", topic, producerName, response.getTopicEpoch()); + } + topicEpoch = response.getTopicEpoch(); + if (this.producerName == null) { this.producerName = producerName; @@ -1332,6 +1344,11 @@ public void connectionOpened(final ClientCnx cnx) { failPendingMessages(cnx(), (PulsarClientException) cause); producerCreatedFuture.completeExceptionally(cause); client.cleanupProducer(this); + } else if (cause instanceof PulsarClientException.ProducerFencedException) { + setState(State.ProducerFenced); + failPendingMessages(cnx(), (PulsarClientException) cause); + producerCreatedFuture.completeExceptionally(cause); + client.cleanupProducer(this); } else if (producerCreatedFuture.isDone() || // (cause instanceof PulsarClientException && PulsarClientException.isRetriableError(cause) && System.currentTimeMillis() < createProducerTimeout)) { diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerResponse.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerResponse.java index edb98f2b04755..36b47f2b6d62f 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerResponse.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerResponse.java @@ -18,6 +18,8 @@ */ package org.apache.pulsar.client.impl; +import java.util.Optional; + import lombok.AllArgsConstructor; import lombok.Data; @@ -27,4 +29,6 @@ public class ProducerResponse { private String producerName; private long lastSequenceId; private byte[] schemaVersion; + + private Optional topicEpoch; } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ProducerConfigurationData.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ProducerConfigurationData.java index c48598a8372bd..eb011ad3a5f8d 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ProducerConfigurationData.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ProducerConfigurationData.java @@ -35,6 +35,7 @@ import org.apache.pulsar.client.api.MessageCrypto; import org.apache.pulsar.client.api.MessageRouter; import org.apache.pulsar.client.api.MessageRoutingMode; +import org.apache.pulsar.client.api.ProducerAccessMode; import org.apache.pulsar.client.api.ProducerCryptoFailureAction; import com.fasterxml.jackson.annotation.JsonIgnore; @@ -99,6 +100,8 @@ public class ProducerConfigurationData implements Serializable, Cloneable { private boolean multiSchema = true; + private ProducerAccessMode accessMode = ProducerAccessMode.Shared; + private SortedMap properties = new TreeMap<>(); /** 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 9e1e9d3d60b34..f30cb92a0b12e 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 @@ -58,6 +58,50 @@ private CompressionType(int index, int value) { // @@protoc_insertion_point(enum_scope:pulsar.proto.CompressionType) } + public enum ProducerAccessMode + implements org.apache.pulsar.shaded.com.google.protobuf.v241.Internal.EnumLite { + Shared(0, 0), + Exclusive(1, 1), + WaitForExclusive(2, 2), + ; + + public static final int Shared_VALUE = 0; + public static final int Exclusive_VALUE = 1; + public static final int WaitForExclusive_VALUE = 2; + + + public final int getNumber() { return value; } + + public static ProducerAccessMode valueOf(int value) { + switch (value) { + case 0: return Shared; + case 1: return Exclusive; + case 2: return WaitForExclusive; + default: return null; + } + } + + public static org.apache.pulsar.shaded.com.google.protobuf.v241.Internal.EnumLiteMap + internalGetValueMap() { + return internalValueMap; + } + private static org.apache.pulsar.shaded.com.google.protobuf.v241.Internal.EnumLiteMap + internalValueMap = + new org.apache.pulsar.shaded.com.google.protobuf.v241.Internal.EnumLiteMap() { + public ProducerAccessMode findValueByNumber(int number) { + return ProducerAccessMode.valueOf(number); + } + }; + + private final int value; + + private ProducerAccessMode(int index, int value) { + this.value = value; + } + + // @@protoc_insertion_point(enum_scope:pulsar.proto.ProducerAccessMode) + } + public enum ServerError implements org.apache.pulsar.shaded.com.google.protobuf.v241.Internal.EnumLite { UnknownError(0, 0), @@ -85,6 +129,7 @@ public enum ServerError NotAllowedError(22, 22), TransactionConflict(23, 23), TransactionNotFound(24, 24), + ProducerFenced(25, 25), ; public static final int UnknownError_VALUE = 0; @@ -112,6 +157,7 @@ public enum ServerError public static final int NotAllowedError_VALUE = 22; public static final int TransactionConflict_VALUE = 23; public static final int TransactionNotFound_VALUE = 24; + public static final int ProducerFenced_VALUE = 25; public final int getNumber() { return value; } @@ -143,6 +189,7 @@ public static ServerError valueOf(int value) { case 22: return NotAllowedError; case 23: return TransactionConflict; case 24: return TransactionNotFound; + case 25: return ProducerFenced; default: return null; } } @@ -15744,6 +15791,14 @@ public interface CommandProducerOrBuilder // optional bool user_provided_producer_name = 9 [default = true]; boolean hasUserProvidedProducerName(); boolean getUserProvidedProducerName(); + + // optional .pulsar.proto.ProducerAccessMode producer_access_mode = 10 [default = Shared]; + boolean hasProducerAccessMode(); + org.apache.pulsar.common.api.proto.PulsarApi.ProducerAccessMode getProducerAccessMode(); + + // optional uint64 topic_epoch = 11; + boolean hasTopicEpoch(); + long getTopicEpoch(); } public static final class CommandProducer extends org.apache.pulsar.shaded.com.google.protobuf.v241.GeneratedMessageLite @@ -15925,6 +15980,26 @@ public boolean getUserProvidedProducerName() { return userProvidedProducerName_; } + // optional .pulsar.proto.ProducerAccessMode producer_access_mode = 10 [default = Shared]; + public static final int PRODUCER_ACCESS_MODE_FIELD_NUMBER = 10; + private org.apache.pulsar.common.api.proto.PulsarApi.ProducerAccessMode producerAccessMode_; + public boolean hasProducerAccessMode() { + return ((bitField0_ & 0x00000100) == 0x00000100); + } + public org.apache.pulsar.common.api.proto.PulsarApi.ProducerAccessMode getProducerAccessMode() { + return producerAccessMode_; + } + + // optional uint64 topic_epoch = 11; + public static final int TOPIC_EPOCH_FIELD_NUMBER = 11; + private long topicEpoch_; + public boolean hasTopicEpoch() { + return ((bitField0_ & 0x00000200) == 0x00000200); + } + public long getTopicEpoch() { + return topicEpoch_; + } + private void initFields() { topic_ = ""; producerId_ = 0L; @@ -15935,6 +16010,8 @@ private void initFields() { schema_ = org.apache.pulsar.common.api.proto.PulsarApi.Schema.getDefaultInstance(); epoch_ = 0L; userProvidedProducerName_ = true; + producerAccessMode_ = org.apache.pulsar.common.api.proto.PulsarApi.ProducerAccessMode.Shared; + topicEpoch_ = 0L; } private byte memoizedIsInitialized = -1; public final boolean isInitialized() { @@ -16004,6 +16081,12 @@ public void writeTo(org.apache.pulsar.common.util.protobuf.ByteBufCodedOutputStr if (((bitField0_ & 0x00000080) == 0x00000080)) { output.writeBool(9, userProvidedProducerName_); } + if (((bitField0_ & 0x00000100) == 0x00000100)) { + output.writeEnum(10, producerAccessMode_.getNumber()); + } + if (((bitField0_ & 0x00000200) == 0x00000200)) { + output.writeUInt64(11, topicEpoch_); + } } private int memoizedSerializedSize = -1; @@ -16048,6 +16131,14 @@ public int getSerializedSize() { size += org.apache.pulsar.shaded.com.google.protobuf.v241.CodedOutputStream .computeBoolSize(9, userProvidedProducerName_); } + if (((bitField0_ & 0x00000100) == 0x00000100)) { + size += org.apache.pulsar.shaded.com.google.protobuf.v241.CodedOutputStream + .computeEnumSize(10, producerAccessMode_.getNumber()); + } + if (((bitField0_ & 0x00000200) == 0x00000200)) { + size += org.apache.pulsar.shaded.com.google.protobuf.v241.CodedOutputStream + .computeUInt64Size(11, topicEpoch_); + } memoizedSerializedSize = size; return size; } @@ -16179,6 +16270,10 @@ public Builder clear() { bitField0_ = (bitField0_ & ~0x00000080); userProvidedProducerName_ = true; bitField0_ = (bitField0_ & ~0x00000100); + producerAccessMode_ = org.apache.pulsar.common.api.proto.PulsarApi.ProducerAccessMode.Shared; + bitField0_ = (bitField0_ & ~0x00000200); + topicEpoch_ = 0L; + bitField0_ = (bitField0_ & ~0x00000400); return this; } @@ -16249,6 +16344,14 @@ public org.apache.pulsar.common.api.proto.PulsarApi.CommandProducer buildPartial to_bitField0_ |= 0x00000080; } result.userProvidedProducerName_ = userProvidedProducerName_; + if (((from_bitField0_ & 0x00000200) == 0x00000200)) { + to_bitField0_ |= 0x00000100; + } + result.producerAccessMode_ = producerAccessMode_; + if (((from_bitField0_ & 0x00000400) == 0x00000400)) { + to_bitField0_ |= 0x00000200; + } + result.topicEpoch_ = topicEpoch_; result.bitField0_ = to_bitField0_; return result; } @@ -16289,6 +16392,12 @@ public Builder mergeFrom(org.apache.pulsar.common.api.proto.PulsarApi.CommandPro if (other.hasUserProvidedProducerName()) { setUserProvidedProducerName(other.getUserProvidedProducerName()); } + if (other.hasProducerAccessMode()) { + setProducerAccessMode(other.getProducerAccessMode()); + } + if (other.hasTopicEpoch()) { + setTopicEpoch(other.getTopicEpoch()); + } return this; } @@ -16393,6 +16502,20 @@ public Builder mergeFrom( userProvidedProducerName_ = input.readBool(); break; } + case 80: { + int rawValue = input.readEnum(); + org.apache.pulsar.common.api.proto.PulsarApi.ProducerAccessMode value = org.apache.pulsar.common.api.proto.PulsarApi.ProducerAccessMode.valueOf(rawValue); + if (value != null) { + bitField0_ |= 0x00000200; + producerAccessMode_ = value; + } + break; + } + case 88: { + bitField0_ |= 0x00000400; + topicEpoch_ = input.readUInt64(); + break; + } } } } @@ -16708,6 +16831,51 @@ public Builder clearUserProvidedProducerName() { return this; } + // optional .pulsar.proto.ProducerAccessMode producer_access_mode = 10 [default = Shared]; + private org.apache.pulsar.common.api.proto.PulsarApi.ProducerAccessMode producerAccessMode_ = org.apache.pulsar.common.api.proto.PulsarApi.ProducerAccessMode.Shared; + public boolean hasProducerAccessMode() { + return ((bitField0_ & 0x00000200) == 0x00000200); + } + public org.apache.pulsar.common.api.proto.PulsarApi.ProducerAccessMode getProducerAccessMode() { + return producerAccessMode_; + } + public Builder setProducerAccessMode(org.apache.pulsar.common.api.proto.PulsarApi.ProducerAccessMode value) { + if (value == null) { + throw new NullPointerException(); + } + bitField0_ |= 0x00000200; + producerAccessMode_ = value; + + return this; + } + public Builder clearProducerAccessMode() { + bitField0_ = (bitField0_ & ~0x00000200); + producerAccessMode_ = org.apache.pulsar.common.api.proto.PulsarApi.ProducerAccessMode.Shared; + + return this; + } + + // optional uint64 topic_epoch = 11; + private long topicEpoch_ ; + public boolean hasTopicEpoch() { + return ((bitField0_ & 0x00000400) == 0x00000400); + } + public long getTopicEpoch() { + return topicEpoch_; + } + public Builder setTopicEpoch(long value) { + bitField0_ |= 0x00000400; + topicEpoch_ = value; + + return this; + } + public Builder clearTopicEpoch() { + bitField0_ = (bitField0_ & ~0x00000400); + topicEpoch_ = 0L; + + return this; + } + // @@protoc_insertion_point(builder_scope:pulsar.proto.CommandProducer) } @@ -24564,6 +24732,10 @@ public interface CommandProducerSuccessOrBuilder // optional bytes schema_version = 4; boolean hasSchemaVersion(); org.apache.pulsar.shaded.com.google.protobuf.v241.ByteString getSchemaVersion(); + + // optional uint64 topic_epoch = 5; + boolean hasTopicEpoch(); + long getTopicEpoch(); } public static final class CommandProducerSuccess extends org.apache.pulsar.shaded.com.google.protobuf.v241.GeneratedMessageLite @@ -24662,11 +24834,22 @@ public org.apache.pulsar.shaded.com.google.protobuf.v241.ByteString getSchemaVer return schemaVersion_; } + // optional uint64 topic_epoch = 5; + public static final int TOPIC_EPOCH_FIELD_NUMBER = 5; + private long topicEpoch_; + public boolean hasTopicEpoch() { + return ((bitField0_ & 0x00000010) == 0x00000010); + } + public long getTopicEpoch() { + return topicEpoch_; + } + private void initFields() { requestId_ = 0L; producerName_ = ""; lastSequenceId_ = -1L; schemaVersion_ = org.apache.pulsar.shaded.com.google.protobuf.v241.ByteString.EMPTY; + topicEpoch_ = 0L; } private byte memoizedIsInitialized = -1; public final boolean isInitialized() { @@ -24705,6 +24888,9 @@ public void writeTo(org.apache.pulsar.common.util.protobuf.ByteBufCodedOutputStr if (((bitField0_ & 0x00000008) == 0x00000008)) { output.writeBytes(4, schemaVersion_); } + if (((bitField0_ & 0x00000010) == 0x00000010)) { + output.writeUInt64(5, topicEpoch_); + } } private int memoizedSerializedSize = -1; @@ -24729,6 +24915,10 @@ public int getSerializedSize() { size += org.apache.pulsar.shaded.com.google.protobuf.v241.CodedOutputStream .computeBytesSize(4, schemaVersion_); } + if (((bitField0_ & 0x00000010) == 0x00000010)) { + size += org.apache.pulsar.shaded.com.google.protobuf.v241.CodedOutputStream + .computeUInt64Size(5, topicEpoch_); + } memoizedSerializedSize = size; return size; } @@ -24850,6 +25040,8 @@ public Builder clear() { bitField0_ = (bitField0_ & ~0x00000004); schemaVersion_ = org.apache.pulsar.shaded.com.google.protobuf.v241.ByteString.EMPTY; bitField0_ = (bitField0_ & ~0x00000008); + topicEpoch_ = 0L; + bitField0_ = (bitField0_ & ~0x00000010); return this; } @@ -24899,6 +25091,10 @@ public org.apache.pulsar.common.api.proto.PulsarApi.CommandProducerSuccess build to_bitField0_ |= 0x00000008; } result.schemaVersion_ = schemaVersion_; + if (((from_bitField0_ & 0x00000010) == 0x00000010)) { + to_bitField0_ |= 0x00000010; + } + result.topicEpoch_ = topicEpoch_; result.bitField0_ = to_bitField0_; return result; } @@ -24917,6 +25113,9 @@ public Builder mergeFrom(org.apache.pulsar.common.api.proto.PulsarApi.CommandPro if (other.hasSchemaVersion()) { setSchemaVersion(other.getSchemaVersion()); } + if (other.hasTopicEpoch()) { + setTopicEpoch(other.getTopicEpoch()); + } return this; } @@ -24974,6 +25173,11 @@ public Builder mergeFrom( schemaVersion_ = input.readBytes(); break; } + case 40: { + bitField0_ |= 0x00000010; + topicEpoch_ = input.readUInt64(); + break; + } } } } @@ -25082,6 +25286,27 @@ public Builder clearSchemaVersion() { return this; } + // optional uint64 topic_epoch = 5; + private long topicEpoch_ ; + public boolean hasTopicEpoch() { + return ((bitField0_ & 0x00000010) == 0x00000010); + } + public long getTopicEpoch() { + return topicEpoch_; + } + public Builder setTopicEpoch(long value) { + bitField0_ |= 0x00000010; + topicEpoch_ = value; + + return this; + } + public Builder clearTopicEpoch() { + bitField0_ = (bitField0_ & ~0x00000010); + topicEpoch_ = 0L; + + return this; + } + // @@protoc_insertion_point(builder_scope:pulsar.proto.CommandProducerSuccess) } diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java index 90f354131c6c9..edf308ed9ce25 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java @@ -37,6 +37,7 @@ import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.tuple.Triple; import org.apache.pulsar.client.api.KeySharedPolicy; +import org.apache.pulsar.client.api.ProducerAccessMode; import org.apache.pulsar.client.api.Range; import org.apache.pulsar.client.api.transaction.TxnID; import org.apache.pulsar.common.allocator.PulsarByteBufAllocator; @@ -357,20 +358,21 @@ public static ByteBuf newSuccess(long requestId) { public static BaseCommand newProducerSuccessCommand(long requestId, String producerName, SchemaVersion schemaVersion) { - return newProducerSuccessCommand(requestId, producerName, -1, schemaVersion); + return newProducerSuccessCommand(requestId, producerName, -1, schemaVersion, Optional.empty()); } public static ByteBuf newProducerSuccess(long requestId, String producerName, SchemaVersion schemaVersion) { - return newProducerSuccess(requestId, producerName, -1, schemaVersion); + return newProducerSuccess(requestId, producerName, -1, schemaVersion, Optional.empty()); } public static BaseCommand newProducerSuccessCommand(long requestId, String producerName, long lastSequenceId, - SchemaVersion schemaVersion) { + SchemaVersion schemaVersion, Optional topicEpoch) { CommandProducerSuccess.Builder producerSuccessBuilder = CommandProducerSuccess.newBuilder(); producerSuccessBuilder.setRequestId(requestId); producerSuccessBuilder.setProducerName(producerName); producerSuccessBuilder.setLastSequenceId(lastSequenceId); producerSuccessBuilder.setSchemaVersion(ByteString.copyFrom(schemaVersion.bytes())); + topicEpoch.ifPresent(producerSuccessBuilder::setTopicEpoch); CommandProducerSuccess producerSuccess = producerSuccessBuilder.build(); BaseCommand.Builder builder = BaseCommand.newBuilder(); BaseCommand res = builder.setType(Type.PRODUCER_SUCCESS).setProducerSuccess(producerSuccess).build(); @@ -380,12 +382,13 @@ public static BaseCommand newProducerSuccessCommand(long requestId, String produ } public static ByteBuf newProducerSuccess(long requestId, String producerName, long lastSequenceId, - SchemaVersion schemaVersion) { + SchemaVersion schemaVersion, Optional topicEpoch) { CommandProducerSuccess.Builder producerSuccessBuilder = CommandProducerSuccess.newBuilder(); producerSuccessBuilder.setRequestId(requestId); producerSuccessBuilder.setProducerName(producerName); producerSuccessBuilder.setLastSequenceId(lastSequenceId); producerSuccessBuilder.setSchemaVersion(ByteString.copyFrom(schemaVersion.bytes())); + topicEpoch.ifPresent(producerSuccessBuilder::setTopicEpoch); CommandProducerSuccess producerSuccess = producerSuccessBuilder.build(); ByteBuf res = serializeWithSize( BaseCommand.newBuilder().setType(Type.PRODUCER_SUCCESS).setProducerSuccess(producerSuccess)); @@ -846,7 +849,8 @@ public static ByteBuf newProducer(String topic, long producerId, long requestId, public static ByteBuf newProducer(String topic, long producerId, long requestId, String producerName, boolean encrypted, Map metadata) { - return newProducer(topic, producerId, requestId, producerName, encrypted, metadata, null, 0, false); + return newProducer(topic, producerId, requestId, producerName, encrypted, metadata, null, 0, false, + ProducerAccessMode.Shared, Optional.empty()); } private static Schema.Type getSchemaType(SchemaType type) { @@ -886,7 +890,9 @@ private static Schema getSchema(SchemaInfo schemaInfo) { public static ByteBuf newProducer(String topic, long producerId, long requestId, String producerName, boolean encrypted, Map metadata, SchemaInfo schemaInfo, - long epoch, boolean userProvidedProducerName) { + long epoch, boolean userProvidedProducerName, + ProducerAccessMode accessMode, + Optional topicEpoch) { CommandProducer.Builder producerBuilder = CommandProducer.newBuilder(); producerBuilder.setTopic(topic); producerBuilder.setProducerId(producerId); @@ -904,6 +910,9 @@ public static ByteBuf newProducer(String topic, long producerId, long requestId, producerBuilder.setSchema(getSchema(schemaInfo)); } + producerBuilder.setProducerAccessMode(convertProducerAccessMode(accessMode)); + topicEpoch.ifPresent(producerBuilder::setTopicEpoch); + CommandProducer producer = producerBuilder.build(); ByteBuf res = serializeWithSize(BaseCommand.newBuilder().setType(Type.PRODUCER).setProducer(producer)); producerBuilder.recycle(); @@ -2161,4 +2170,17 @@ public static boolean peerSupportJsonSchemaAvroFormat(int peerVersion) { public static boolean peerSupportsGetOrCreateSchema(int peerVersion) { return peerVersion >= ProtocolVersion.v15.getNumber(); } + + private static PulsarApi.ProducerAccessMode convertProducerAccessMode(ProducerAccessMode accessMode) { + switch (accessMode) { + case Exclusive: + return PulsarApi.ProducerAccessMode.Exclusive; + case Shared: + return PulsarApi.ProducerAccessMode.Shared; +// case WaitForExclusive: +// return PulsarApi.ProducerAccessMode.WaitForExclusive; + default: + throw new IllegalArgumentException("Unknonw access mode: " + accessMode); + } + } } diff --git a/pulsar-common/src/main/proto/PulsarApi.proto b/pulsar-common/src/main/proto/PulsarApi.proto index 5c12ca381ee81..b6e7e637fa4fb 100644 --- a/pulsar-common/src/main/proto/PulsarApi.proto +++ b/pulsar-common/src/main/proto/PulsarApi.proto @@ -92,6 +92,12 @@ enum CompressionType { SNAPPY = 4; } +enum ProducerAccessMode { + Shared = 0; // By default multiple producers can publish on a topic + Exclusive = 1; // Require exclusive access for producer. Fail immediately if there's already a producer connected. + WaitForExclusive = 2; // Producer creation is pending until it can acquire exclusive access +} + message MessageMetadata { required string producer_name = 1; required uint64 sequence_id = 2; @@ -206,6 +212,12 @@ enum ServerError { TransactionConflict = 23; // Ack with transaction conflict TransactionNotFound = 24; // Transaction not found + + ProducerFenced = 25; // When a producer asks and fail to get exclusive producer access, + // or loses the eclusive status after a reconnection, the broker will + // use this error to indicate that this producer is now permanently + // fenced. Applications are now supposed to close it and create a + // new producer } enum AuthMethod { @@ -450,6 +462,16 @@ message CommandProducer { // Indicate the name of the producer is generated or user provided // Use default true here is in order to be forward compatible with the client optional bool user_provided_producer_name = 9 [default = true]; + + // Require that this producers will be the only producer allowed on the topic + optional ProducerAccessMode producer_access_mode = 10 [default = Shared]; + + // Topic epoch is used to fence off producers that reconnects after a new + // exclusive producer has already taken over. This id is assigned by the + // broker on the CommandProducerSuccess. The first time, the client will + // leave it empty and then it will always carry the same epoch number on + // the subsequent reconnections. + optional uint64 topic_epoch = 11; } message CommandSend { @@ -589,6 +611,10 @@ message CommandProducerSuccess { // This will only be meaningful if deduplication has been enabled. optional int64 last_sequence_id = 3 [default = -1]; optional bytes schema_version = 4; + + // The topic epoch assigned by the broker. This field will only be set if we + // were requiring exclusive access when creating the producer. + optional uint64 topic_epoch = 5; } message CommandError { diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java index 9153812fccaf8..49e80b9dc49a9 100644 --- a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java +++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java @@ -61,6 +61,7 @@ import org.apache.pulsar.client.api.EncryptionKeyInfo; import org.apache.pulsar.client.api.MessageRoutingMode; import org.apache.pulsar.client.api.Producer; +import org.apache.pulsar.client.api.ProducerAccessMode; import org.apache.pulsar.client.api.ProducerBuilder; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.PulsarClientException; @@ -215,6 +216,9 @@ static class Arguments { @Parameter(names = {"-ioThreads", "--num-io-threads"}, description = "Set the number of threads to be " + "used for handling connections to brokers, default is 1 thread") public int ioThreads = 1; + + @Parameter(names = { "-am", "--access-mode" }, description = "Producer access mode") + public ProducerAccessMode producerAccessMode = ProducerAccessMode.Shared; } static class EncKeyReader implements CryptoKeyReader { @@ -452,6 +456,7 @@ private static void runProducer(Arguments arguments, .compressionType(arguments.compression) // .maxPendingMessages(arguments.maxOutstanding) // .maxPendingMessagesAcrossPartitions(arguments.maxPendingMessagesAcrossPartitions) + .accessMode(arguments.producerAccessMode) // enable round robin message routing if it is a partitioned topic .messageRoutingMode(MessageRoutingMode.RoundRobinPartition); From 3524cc00f57002a6f272d5e4f1560689f59405f9 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Tue, 24 Nov 2020 17:57:47 -0800 Subject: [PATCH 2/6] Added missing enums cases in C++ --- pulsar-client-cpp/include/pulsar/Result.h | 1 + pulsar-client-cpp/lib/ClientConnection.cc | 3 +++ pulsar-client-cpp/lib/Result.cc | 3 +++ 3 files changed, 7 insertions(+) diff --git a/pulsar-client-cpp/include/pulsar/Result.h b/pulsar-client-cpp/include/pulsar/Result.h index 1106aaeaf9a93..3b4cbf0417c01 100644 --- a/pulsar-client-cpp/include/pulsar/Result.h +++ b/pulsar-client-cpp/include/pulsar/Result.h @@ -85,6 +85,7 @@ enum Result ResultNotAllowedError, /// Not allowed ResultTransactionConflict, /// Transaction ack conflict ResultTransactionNotFound, /// Transaction not found + ResultProducerFenced, /// Producer was fenced by broker }; // Return string representation of result code diff --git a/pulsar-client-cpp/lib/ClientConnection.cc b/pulsar-client-cpp/lib/ClientConnection.cc index d17a9b609bc3c..027073084c313 100644 --- a/pulsar-client-cpp/lib/ClientConnection.cc +++ b/pulsar-client-cpp/lib/ClientConnection.cc @@ -126,6 +126,9 @@ static Result getResult(ServerError serverError) { case TransactionNotFound: return ResultTransactionNotFound; + + case ProducerFenced: + return ResultProducerFenced; } // NOTE : Do not add default case in the switch above. In future if we get new cases for // ServerError and miss them in the switch above we would like to get notified. Adding diff --git a/pulsar-client-cpp/lib/Result.cc b/pulsar-client-cpp/lib/Result.cc index 89dfe8e5303e7..86d7417b73674 100644 --- a/pulsar-client-cpp/lib/Result.cc +++ b/pulsar-client-cpp/lib/Result.cc @@ -153,6 +153,9 @@ const char* strResult(Result result) { case ResultTransactionNotFound: return "ResultTransactionNotFound"; + + case ResultProducerFenced: + return "ResultProducerFenced"; }; // NOTE : Do not add default case in the switch above. In future if we get new cases for // ServerError and miss them in the switch above we would like to get notified. Adding From 1f86250bdc81a8174f34718f1b3b6e02ccaf836e Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Tue, 1 Dec 2020 10:46:28 -0800 Subject: [PATCH 3/6] Addressed comments --- .../pulsar/broker/service/AbstractTopic.java | 25 ++++++++++++------- .../pulsar/broker/service/Producer.java | 1 + .../nonpersistent/NonPersistentTopic.java | 1 + .../service/persistent/PersistentTopic.java | 2 +- .../common/policies/data/PublisherStats.java | 4 +++ .../common/policies/data/TopicStats.java | 4 +++ .../pulsar/common/protocol/Commands.java | 13 ++++++++++ 7 files changed, 40 insertions(+), 10 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java index 7018bb80dcca0..be51770956a82 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java @@ -375,19 +375,26 @@ protected CompletableFuture> incrementTopicEpochIfNeeded(Producer switch (producer.getAccessMode()) { case Shared: if (hasExclusiveProducer) { - return FutureUtil.failedFuture(new ProducerBusyException("Topic has an existing exclusive producer")); + return FutureUtil.failedFuture(new ProducerBusyException( + "Topic has an existing exclusive producer: " + producers.keys().nextElement())); } else { // Normal producer getting added, we don't need a new epoch return CompletableFuture.completedFuture(topicEpoch); } case Exclusive: - if (hasExclusiveProducer || !producers.isEmpty()) { - return FutureUtil.failedFuture(new ProducerFencedException("Topic has existing producers")); - } else if (producer.getTopicEpoch().isPresent() && producer.getTopicEpoch().get() < topicEpoch.orElse(-1L)){ - // If a producer reconnects, but all the topic epoch has already moved forward, this producer needs to - // be fenced, because a new producer had been present in between. - return FutureUtil.failedFuture(new ProducerFencedException("Topic epoch has already moved")); + if (hasExclusiveProducer) { + return FutureUtil.failedFuture(new ProducerBusyException( + "Topic has an existing exclusive producer: " + producers.keys().nextElement())); + } else if (!producers.isEmpty()) { + return FutureUtil.failedFuture(new ProducerFencedException("Topic has existing shared producers")); + } else if (producer.getTopicEpoch().isPresent() + && producer.getTopicEpoch().get() < topicEpoch.orElse(-1L)) { + // If a producer reconnects, but all the topic epoch has already moved forward, this producer needs + // to be fenced, because a new producer had been present in between. + return FutureUtil.failedFuture(new ProducerFencedException( + String.format("Topic epoch has already moved. Current epoch: %d, Producer epoch: %d", + topicEpoch.get(), producer.getTopicEpoch().get()))); } else { // There are currently no existing producers hasExclusiveProducer = true; @@ -404,8 +411,8 @@ protected CompletableFuture> incrementTopicEpochIfNeeded(Producer return future; } - // case WaitForExclusive: - // TODO: Implementation + // case WaitForExclusive: + // TODO: Implementation default: return FutureUtil.failedFuture( diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java index 6602814bcf339..21a9cd1e275f5 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java @@ -124,6 +124,7 @@ public Producer(Topic topic, TransportCnx cnx, long producerId, String producerN stats.setProducerName(producerName); stats.producerId = producerId; stats.metadata = this.metadata; + stats.accessMode = Commands.convertProducerAccessMode(accessMode); this.isRemote = producerName .startsWith(cnx.getBrokerService().pulsar().getConfiguration().getReplicatorPrefix()); 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 56875a7892584..9012a20712dc8 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 @@ -780,6 +780,7 @@ public NonPersistentTopicStats getStats(boolean getPreciseBacklog) { stats.getReplication().put(replicator.getRemoteCluster(), replicatorStats); }); + stats.topicEpoch = topicEpoch.orElse(null); return stats; } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java index 0878e8e9c6a74..bc131ddba54f3 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java @@ -1616,7 +1616,7 @@ public TopicStats getStats(boolean getPreciseBacklog) { stats.storageSize = ledger.getTotalSize(); stats.backlogSize = ledger.getEstimatedBacklogSize(); stats.deduplicationStatus = messageDeduplication.getStatus().toString(); - + stats.topicEpoch = topicEpoch.orElse(null); return stats; } diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/PublisherStats.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/PublisherStats.java index a1e3f7c272da1..2cff042c5455e 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/PublisherStats.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/PublisherStats.java @@ -21,12 +21,16 @@ import static com.google.common.base.Preconditions.checkNotNull; import java.util.Map; +import org.apache.pulsar.client.api.ProducerAccessMode; + /** * Statistics about a publisher. */ public class PublisherStats { private int count; + public ProducerAccessMode accessMode; + /** Total rate of messages published by this publisher (msg/s). */ public double msgRateIn; diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/TopicStats.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/TopicStats.java index ebc0c41af878d..66ce2ea5e2e86 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/TopicStats.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/TopicStats.java @@ -77,6 +77,9 @@ public class TopicStats { public String deduplicationStatus; + /** The topic epoch or empty if not set. */ + public Long topicEpoch; + public TopicStats() { this.publishers = Lists.newArrayList(); this.subscriptions = Maps.newHashMap(); @@ -100,6 +103,7 @@ public void reset() { this.subscriptions.clear(); this.replication.clear(); this.deduplicationStatus = null; + this.topicEpoch = null; } // if the stats are added for the 1st time, we will need to make a copy of these stats and add it to the current diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java index edf308ed9ce25..dab73524022ea 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java @@ -2183,4 +2183,17 @@ private static PulsarApi.ProducerAccessMode convertProducerAccessMode(ProducerAc throw new IllegalArgumentException("Unknonw access mode: " + accessMode); } } + + public static ProducerAccessMode convertProducerAccessMode(PulsarApi.ProducerAccessMode accessMode) { + switch (accessMode) { + case Exclusive: + return ProducerAccessMode.Exclusive; + case Shared: + return ProducerAccessMode.Shared; +// case WaitForExclusive: +// return ProducerAccessMode.WaitForExclusive; + default: + throw new IllegalArgumentException("Unknonw access mode: " + accessMode); + } + } } From 69a0a8f91e6624b8f5809c448295b26fa586eb1b Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Tue, 1 Dec 2020 10:55:02 -0800 Subject: [PATCH 4/6] Moved constant to top of file --- .../pulsar/broker/service/persistent/PersistentTopic.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java index bc131ddba54f3..65f7323c105d7 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java @@ -151,6 +151,7 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal private final ConcurrentOpenHashMap replicators; static final String DEDUPLICATION_CURSOR_NAME = "pulsar.dedup"; + private static final String TOPIC_EPOCH_PROPERTY_NAME = "pulsar.topic.epoch"; private static final double MESSAGE_EXPIRY_THRESHOLD = 1.5; @@ -475,8 +476,6 @@ public CompletableFuture> addProducer(Producer producer) { }); } - private static final String TOPIC_EPOCH_PROPERTY_NAME = "pulsar.topic.epoch"; - protected CompletableFuture incrementTopicEpoch(Optional currentEpoch) { long newEpoch = currentEpoch.orElse(-1L) + 1; From 9c0dda252e4afa5ad541728ddbe93f641f4fa282 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Tue, 1 Dec 2020 17:02:16 -0800 Subject: [PATCH 5/6] Fix mistake in previous update --- .../java/org/apache/pulsar/broker/service/AbstractTopic.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java index be51770956a82..d1c7bb4495462 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java @@ -384,7 +384,7 @@ protected CompletableFuture> incrementTopicEpochIfNeeded(Producer case Exclusive: if (hasExclusiveProducer) { - return FutureUtil.failedFuture(new ProducerBusyException( + return FutureUtil.failedFuture(new ProducerFencedException( "Topic has an existing exclusive producer: " + producers.keys().nextElement())); } else if (!producers.isEmpty()) { return FutureUtil.failedFuture(new ProducerFencedException("Topic has existing shared producers")); From eba7bbe04d50dd09a111eae521aa32e970262d6d Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Wed, 2 Dec 2020 10:38:49 -0800 Subject: [PATCH 6/6] Added handling for topic deletion --- .../pulsar/broker/service/AbstractTopic.java | 15 ++++++++----- .../nonpersistent/NonPersistentTopic.java | 7 +++++++ .../service/persistent/PersistentTopic.java | 3 +++ .../broker/service/ExclusiveProducerTest.java | 21 +++++++++++++++++++ 4 files changed, 41 insertions(+), 5 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java index d1c7bb4495462..ce562165489e1 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java @@ -399,16 +399,19 @@ protected CompletableFuture> incrementTopicEpochIfNeeded(Producer // There are currently no existing producers hasExclusiveProducer = true; - CompletableFuture> future = incrementTopicEpoch(topicEpoch).thenApply(epoch -> { + CompletableFuture future; + if (producer.getTopicEpoch().isPresent()) { + future = setTopicEpoch(producer.getTopicEpoch().get()); + } else { + future = incrementTopicEpoch(topicEpoch); + } + return future.thenApply(epoch -> { topicEpoch = Optional.of(epoch); return topicEpoch; - }); - - future.exceptionally(ex -> { + }).exceptionally(ex -> { hasExclusiveProducer = false; return null; }); - return future; } // case WaitForExclusive: @@ -424,6 +427,8 @@ protected CompletableFuture> incrementTopicEpochIfNeeded(Producer } } + protected abstract CompletableFuture setTopicEpoch(long newEpoch); + protected abstract CompletableFuture incrementTopicEpoch(Optional currentEpoch); @Override 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 9012a20712dc8..f3094dbba94eb 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 @@ -195,6 +195,13 @@ protected CompletableFuture incrementTopicEpoch(Optional currentEpoc return CompletableFuture.completedFuture(currentEpoch.orElse(-1L) + 1); } + protected CompletableFuture setTopicEpoch(long newEpoch) { + // Non-persistent topic does not have any durable metadata, so we're just + // keeping the epoch in memory + return CompletableFuture.completedFuture(newEpoch); + } + + @Override public void checkMessageDeduplicationInfo() { // No-op diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java index 65f7323c105d7..a88eb6f9baa26 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java @@ -478,7 +478,10 @@ public CompletableFuture> addProducer(Producer producer) { protected CompletableFuture incrementTopicEpoch(Optional currentEpoch) { long newEpoch = currentEpoch.orElse(-1L) + 1; + return setTopicEpoch(newEpoch); + } + protected CompletableFuture setTopicEpoch(long newEpoch) { CompletableFuture future = new CompletableFuture<>(); ledger.asyncSetProperty(TOPIC_EPOCH_PROPERTY_NAME, String.valueOf(newEpoch), new UpdatePropertiesCallback() { @Override diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ExclusiveProducerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ExclusiveProducerTest.java index 4157fedc54064..0ab3b70f39439 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ExclusiveProducerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ExclusiveProducerTest.java @@ -179,6 +179,27 @@ public void producerFenced(boolean partitioned) throws Exception { } } + @Test(dataProvider = "topics") + public void topicDeleted(String type, boolean partitioned) throws Exception { + String topic = newTopic("persistent", partitioned); + + Producer p1 = pulsarClient.newProducer(Schema.STRING) + .topic(topic) + .accessMode(ProducerAccessMode.Exclusive) + .create(); + + p1.send("msg-1"); + + if (partitioned) { + admin.topics().deletePartitionedTopic(topic, true); + } else { + admin.topics().delete(topic, true); + } + + // The producer should be able to publish again on the topic + p1.send("msg-2"); + } + private String newTopic(String type, boolean isPartitioned) throws Exception { String topic = type + "://" + newTopicName(); if (isPartitioned) {