You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by he...@apache.org on 2023/11/06 01:33:30 UTC
(pulsar) branch master updated (720d6d24327 -> c30eae1296d)
This is an automated email from the ASF dual-hosted git repository.
heesung pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 720d6d24327 [fix][broker] Fix failure while creating non-durable cursor with inactive managed-ledger (#21508)
new e4a25156c7e [improve][broker] PIP-307 Added assignedBrokerUrl to CloseProducerCmd to skip lookups upon producer reconnections during unloading
new 6018b2bf8c0 Resolved comments
new c0eec1e46ed resolved comments
new c30eae1296d resolved comment
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../extensions/ExtensibleLoadManagerImpl.java | 49 ++++++++
.../channel/ServiceUnitStateChannel.java | 10 ++
.../channel/ServiceUnitStateChannelImpl.java | 63 ++++++++--
.../pulsar/broker/namespace/OwnedBundle.java | 2 +-
.../pulsar/broker/service/AbstractTopic.java | 5 +
.../pulsar/broker/service/BrokerService.java | 9 +-
.../org/apache/pulsar/broker/service/Producer.java | 9 +-
.../apache/pulsar/broker/service/ServerCnx.java | 38 +++++-
.../org/apache/pulsar/broker/service/Topic.java | 5 +
.../apache/pulsar/broker/service/TransportCnx.java | 3 +
.../service/nonpersistent/NonPersistentTopic.java | 26 +++-
.../broker/service/persistent/PersistentTopic.java | 33 ++++-
.../extensions/ExtensibleLoadManagerImplTest.java | 136 +++++++++++++++++++--
.../channel/ServiceUnitStateChannelTest.java | 1 +
.../broker/namespace/OwnershipCacheTest.java | 2 +-
.../org/apache/pulsar/client/impl/ClientCnx.java | 26 +++-
.../pulsar/client/impl/ConnectionHandler.java | 21 +++-
.../apache/pulsar/common/protocol/Commands.java | 26 +++-
pulsar-common/src/main/proto/PulsarApi.proto | 2 +
19 files changed, 420 insertions(+), 46 deletions(-)
(pulsar) 01/04: [improve][broker] PIP-307 Added assignedBrokerUrl to CloseProducerCmd to skip lookups upon producer reconnections during unloading
Posted by he...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
heesung pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit e4a25156c7e248e9b93d57b19e411a260f47bc98
Author: Heesung Sohn <he...@streamnative.io>
AuthorDate: Thu Oct 19 14:45:16 2023 -0700
[improve][broker] PIP-307 Added assignedBrokerUrl to CloseProducerCmd to skip lookups upon producer reconnections during unloading
---
.../extensions/ExtensibleLoadManagerImpl.java | 45 +++++++++
.../channel/ServiceUnitStateChannel.java | 10 ++
.../channel/ServiceUnitStateChannelImpl.java | 66 ++++++++++--
.../pulsar/broker/namespace/OwnedBundle.java | 2 +-
.../pulsar/broker/service/AbstractTopic.java | 5 +
.../pulsar/broker/service/BrokerService.java | 9 +-
.../org/apache/pulsar/broker/service/Producer.java | 9 +-
.../apache/pulsar/broker/service/ServerCnx.java | 39 +++++++-
.../org/apache/pulsar/broker/service/Topic.java | 5 +
.../apache/pulsar/broker/service/TransportCnx.java | 3 +
.../service/nonpersistent/NonPersistentTopic.java | 26 ++++-
.../broker/service/persistent/PersistentTopic.java | 33 ++++--
.../extensions/ExtensibleLoadManagerImplTest.java | 111 ++++++++++++++++++++-
.../channel/ServiceUnitStateChannelTest.java | 1 +
.../broker/namespace/OwnershipCacheTest.java | 2 +-
.../org/apache/pulsar/client/impl/ClientCnx.java | 23 ++++-
.../pulsar/client/impl/ConnectionHandler.java | 21 +++-
.../apache/pulsar/common/protocol/Commands.java | 26 ++++-
pulsar-common/src/main/proto/PulsarApi.proto | 2 +
19 files changed, 398 insertions(+), 40 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java
index d3119365ddf..8e166f00bc9 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java
@@ -38,6 +38,7 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.stream.Collectors;
@@ -178,6 +179,10 @@ public class ExtensibleLoadManagerImpl implements ExtensibleLoadManager {
private final UnloadCounter unloadCounter = new UnloadCounter();
private final SplitCounter splitCounter = new SplitCounter();
+ // Record the ignored send msg count during unloading
+ @Getter
+ private final AtomicLong ignoredSendMsgCounter = new AtomicLong();
+
// record unload metrics
private final AtomicReference<List<Metrics>> unloadMetrics = new AtomicReference<>();
// record split metrics
@@ -269,6 +274,15 @@ public class ExtensibleLoadManagerImpl implements ExtensibleLoadManager {
return loadManagerWrapper.get();
}
+ /**
+ * A static util func to get the ExtensibleLoadManagerImpl instance.
+ * @param pulsar PulsarService
+ * @return the ExtensibleLoadManagerImpl instance
+ */
+ public static ExtensibleLoadManagerImpl get(PulsarService pulsar) {
+ return get(pulsar.getLoadManager().get());
+ }
+
public static boolean debug(ServiceConfiguration config, Logger log) {
return config.isLoadBalancerDebugModeEnabled() || log.isDebugEnabled();
}
@@ -286,6 +300,37 @@ public class ExtensibleLoadManagerImpl implements ExtensibleLoadManager {
}
}
+ /**
+ * Gets the assigned broker for the given topic.
+ * @param pulsar PulsarService instance
+ * @param topic Topic Name
+ * @return the assigned broker's BrokerLookupData instance. Empty, if not assigned by Extensible LoadManager.
+ */
+ public static CompletableFuture<Optional<BrokerLookupData>> getAssignedBrokerLookupData(PulsarService pulsar,
+ String topic) {
+ if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar.getConfig())) {
+ var topicName = TopicName.get(topic);
+ try {
+ return pulsar.getNamespaceService().getBundleAsync(topicName)
+ .thenCompose(bundle -> {
+ var loadManager = ExtensibleLoadManagerImpl.get(pulsar);
+ var assigned = loadManager.getServiceUnitStateChannel()
+ .getAssigned(bundle.toString());
+ if (assigned.isPresent()) {
+ return loadManager.getBrokerRegistry().lookupAsync(assigned.get());
+ } else {
+ return CompletableFuture.completedFuture(Optional.empty());
+ }
+ }
+ );
+ } catch (Throwable e) {
+ log.error("Failed to DestinationBrokerLookupData for topic:{}", topic, e);
+ return CompletableFuture.completedFuture(Optional.empty());
+ }
+ }
+ return CompletableFuture.completedFuture(Optional.empty());
+ }
+
@Override
public void start() throws PulsarServerException {
if (this.started) {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannel.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannel.java
index 6e75fe91a91..9be76e1b0f4 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannel.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannel.java
@@ -131,6 +131,16 @@ public interface ServiceUnitStateChannel extends Closeable {
*/
CompletableFuture<Optional<String>> getOwnerAsync(String serviceUnit);
+ /**
+ * Gets the assigned broker of the service unit.
+ *
+ *
+ * @param serviceUnit (e.g. bundle))
+ * @return the future object of the assigned broker
+ */
+ Optional<String> getAssigned(String serviceUnit);
+
+
/**
* Checks if the target broker is the owner of the service unit.
*
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java
index 3cf16709cde..e5fa504e1db 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java
@@ -538,6 +538,44 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
}
}
+ @Override
+ public Optional<String> getAssigned(String serviceUnit) {
+ if (!validateChannelState(Started, true)) {
+ return Optional.empty();
+ }
+
+ ServiceUnitStateData data = tableview.get(serviceUnit);
+ if (data == null) {
+ return Optional.empty();
+ }
+ ServiceUnitState state = state(data);
+ switch (state) {
+ case Owned, Assigning -> {
+ return Optional.of(data.dstBroker());
+ }
+ case Releasing -> {
+ if (data.dstBroker() != null) {
+ return Optional.of(data.dstBroker());
+ }
+ return Optional.empty();
+ }
+ case Splitting -> {
+ return Optional.of(data.sourceBroker());
+ }
+ case Init, Free -> {
+ return Optional.empty();
+ }
+ case Deleted -> {
+ log.warn("Trying to get the assigned broker from the deleted serviceUnit:{}", serviceUnit);
+ return Optional.empty();
+ }
+ default -> {
+ log.warn("Trying to get the assigned broker from unknown state:{} serviceUnit:{}", state, serviceUnit);
+ return Optional.empty();
+ }
+ }
+ }
+
private long getNextVersionId(String serviceUnit) {
var data = tableview.get(serviceUnit);
return getNextVersionId(data);
@@ -732,15 +770,20 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
if (getOwnerRequest != null) {
getOwnerRequest.complete(data.dstBroker());
}
- stateChangeListeners.notify(serviceUnit, data, null);
+ CompletableFuture<Integer> ownFuture = null;
if (isTargetBroker(data.dstBroker())) {
- log(null, serviceUnit, data, null);
pulsar.getNamespaceService()
.onNamespaceBundleOwned(LoadManagerShared.getNamespaceBundle(pulsar, serviceUnit));
lastOwnEventHandledAt = System.currentTimeMillis();
- } else if (data.force() && isTargetBroker(data.sourceBroker())) {
- closeServiceUnit(serviceUnit);
+ ownFuture = CompletableFuture.completedFuture(null);
+ } else if ((data.force() || isTransferCommand(data)) && isTargetBroker(data.sourceBroker())) {
+ ownFuture = closeServiceUnit(serviceUnit, false);
+ } else {
+ ownFuture = CompletableFuture.completedFuture(null);
}
+
+ stateChangeListeners.notifyOnCompletion(ownFuture, serviceUnit, data)
+ .whenComplete((__, e) -> log(e, serviceUnit, data, null));
}
private void handleAssignEvent(String serviceUnit, ServiceUnitStateData data) {
@@ -755,15 +798,17 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
private void handleReleaseEvent(String serviceUnit, ServiceUnitStateData data) {
if (isTargetBroker(data.sourceBroker())) {
ServiceUnitStateData next;
+ CompletableFuture<Integer> unloadFuture;
if (isTransferCommand(data)) {
next = new ServiceUnitStateData(
Assigning, data.dstBroker(), data.sourceBroker(), getNextVersionId(data));
- // TODO: when close, pass message to clients to connect to the new broker
+ unloadFuture = closeServiceUnit(serviceUnit, true);
} else {
next = new ServiceUnitStateData(
Free, null, data.sourceBroker(), getNextVersionId(data));
+ unloadFuture = closeServiceUnit(serviceUnit, false);
}
- stateChangeListeners.notifyOnCompletion(closeServiceUnit(serviceUnit)
+ stateChangeListeners.notifyOnCompletion(unloadFuture
.thenCompose(__ -> pubAsync(serviceUnit, next)), serviceUnit, data)
.whenComplete((__, e) -> log(e, serviceUnit, data, next));
}
@@ -861,12 +906,13 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
}
}
- private CompletableFuture<Integer> closeServiceUnit(String serviceUnit) {
+ private CompletableFuture<Integer> closeServiceUnit(String serviceUnit, boolean closeWithoutDisconnectingClients) {
long startTime = System.nanoTime();
MutableInt unloadedTopics = new MutableInt();
NamespaceBundle bundle = LoadManagerShared.getNamespaceBundle(pulsar, serviceUnit);
return pulsar.getBrokerService().unloadServiceUnit(
bundle,
+ closeWithoutDisconnectingClients,
true,
pulsar.getConfig().getNamespaceBundleUnloadingTimeoutMs(),
TimeUnit.MILLISECONDS)
@@ -875,8 +921,10 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
return numUnloadedTopics;
})
.whenComplete((__, ex) -> {
- // clean up topics that failed to unload from the broker ownership cache
- pulsar.getBrokerService().cleanUnloadedTopicFromCache(bundle);
+ if (!closeWithoutDisconnectingClients) {
+ // clean up topics that failed to unload from the broker ownership cache
+ pulsar.getBrokerService().cleanUnloadedTopicFromCache(bundle);
+ }
pulsar.getNamespaceService().onNamespaceBundleUnload(bundle);
double unloadBundleTime = TimeUnit.NANOSECONDS
.toMillis((System.nanoTime() - startTime));
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/OwnedBundle.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/OwnedBundle.java
index e7cf23a0427..a87d45395db 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/OwnedBundle.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/OwnedBundle.java
@@ -136,7 +136,7 @@ public class OwnedBundle {
return pulsar.getNamespaceService().getOwnershipCache()
.updateBundleState(this.bundle, false)
.thenCompose(v -> pulsar.getBrokerService().unloadServiceUnit(
- bundle, closeWithoutWaitingClientDisconnect, timeout, timeoutUnit))
+ bundle, false, closeWithoutWaitingClientDisconnect, timeout, timeoutUnit))
.handle((numUnloadedTopics, ex) -> {
if (ex != null) {
// ignore topic-close failure to unload bundle
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 4e176c4fc0b..bc2b358200c 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
@@ -956,6 +956,11 @@ public abstract class AbstractTopic implements Topic, TopicPolicyListener<TopicP
}
}
+ @Override
+ public boolean isFenced() {
+ return isFenced;
+ }
+
protected CompletableFuture<Void> internalAddProducer(Producer producer) {
if (isProducersExceeded(producer)) {
log.warn("[{}] Attempting to add producer to topic which reached max producers limit", topic);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index 93337bafd90..e74801ca455 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -2197,8 +2197,10 @@ public class BrokerService implements Closeable {
}
public CompletableFuture<Integer> unloadServiceUnit(NamespaceBundle serviceUnit,
+ boolean closeWithoutDisconnectingClients,
boolean closeWithoutWaitingClientDisconnect, long timeout, TimeUnit unit) {
- CompletableFuture<Integer> future = unloadServiceUnit(serviceUnit, closeWithoutWaitingClientDisconnect);
+ CompletableFuture<Integer> future = unloadServiceUnit(
+ serviceUnit, closeWithoutDisconnectingClients, closeWithoutWaitingClientDisconnect);
ScheduledFuture<?> taskTimeout = executor().schedule(() -> {
if (!future.isDone()) {
log.warn("Unloading of {} has timed out", serviceUnit);
@@ -2215,11 +2217,13 @@ public class BrokerService implements Closeable {
* Unload all the topic served by the broker service under the given service unit.
*
* @param serviceUnit
+ * @param closeWithoutDisconnectingClients don't disconnect clients
* @param closeWithoutWaitingClientDisconnect don't wait for clients to disconnect
* and forcefully close managed-ledger
* @return
*/
private CompletableFuture<Integer> unloadServiceUnit(NamespaceBundle serviceUnit,
+ boolean closeWithoutDisconnectingClients,
boolean closeWithoutWaitingClientDisconnect) {
List<CompletableFuture<Void>> closeFutures = new ArrayList<>();
topics.forEach((name, topicFuture) -> {
@@ -2243,7 +2247,8 @@ public class BrokerService implements Closeable {
}
}
closeFutures.add(topicFuture
- .thenCompose(t -> t.isPresent() ? t.get().close(closeWithoutWaitingClientDisconnect)
+ .thenCompose(t -> t.isPresent() ? t.get().close(
+ closeWithoutDisconnectingClients, closeWithoutWaitingClientDisconnect)
: CompletableFuture.completedFuture(null)));
}
});
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 acaa7c02d19..1a4490d6b1b 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
@@ -39,6 +39,7 @@ import java.util.concurrent.atomic.AtomicLongFieldUpdater;
import org.apache.bookkeeper.mledger.Position;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.intercept.BrokerInterceptor;
+import org.apache.pulsar.broker.loadbalance.extensions.data.BrokerLookupData;
import org.apache.pulsar.broker.service.BrokerServiceException.TopicClosedException;
import org.apache.pulsar.broker.service.BrokerServiceException.TopicTerminatedException;
import org.apache.pulsar.broker.service.Topic.PublishContext;
@@ -699,17 +700,21 @@ public class Producer {
isDisconnecting.set(false);
}
+ public CompletableFuture<Void> disconnect() {
+ return disconnect(Optional.empty());
+ }
+
/**
* It closes the producer from server-side and sends command to client to disconnect producer from existing
* connection without closing that connection.
*
* @return Completable future indicating completion of producer close
*/
- public CompletableFuture<Void> disconnect() {
+ public CompletableFuture<Void> disconnect(Optional<BrokerLookupData> assignedBrokerLookupData) {
if (!closeFuture.isDone() && isDisconnecting.compareAndSet(false, true)) {
log.info("Disconnecting producer: {}", this);
cnx.execute(() -> {
- cnx.closeProducer(this);
+ cnx.closeProducer(this, assignedBrokerLookupData);
closeNow(true);
});
}
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 da51eaea8fb..4a43649e902 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
@@ -81,6 +81,8 @@ import org.apache.pulsar.broker.authentication.AuthenticationProvider;
import org.apache.pulsar.broker.authentication.AuthenticationState;
import org.apache.pulsar.broker.intercept.BrokerInterceptor;
import org.apache.pulsar.broker.limiter.ConnectionController;
+import org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl;
+import org.apache.pulsar.broker.loadbalance.extensions.data.BrokerLookupData;
import org.apache.pulsar.broker.service.BrokerServiceException.ConsumerBusyException;
import org.apache.pulsar.broker.service.BrokerServiceException.ServerMetadataException;
import org.apache.pulsar.broker.service.BrokerServiceException.ServiceUnitNotReadyException;
@@ -1598,7 +1600,7 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
remoteAddress, producerId);
}
producers.remove(producerId, producerFuture);
- closeProducer(producerId, -1L);
+ closeProducer(producerId, -1L, Optional.empty());
return null;
}
}
@@ -1760,6 +1762,20 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
printSendCommandDebug(send, headersAndPayload);
}
+
+ ServiceConfiguration conf = getBrokerService().pulsar().getConfiguration();
+ if (producer.getTopic().isFenced()
+ && ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(conf)) {
+ long ignoredMsgCount = ExtensibleLoadManagerImpl.get(getBrokerService().pulsar())
+ .getIgnoredSendMsgCounter().incrementAndGet();
+ if (log.isDebugEnabled()) {
+ log.debug("Ignored send msg from:{}:{} to fenced topic:{} during unloading."
+ + " Ignored message count:{}.",
+ remoteAddress, send.getProducerId(), producer.getTopic().getName(), ignoredMsgCount);
+ }
+ return;
+ }
+
if (producer.isNonPersistentTopic()) {
// avoid processing non-persist message if reached max concurrent-message limit
if (nonPersistentPendingMessages > maxNonPersistentPendingMessages) {
@@ -3020,13 +3036,26 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
public void closeProducer(Producer producer) {
// removes producer-connection from map and send close command to producer
safelyRemoveProducer(producer);
- closeProducer(producer.getProducerId(), producer.getEpoch());
+ closeProducer(producer.getProducerId(), producer.getEpoch(), Optional.empty());
+ }
+ @Override
+ public void closeProducer(Producer producer, Optional<BrokerLookupData> assignedBrokerLookupData) {
+ // removes producer-connection from map and send close command to producer
+ safelyRemoveProducer(producer);
+ closeProducer(producer.getProducerId(), producer.getEpoch(), assignedBrokerLookupData);
}
- public void closeProducer(long producerId, long epoch) {
+ private void closeProducer(long producerId, long epoch, Optional<BrokerLookupData> assignedBrokerLookupData) {
if (getRemoteEndpointProtocolVersion() >= v5.getValue()) {
- writeAndFlush(Commands.newCloseProducer(producerId, -1L));
+ if (assignedBrokerLookupData.isPresent()) {
+ writeAndFlush(Commands.newCloseProducer(producerId, -1L,
+ assignedBrokerLookupData.get().pulsarServiceUrl(),
+ assignedBrokerLookupData.get().pulsarServiceUrlTls()));
+ } else {
+ writeAndFlush(Commands.newCloseProducer(producerId, -1L));
+ }
+
// The client does not necessarily know that the producer is closed, but the connection is still
// active, and there could be messages in flight already. We want to ignore these messages for a time
// because they are expected. Once the interval has passed, the client should have received the
@@ -3049,7 +3078,7 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
closeConsumer(consumer.consumerId());
}
- public void closeConsumer(long consumerId) {
+ private void closeConsumer(long consumerId) {
if (getRemoteEndpointProtocolVersion() >= v5.getValue()) {
writeAndFlush(Commands.newCloseConsumer(consumerId, -1L));
} else {
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 c697639ff4f..41040a9f830 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
@@ -196,6 +196,9 @@ public interface Topic {
CompletableFuture<Void> close(boolean closeWithoutWaitingClientDisconnect);
+ CompletableFuture<Void> close(
+ boolean closeWithoutDisconnectingClients, boolean closeWithoutWaitingClientDisconnect);
+
void checkGC();
CompletableFuture<Void> checkClusterMigration();
@@ -331,6 +334,8 @@ public interface Topic {
boolean isPersistent();
+ boolean isFenced();
+
/* ------ Transaction related ------ */
/**
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TransportCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TransportCnx.java
index 94f934fec68..c09d63a9232 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TransportCnx.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TransportCnx.java
@@ -21,8 +21,10 @@ package org.apache.pulsar.broker.service;
import io.netty.handler.codec.haproxy.HAProxyMessage;
import io.netty.util.concurrent.Promise;
import java.net.SocketAddress;
+import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
+import org.apache.pulsar.broker.loadbalance.extensions.data.BrokerLookupData;
public interface TransportCnx {
@@ -57,6 +59,7 @@ public interface TransportCnx {
void removedProducer(Producer producer);
void closeProducer(Producer producer);
+ void closeProducer(Producer producer, Optional<BrokerLookupData> assignedBrokerLookupData);
void cancelPublishRateLimiting();
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 f3857b5ad2a..c299652adf8 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
@@ -40,6 +40,7 @@ import java.util.concurrent.atomic.AtomicLongFieldUpdater;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.Position;
import org.apache.pulsar.broker.PulsarServerException;
+import org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl;
import org.apache.pulsar.broker.namespace.NamespaceService;
import org.apache.pulsar.broker.resources.NamespaceResources;
import org.apache.pulsar.broker.service.AbstractReplicator;
@@ -484,14 +485,22 @@ public class NonPersistentTopic extends AbstractTopic implements Topic, TopicPol
return deleteFuture;
}
+
+ @Override
+ public CompletableFuture<Void> close(boolean closeWithoutWaitingClientDisconnect) {
+ return close(false, closeWithoutWaitingClientDisconnect);
+ }
+
/**
* Close this topic - close all producers and subscriptions associated with this topic.
*
+ * @param closeWithoutDisconnectingClients don't disconnect clients
* @param closeWithoutWaitingClientDisconnect don't wait for client disconnect and forcefully close managed-ledger
* @return Completable future indicating completion of close operation
*/
@Override
- public CompletableFuture<Void> close(boolean closeWithoutWaitingClientDisconnect) {
+ public CompletableFuture<Void> close(
+ boolean closeWithoutDisconnectingClients, boolean closeWithoutWaitingClientDisconnect) {
CompletableFuture<Void> closeFuture = new CompletableFuture<>();
lock.writeLock().lock();
@@ -510,7 +519,12 @@ public class NonPersistentTopic extends AbstractTopic implements Topic, TopicPol
List<CompletableFuture<Void>> futures = new ArrayList<>();
replicators.forEach((cluster, replicator) -> futures.add(replicator.disconnect()));
- producers.values().forEach(producer -> futures.add(producer.disconnect()));
+ if (!closeWithoutDisconnectingClients) {
+ futures.add(ExtensibleLoadManagerImpl.getAssignedBrokerLookupData(
+ brokerService.getPulsar(), topic).thenAccept(lookupData ->
+ producers.values().forEach(producer -> futures.add(producer.disconnect(lookupData)))
+ ));
+ }
if (topicPublishRateLimiter != null) {
topicPublishRateLimiter.close();
}
@@ -538,9 +552,13 @@ public class NonPersistentTopic extends AbstractTopic implements Topic, TopicPol
// unload topic iterates over topics map and removing from the map with the same thread creates deadlock.
// so, execute it in different thread
brokerService.executor().execute(() -> {
- brokerService.removeTopicFromCache(NonPersistentTopic.this);
- unregisterTopicPolicyListener();
+
+ if (!closeWithoutDisconnectingClients) {
+ brokerService.removeTopicFromCache(NonPersistentTopic.this);
+ unregisterTopicPolicyListener();
+ }
closeFuture.complete(null);
+
});
}).exceptionally(exception -> {
log.error("[{}] Error closing topic", topic, exception);
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 a8a921f3b62..66a8f9da0cc 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
@@ -84,6 +84,7 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.broker.PulsarServerException;
import org.apache.pulsar.broker.delayed.BucketDelayedDeliveryTrackerFactory;
import org.apache.pulsar.broker.delayed.DelayedDeliveryTrackerFactory;
+import org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl;
import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateChannelImpl;
import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateCompactionStrategy;
import org.apache.pulsar.broker.namespace.NamespaceService;
@@ -1448,17 +1449,24 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
}
public CompletableFuture<Void> close() {
- return close(false);
+ return close(false, false);
+ }
+
+ @Override
+ public CompletableFuture<Void> close(boolean closeWithoutWaitingClientDisconnect) {
+ return close(false, closeWithoutWaitingClientDisconnect);
}
/**
* Close this topic - close all producers and subscriptions associated with this topic.
*
+ * @param closeWithoutDisconnectingClients don't disconnect clients
* @param closeWithoutWaitingClientDisconnect don't wait for client disconnect and forcefully close managed-ledger
* @return Completable future indicating completion of close operation
*/
@Override
- public CompletableFuture<Void> close(boolean closeWithoutWaitingClientDisconnect) {
+ public CompletableFuture<Void> close(
+ boolean closeWithoutDisconnectingClients, boolean closeWithoutWaitingClientDisconnect) {
CompletableFuture<Void> closeFuture = new CompletableFuture<>();
lock.writeLock().lock();
@@ -1481,7 +1489,12 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
futures.add(transactionBuffer.closeAsync());
replicators.forEach((cluster, replicator) -> futures.add(replicator.disconnect()));
shadowReplicators.forEach((__, replicator) -> futures.add(replicator.disconnect()));
- producers.values().forEach(producer -> futures.add(producer.disconnect()));
+ if (!closeWithoutDisconnectingClients) {
+ futures.add(ExtensibleLoadManagerImpl.getAssignedBrokerLookupData(
+ brokerService.getPulsar(), topic).thenAccept(lookupData ->
+ producers.values().forEach(producer -> futures.add(producer.disconnect(lookupData)))
+ ));
+ }
if (topicPublishRateLimiter != null) {
topicPublishRateLimiter.close();
}
@@ -1518,14 +1531,22 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
ledger.asyncClose(new CloseCallback() {
@Override
public void closeComplete(Object ctx) {
- // Everything is now closed, remove the topic from map
- disposeTopic(closeFuture);
+ if (closeWithoutDisconnectingClients) {
+ closeFuture.complete(null);
+ } else {
+ // Everything is now closed, remove the topic from map
+ disposeTopic(closeFuture);
+ }
}
@Override
public void closeFailed(ManagedLedgerException exception, Object ctx) {
log.error("[{}] Failed to close managed ledger, proceeding anyway.", topic, exception);
- disposeTopic(closeFuture);
+ if (closeWithoutDisconnectingClients) {
+ closeFuture.complete(null);
+ } else {
+ disposeTopic(closeFuture);
+ }
}
}, null);
}).exceptionally(exception -> {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java
index c7cd55e183b..2c2d6653627 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java
@@ -69,6 +69,7 @@ import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.mutable.MutableInt;
import org.apache.commons.lang3.reflect.FieldUtils;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
@@ -98,6 +99,8 @@ import org.apache.pulsar.broker.namespace.NamespaceEphemeralData;
import org.apache.pulsar.broker.namespace.NamespaceService;
import org.apache.pulsar.broker.testcontext.PulsarTestContext;
import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.impl.LookupService;
import org.apache.pulsar.client.impl.TableViewImpl;
import org.apache.pulsar.common.naming.NamespaceBundle;
import org.apache.pulsar.common.naming.NamespaceName;
@@ -119,6 +122,7 @@ import org.mockito.MockedStatic;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
/**
@@ -126,6 +130,7 @@ import org.testng.annotations.Test;
*/
@Slf4j
@Test(groups = "broker")
+@SuppressWarnings("unchecked")
public class ExtensibleLoadManagerImplTest extends MockedPulsarServiceBaseTest {
private PulsarService pulsar1;
@@ -392,7 +397,7 @@ public class ExtensibleLoadManagerImplTest extends MockedPulsarServiceBaseTest {
admin.namespaces().unloadNamespaceBundle(topicName.getNamespace(), bundle.getBundleRange(), dstBrokerUrl);
Awaitility.await().untilAsserted(() -> {
assertEquals(onloadCount.get(), 3);
- assertEquals(unloadCount.get(), 2);
+ assertEquals(unloadCount.get(), 3); //one from releasing and one from owned
});
assertEquals(admin.lookups().lookupTopic(topicName.toString()), dstBrokerServiceUrl);
@@ -406,6 +411,110 @@ public class ExtensibleLoadManagerImplTest extends MockedPulsarServiceBaseTest {
assertTrue(ex.getMessage().contains("cannot be transfer to same broker"));
}
}
+ @DataProvider(name = "isPersistentTopicTest")
+ public Object[][] isPersistentTopicTest() {
+ return new Object[][] { { true }, { false }};
+ }
+ @Test(timeOut = 30 * 1000, dataProvider = "isPersistentTopicTest")
+ public void testTransferClientReconnectionWithoutLookup(boolean isPersistentTopicTest) throws Exception {
+ String topicType = isPersistentTopicTest? "persistent" : "non-persistent";
+ String topic = topicType + "://" + defaultTestNamespace + "/test-transfer-client-reconnect";
+ TopicName topicName = TopicName.get(topic);
+
+ AtomicInteger lookupCount = new AtomicInteger();
+ var lookup = spyLookupService(lookupCount, topicName);
+ var producer = pulsarClient.newProducer().topic(topic).create();
+ int lookupCountBeforeUnload = lookupCount.get();
+
+ NamespaceBundle bundle = getBundleAsync(pulsar1, TopicName.get(topic)).get();
+ String broker = admin.lookups().lookupTopic(topic);
+ String dstBrokerUrl = pulsar1.getLookupServiceAddress();
+ String dstBrokerServiceUrl;
+ if (broker.equals(pulsar1.getBrokerServiceUrl())) {
+ dstBrokerUrl = pulsar2.getLookupServiceAddress();
+ dstBrokerServiceUrl = pulsar2.getBrokerServiceUrl();
+ } else {
+ dstBrokerServiceUrl = pulsar1.getBrokerServiceUrl();
+ }
+ checkOwnershipState(broker, bundle);
+
+ final String finalDstBrokerUrl = dstBrokerUrl;
+ CompletableFuture.runAsync(() -> {
+ try {
+ admin.namespaces().unloadNamespaceBundle(
+ defaultTestNamespace, bundle.getBundleRange(), finalDstBrokerUrl);
+ } catch (PulsarAdminException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ );
+
+ Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> {
+ try {
+ producer.send("hi".getBytes());
+ String newOwner = admin.lookups().lookupTopic(topic);
+ assertEquals(dstBrokerServiceUrl, newOwner);
+ } catch (PulsarClientException e) {
+ throw new RuntimeException(e);
+ } catch (PulsarAdminException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ assertTrue(producer.isConnected());
+ verify(lookup, times(lookupCountBeforeUnload)).getBroker(topicName);
+ producer.close();
+ }
+
+
+
+ @Test(timeOut = 30 * 1000, dataProvider = "isPersistentTopicTest")
+ public void testUnloadClientReconnectionWithLookup(boolean isPersistentTopicTest) throws Exception {
+ String topicType = isPersistentTopicTest? "persistent" : "non-persistent";
+ String topic = topicType + "://" + defaultTestNamespace + "/test-unload-client-reconnect-"
+ + isPersistentTopicTest;
+ TopicName topicName = TopicName.get(topic);
+
+ AtomicInteger lookupCount = new AtomicInteger();
+ var lookup = spyLookupService(lookupCount, topicName);
+
+ var producer = pulsarClient.newProducer().topic(topic).create();
+ int lookupCountBeforeUnload = lookupCount.get();
+
+ NamespaceBundle bundle = getBundleAsync(pulsar1, TopicName.get(topic)).get();
+ CompletableFuture.runAsync(() -> {
+ try {
+ admin.namespaces().unloadNamespaceBundle(
+ defaultTestNamespace, bundle.getBundleRange());
+ } catch (PulsarAdminException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ );
+ MutableInt sendCount = new MutableInt();
+ Awaitility.await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+ try {
+ producer.send("hi".getBytes());
+ assertEquals(sendCount.incrementAndGet(), 10);
+ } catch (PulsarClientException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ assertTrue(producer.isConnected());
+ verify(lookup, times(lookupCountBeforeUnload + 1)).getBroker(topicName);
+ producer.close();
+ }
+
+ private LookupService spyLookupService(AtomicInteger lookupCount, TopicName topicName)
+ throws IllegalAccessException {
+ var lookup = spy((LookupService)
+ FieldUtils.readDeclaredField(pulsarClient, "lookup", true));
+ FieldUtils.writeDeclaredField(pulsarClient, "lookup", lookup, true);
+ doAnswer(invocationOnMock -> {
+ lookupCount.incrementAndGet();
+ return invocationOnMock.callRealMethod();
+ }).when(lookup).getBroker(topicName);
+ return lookup;
+ }
private void checkOwnershipState(String broker, NamespaceBundle bundle)
throws ExecutionException, InterruptedException {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelTest.java
index 31705264fba..df1bfd12d3e 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelTest.java
@@ -101,6 +101,7 @@ import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
@Test(groups = "broker")
+@SuppressWarnings("unchecked")
public class ServiceUnitStateChannelTest extends MockedPulsarServiceBaseTest {
private PulsarService pulsar1;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/OwnershipCacheTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/OwnershipCacheTest.java
index 9e3d9e3a413..c92127457aa 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/OwnershipCacheTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/OwnershipCacheTest.java
@@ -102,7 +102,7 @@ public class OwnershipCacheTest {
nsService = mock(NamespaceService.class);
brokerService = mock(BrokerService.class);
doReturn(CompletableFuture.completedFuture(1)).when(brokerService)
- .unloadServiceUnit(any(), anyBoolean(), anyLong(), any());
+ .unloadServiceUnit(any(), anyBoolean(), anyBoolean(), anyLong(), any());
doReturn(config).when(pulsar).getConfiguration();
doReturn(nsService).when(pulsar).getNamespaceService();
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 f3e8b2354b3..abdbe2f872c 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
@@ -34,6 +34,7 @@ import io.netty.handler.ssl.SslHandshakeCompletionEvent;
import io.netty.util.concurrent.Promise;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
+import java.net.URI;
import java.net.URISyntaxException;
import java.nio.channels.ClosedChannelException;
import java.util.Arrays;
@@ -801,11 +802,29 @@ public class ClientCnx extends PulsarHandler {
@Override
protected void handleCloseProducer(CommandCloseProducer closeProducer) {
- log.info("[{}] Broker notification of Closed producer: {}", remoteAddress, closeProducer.getProducerId());
final long producerId = closeProducer.getProducerId();
ProducerImpl<?> producer = producers.remove(producerId);
if (producer != null) {
- producer.connectionClosed(this);
+ if (closeProducer.hasAssignedBrokerServiceUrl() || closeProducer.hasAssignedBrokerServiceUrlTls()) {
+ try {
+ final URI uri = new URI(producer.client.conf.isUseTls()
+ ? closeProducer.getAssignedBrokerServiceUrlTls()
+ : closeProducer.getAssignedBrokerServiceUrl());
+ log.info("[{}] Broker notification of Closed producer: {}. Redirecting to {}.",
+ remoteAddress, closeProducer.getProducerId(), uri);
+ producer.getConnectionHandler().connectionClosed(this, 0L, Optional.of(uri));
+ } catch (URISyntaxException e) {
+ log.error("[{}] Invalid redirect url {}/{} for {}", remoteAddress,
+ closeProducer.getAssignedBrokerServiceUrl(),
+ closeProducer.getAssignedBrokerServiceUrlTls(),
+ closeProducer.getRequestId());
+ producer.connectionClosed(this);
+ }
+ } else {
+ log.info("[{}] Broker notification of Closed producer: {}.",
+ remoteAddress, closeProducer.getProducerId());
+ producer.connectionClosed(this);
+ }
} else {
log.warn("Producer with id {} not found while closing producer ", producerId);
}
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 6979914274e..90210803242 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
@@ -19,6 +19,8 @@
package org.apache.pulsar.client.impl;
import java.net.InetSocketAddress;
+import java.net.URI;
+import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -66,6 +68,10 @@ public class ConnectionHandler {
}
protected void grabCnx() {
+ grabCnx(Optional.empty());
+ }
+
+ protected void grabCnx(Optional<URI> hostURI) {
if (!duringConnect.compareAndSet(false, true)) {
log.info("[{}] [{}] Skip grabbing the connection since there is a pending connection",
state.topic, state.getHandlerName());
@@ -87,7 +93,12 @@ public class ConnectionHandler {
try {
CompletableFuture<ClientCnx> cnxFuture;
- if (state.redirectedClusterURI != null) {
+ if (hostURI.isPresent()) {
+ InetSocketAddress address = InetSocketAddress.createUnresolved(
+ hostURI.get().getHost(),
+ hostURI.get().getPort());
+ cnxFuture = state.client.getConnection(address, address, randomKeyForSelectConnection);
+ } else if (state.redirectedClusterURI != null) {
if (state.topic == null) {
InetSocketAddress address = InetSocketAddress.createUnresolved(state.redirectedClusterURI.getHost(),
state.redirectedClusterURI.getPort());
@@ -156,6 +167,10 @@ public class ConnectionHandler {
}
public void connectionClosed(ClientCnx cnx) {
+ connectionClosed(cnx, null, Optional.empty());
+ }
+
+ public void connectionClosed(ClientCnx cnx, Long initialConnectionDelayMs, Optional<URI> hostUrl) {
lastConnectionClosedTimestamp = System.currentTimeMillis();
duringConnect.set(false);
state.client.getCnxPool().releaseConnection(cnx);
@@ -165,14 +180,14 @@ public class ConnectionHandler {
state.topic, state.getHandlerName(), state.getState());
return;
}
- long delayMs = backoff.next();
+ long delayMs = initialConnectionDelayMs != null ? initialConnectionDelayMs.longValue() : backoff.next();
state.setState(State.Connecting);
log.info("[{}] [{}] Closed connection {} -- Will try again in {} s",
state.topic, state.getHandlerName(), cnx.channel(),
delayMs / 1000.0);
state.client.timer().newTimeout(timeout -> {
log.info("[{}] [{}] Reconnecting after timeout", state.topic, state.getHandlerName());
- grabCnx();
+ grabCnx(hostUrl);
}, delayMs, TimeUnit.MILLISECONDS);
}
}
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 cf0cd820a6d..ff116d2406b 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
@@ -60,6 +60,7 @@ import org.apache.pulsar.common.api.proto.CommandAddPartitionToTxnResponse;
import org.apache.pulsar.common.api.proto.CommandAddSubscriptionToTxn;
import org.apache.pulsar.common.api.proto.CommandAddSubscriptionToTxnResponse;
import org.apache.pulsar.common.api.proto.CommandAuthChallenge;
+import org.apache.pulsar.common.api.proto.CommandCloseProducer;
import org.apache.pulsar.common.api.proto.CommandConnect;
import org.apache.pulsar.common.api.proto.CommandConnected;
import org.apache.pulsar.common.api.proto.CommandEndTxnOnPartitionResponse;
@@ -761,11 +762,28 @@ public class Commands {
return serializeWithSize(cmd);
}
- public static ByteBuf newCloseProducer(long producerId, long requestId) {
+ public static ByteBuf newCloseProducer(
+ long producerId, long requestId) {
+ return newCloseProducer(producerId, requestId, null, null);
+ }
+
+ public static ByteBuf newCloseProducer(
+ long producerId, long requestId, String assignedBrokerUrl, String assignedBrokerUrlTls) {
BaseCommand cmd = localCmd(Type.CLOSE_PRODUCER);
- cmd.setCloseProducer()
- .setProducerId(producerId)
- .setRequestId(requestId);
+ CommandCloseProducer commandCloseProducer = cmd.setCloseProducer()
+ .setProducerId(producerId)
+ .setRequestId(requestId);
+
+ if (assignedBrokerUrl != null) {
+ commandCloseProducer
+ .setAssignedBrokerServiceUrl(assignedBrokerUrl);
+ }
+
+ if (assignedBrokerUrlTls != null){
+ commandCloseProducer
+ .setAssignedBrokerServiceUrlTls(assignedBrokerUrlTls);
+ }
+
return serializeWithSize(cmd);
}
diff --git a/pulsar-common/src/main/proto/PulsarApi.proto b/pulsar-common/src/main/proto/PulsarApi.proto
index afe193eeb7e..2c350aaf8a1 100644
--- a/pulsar-common/src/main/proto/PulsarApi.proto
+++ b/pulsar-common/src/main/proto/PulsarApi.proto
@@ -641,6 +641,8 @@ message CommandTopicMigrated {
message CommandCloseProducer {
required uint64 producer_id = 1;
required uint64 request_id = 2;
+ optional string assignedBrokerServiceUrl = 3;
+ optional string assignedBrokerServiceUrlTls = 4;
}
message CommandCloseConsumer {
(pulsar) 03/04: resolved comments
Posted by he...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
heesung pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit c0eec1e46edeb46c888fa28f27b199ea7e7a1574
Author: Heesung Sohn <he...@streamnative.io>
AuthorDate: Fri Nov 3 15:08:07 2023 -0700
resolved comments
---
.../extensions/ExtensibleLoadManagerImpl.java | 4 ++++
.../channel/ServiceUnitStateChannelImpl.java | 12 ++++++------
.../pulsar/broker/namespace/OwnedBundle.java | 2 +-
.../pulsar/broker/service/BrokerService.java | 10 +++++-----
.../apache/pulsar/broker/service/ServerCnx.java | 7 +++----
.../org/apache/pulsar/broker/service/Topic.java | 2 +-
.../service/nonpersistent/NonPersistentTopic.java | 10 +++++-----
.../broker/service/persistent/PersistentTopic.java | 22 +++++++++++-----------
.../extensions/ExtensibleLoadManagerImplTest.java | 20 +++++++++++++++-----
.../pulsar/client/impl/ConnectionHandler.java | 2 +-
10 files changed, 52 insertions(+), 39 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java
index 14c81a6a492..67bab9b12ff 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java
@@ -267,6 +267,10 @@ public class ExtensibleLoadManagerImpl implements ExtensibleLoadManager {
return ExtensibleLoadManagerImpl.class.getName().equals(conf.getLoadManagerClassName());
}
+ public static boolean isLoadManagerExtensionEnabled(PulsarService pulsar) {
+ return pulsar.getLoadManager().get() instanceof ExtensibleLoadManagerImpl;
+ }
+
public static ExtensibleLoadManagerImpl get(LoadManager loadManager) {
if (!(loadManager instanceof ExtensibleLoadManagerWrapper loadManagerWrapper)) {
throw new IllegalArgumentException("The load manager should be 'ExtensibleLoadManagerWrapper'.");
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java
index 1fe7b83d77d..cff45b18ec8 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java
@@ -776,7 +776,7 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
log(null, serviceUnit, data, null);
} else if ((data.force() || isTransferCommand(data)) && isTargetBroker(data.sourceBroker())) {
stateChangeListeners.notifyOnCompletion(
- closeServiceUnit(serviceUnit, false), serviceUnit, data)
+ closeServiceUnit(serviceUnit, true), serviceUnit, data)
.whenComplete((__, e) -> log(e, serviceUnit, data, null));
} else {
stateChangeListeners.notify(serviceUnit, data, null);
@@ -799,11 +799,11 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
if (isTransferCommand(data)) {
next = new ServiceUnitStateData(
Assigning, data.dstBroker(), data.sourceBroker(), getNextVersionId(data));
- unloadFuture = closeServiceUnit(serviceUnit, true);
+ unloadFuture = closeServiceUnit(serviceUnit, false);
} else {
next = new ServiceUnitStateData(
Free, null, data.sourceBroker(), getNextVersionId(data));
- unloadFuture = closeServiceUnit(serviceUnit, false);
+ unloadFuture = closeServiceUnit(serviceUnit, true);
}
stateChangeListeners.notifyOnCompletion(unloadFuture
.thenCompose(__ -> pubAsync(serviceUnit, next)), serviceUnit, data)
@@ -903,13 +903,13 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
}
}
- private CompletableFuture<Integer> closeServiceUnit(String serviceUnit, boolean closeWithoutDisconnectingClients) {
+ private CompletableFuture<Integer> closeServiceUnit(String serviceUnit, boolean disconnectClients) {
long startTime = System.nanoTime();
MutableInt unloadedTopics = new MutableInt();
NamespaceBundle bundle = LoadManagerShared.getNamespaceBundle(pulsar, serviceUnit);
return pulsar.getBrokerService().unloadServiceUnit(
bundle,
- closeWithoutDisconnectingClients,
+ disconnectClients,
true,
pulsar.getConfig().getNamespaceBundleUnloadingTimeoutMs(),
TimeUnit.MILLISECONDS)
@@ -918,7 +918,7 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
return numUnloadedTopics;
})
.whenComplete((__, ex) -> {
- if (!closeWithoutDisconnectingClients) {
+ if (disconnectClients) {
// clean up topics that failed to unload from the broker ownership cache
pulsar.getBrokerService().cleanUnloadedTopicFromCache(bundle);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/OwnedBundle.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/OwnedBundle.java
index a87d45395db..cdedac1136e 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/OwnedBundle.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/OwnedBundle.java
@@ -136,7 +136,7 @@ public class OwnedBundle {
return pulsar.getNamespaceService().getOwnershipCache()
.updateBundleState(this.bundle, false)
.thenCompose(v -> pulsar.getBrokerService().unloadServiceUnit(
- bundle, false, closeWithoutWaitingClientDisconnect, timeout, timeoutUnit))
+ bundle, true, closeWithoutWaitingClientDisconnect, timeout, timeoutUnit))
.handle((numUnloadedTopics, ex) -> {
if (ex != null) {
// ignore topic-close failure to unload bundle
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index e74801ca455..da556e4422f 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -2197,10 +2197,10 @@ public class BrokerService implements Closeable {
}
public CompletableFuture<Integer> unloadServiceUnit(NamespaceBundle serviceUnit,
- boolean closeWithoutDisconnectingClients,
+ boolean disconnectClients,
boolean closeWithoutWaitingClientDisconnect, long timeout, TimeUnit unit) {
CompletableFuture<Integer> future = unloadServiceUnit(
- serviceUnit, closeWithoutDisconnectingClients, closeWithoutWaitingClientDisconnect);
+ serviceUnit, disconnectClients, closeWithoutWaitingClientDisconnect);
ScheduledFuture<?> taskTimeout = executor().schedule(() -> {
if (!future.isDone()) {
log.warn("Unloading of {} has timed out", serviceUnit);
@@ -2217,13 +2217,13 @@ public class BrokerService implements Closeable {
* Unload all the topic served by the broker service under the given service unit.
*
* @param serviceUnit
- * @param closeWithoutDisconnectingClients don't disconnect clients
+ * @param disconnectClients disconnect clients
* @param closeWithoutWaitingClientDisconnect don't wait for clients to disconnect
* and forcefully close managed-ledger
* @return
*/
private CompletableFuture<Integer> unloadServiceUnit(NamespaceBundle serviceUnit,
- boolean closeWithoutDisconnectingClients,
+ boolean disconnectClients,
boolean closeWithoutWaitingClientDisconnect) {
List<CompletableFuture<Void>> closeFutures = new ArrayList<>();
topics.forEach((name, topicFuture) -> {
@@ -2248,7 +2248,7 @@ public class BrokerService implements Closeable {
}
closeFutures.add(topicFuture
.thenCompose(t -> t.isPresent() ? t.get().close(
- closeWithoutDisconnectingClients, closeWithoutWaitingClientDisconnect)
+ disconnectClients, closeWithoutWaitingClientDisconnect)
: CompletableFuture.completedFuture(null)));
}
});
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 4a43649e902..e0523846423 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
@@ -1762,11 +1762,10 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
printSendCommandDebug(send, headersAndPayload);
}
-
- ServiceConfiguration conf = getBrokerService().pulsar().getConfiguration();
+ PulsarService pulsar = getBrokerService().pulsar();
if (producer.getTopic().isFenced()
- && ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(conf)) {
- long ignoredMsgCount = ExtensibleLoadManagerImpl.get(getBrokerService().pulsar())
+ && ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar)) {
+ long ignoredMsgCount = ExtensibleLoadManagerImpl.get(pulsar)
.getIgnoredSendMsgCounter().incrementAndGet();
if (log.isDebugEnabled()) {
log.debug("Ignored send msg from:{}:{} to fenced topic:{} during unloading."
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 41040a9f830..6e2eb75a795 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
@@ -197,7 +197,7 @@ public interface Topic {
CompletableFuture<Void> close(boolean closeWithoutWaitingClientDisconnect);
CompletableFuture<Void> close(
- boolean closeWithoutDisconnectingClients, boolean closeWithoutWaitingClientDisconnect);
+ boolean disconnectClients, boolean closeWithoutWaitingClientDisconnect);
void checkGC();
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 c299652adf8..0d80de3aa6d 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
@@ -488,19 +488,19 @@ public class NonPersistentTopic extends AbstractTopic implements Topic, TopicPol
@Override
public CompletableFuture<Void> close(boolean closeWithoutWaitingClientDisconnect) {
- return close(false, closeWithoutWaitingClientDisconnect);
+ return close(true, closeWithoutWaitingClientDisconnect);
}
/**
* Close this topic - close all producers and subscriptions associated with this topic.
*
- * @param closeWithoutDisconnectingClients don't disconnect clients
+ * @param disconnectClients disconnect clients
* @param closeWithoutWaitingClientDisconnect don't wait for client disconnect and forcefully close managed-ledger
* @return Completable future indicating completion of close operation
*/
@Override
public CompletableFuture<Void> close(
- boolean closeWithoutDisconnectingClients, boolean closeWithoutWaitingClientDisconnect) {
+ boolean disconnectClients, boolean closeWithoutWaitingClientDisconnect) {
CompletableFuture<Void> closeFuture = new CompletableFuture<>();
lock.writeLock().lock();
@@ -519,7 +519,7 @@ public class NonPersistentTopic extends AbstractTopic implements Topic, TopicPol
List<CompletableFuture<Void>> futures = new ArrayList<>();
replicators.forEach((cluster, replicator) -> futures.add(replicator.disconnect()));
- if (!closeWithoutDisconnectingClients) {
+ if (disconnectClients) {
futures.add(ExtensibleLoadManagerImpl.getAssignedBrokerLookupData(
brokerService.getPulsar(), topic).thenAccept(lookupData ->
producers.values().forEach(producer -> futures.add(producer.disconnect(lookupData)))
@@ -553,7 +553,7 @@ public class NonPersistentTopic extends AbstractTopic implements Topic, TopicPol
// so, execute it in different thread
brokerService.executor().execute(() -> {
- if (!closeWithoutDisconnectingClients) {
+ if (disconnectClients) {
brokerService.removeTopicFromCache(NonPersistentTopic.this);
unregisterTopicPolicyListener();
}
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 66a8f9da0cc..ba40a75651d 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
@@ -1449,24 +1449,24 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
}
public CompletableFuture<Void> close() {
- return close(false, false);
+ return close(true, false);
}
@Override
public CompletableFuture<Void> close(boolean closeWithoutWaitingClientDisconnect) {
- return close(false, closeWithoutWaitingClientDisconnect);
+ return close(true, closeWithoutWaitingClientDisconnect);
}
/**
* Close this topic - close all producers and subscriptions associated with this topic.
*
- * @param closeWithoutDisconnectingClients don't disconnect clients
+ * @param disconnectClients disconnect clients
* @param closeWithoutWaitingClientDisconnect don't wait for client disconnect and forcefully close managed-ledger
* @return Completable future indicating completion of close operation
*/
@Override
public CompletableFuture<Void> close(
- boolean closeWithoutDisconnectingClients, boolean closeWithoutWaitingClientDisconnect) {
+ boolean disconnectClients, boolean closeWithoutWaitingClientDisconnect) {
CompletableFuture<Void> closeFuture = new CompletableFuture<>();
lock.writeLock().lock();
@@ -1489,7 +1489,7 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
futures.add(transactionBuffer.closeAsync());
replicators.forEach((cluster, replicator) -> futures.add(replicator.disconnect()));
shadowReplicators.forEach((__, replicator) -> futures.add(replicator.disconnect()));
- if (!closeWithoutDisconnectingClients) {
+ if (disconnectClients) {
futures.add(ExtensibleLoadManagerImpl.getAssignedBrokerLookupData(
brokerService.getPulsar(), topic).thenAccept(lookupData ->
producers.values().forEach(producer -> futures.add(producer.disconnect(lookupData)))
@@ -1531,21 +1531,21 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
ledger.asyncClose(new CloseCallback() {
@Override
public void closeComplete(Object ctx) {
- if (closeWithoutDisconnectingClients) {
- closeFuture.complete(null);
- } else {
+ if (disconnectClients) {
// Everything is now closed, remove the topic from map
disposeTopic(closeFuture);
+ } else {
+ closeFuture.complete(null);
}
}
@Override
public void closeFailed(ManagedLedgerException exception, Object ctx) {
log.error("[{}] Failed to close managed ledger, proceeding anyway.", topic, exception);
- if (closeWithoutDisconnectingClients) {
- closeFuture.complete(null);
- } else {
+ if (disconnectClients) {
disposeTopic(closeFuture);
+ } else {
+ closeFuture.complete(null);
}
}
}, null);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java
index b8873ebb6bb..158488bc84a 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java
@@ -662,9 +662,11 @@ public class ExtensibleLoadManagerImplTest extends MockedPulsarServiceBaseTest {
@Test
public void testMoreThenOneFilter() throws Exception {
- TopicName topicName = TopicName.get(defaultTestNamespace + "/test-filter-has-exception");
+ // Use a different namespace to avoid flaky test failures
+ // from unloading the default namespace and the following topic policy lookups at the init state step
+ String namespace = "public/my-namespace";
+ TopicName topicName = TopicName.get(namespace + "/test-filter-has-exception");
NamespaceBundle bundle = getBundleAsync(pulsar1, topicName).get();
-
String lookupServiceAddress1 = pulsar1.getLookupServiceAddress();
doReturn(List.of(new MockBrokerFilter() {
@Override
@@ -682,10 +684,18 @@ public class ExtensibleLoadManagerImplTest extends MockedPulsarServiceBaseTest {
return FutureUtil.failedFuture(new BrokerFilterException("Test"));
}
})).when(primaryLoadManager).getBrokerFilterPipeline();
-
+ admin.namespaces().createNamespace(namespace);
Optional<BrokerLookupData> brokerLookupData = primaryLoadManager.assign(Optional.empty(), bundle).get();
- assertTrue(brokerLookupData.isPresent());
- assertEquals(brokerLookupData.get().getWebServiceUrl(), pulsar2.getWebServiceAddress());
+ Awaitility.waitAtMost(5, TimeUnit.SECONDS).untilAsserted(() -> {
+ assertTrue(brokerLookupData.isPresent());
+ assertEquals(brokerLookupData.get().getWebServiceUrl(), pulsar2.getWebServiceAddress());
+ assertEquals(brokerLookupData.get().getPulsarServiceUrl(),
+ pulsar1.getAdminClient().lookups().lookupTopic(topicName.toString()));
+ assertEquals(brokerLookupData.get().getPulsarServiceUrl(),
+ pulsar2.getAdminClient().lookups().lookupTopic(topicName.toString()));
+ });
+
+ admin.namespaces().deleteNamespace(namespace, true);
}
@Test
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 7361c27155c..600dc17a1b0 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
@@ -167,7 +167,7 @@ public class ConnectionHandler {
}
public void connectionClosed(ClientCnx cnx) {
- connectionClosed(cnx, null, Optional.empty());
+ connectionClosed(cnx, Optional.empty(), Optional.empty());
}
public void connectionClosed(ClientCnx cnx, Optional<Long> initialConnectionDelayMs, Optional<URI> hostUrl) {
(pulsar) 02/04: Resolved comments
Posted by he...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
heesung pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 6018b2bf8c0b6c04680518f873579aa601753086
Author: Heesung Sohn <he...@streamnative.io>
AuthorDate: Fri Oct 27 14:51:43 2023 -0700
Resolved comments
---
.../extensions/ExtensibleLoadManagerImpl.java | 2 +-
.../channel/ServiceUnitStateChannelImpl.java | 19 ++++++++-----------
.../extensions/ExtensibleLoadManagerImplTest.java | 9 ++++++---
.../java/org/apache/pulsar/client/impl/ClientCnx.java | 9 ++++++---
.../apache/pulsar/client/impl/ConnectionHandler.java | 4 ++--
5 files changed, 23 insertions(+), 20 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java
index 8e166f00bc9..14c81a6a492 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java
@@ -324,7 +324,7 @@ public class ExtensibleLoadManagerImpl implements ExtensibleLoadManager {
}
);
} catch (Throwable e) {
- log.error("Failed to DestinationBrokerLookupData for topic:{}", topic, e);
+ log.error("Failed to lookup destination broker for topic:{}", topic, e);
return CompletableFuture.completedFuture(Optional.empty());
}
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java
index e5fa504e1db..1fe7b83d77d 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java
@@ -554,10 +554,7 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
return Optional.of(data.dstBroker());
}
case Releasing -> {
- if (data.dstBroker() != null) {
- return Optional.of(data.dstBroker());
- }
- return Optional.empty();
+ return Optional.ofNullable(data.dstBroker());
}
case Splitting -> {
return Optional.of(data.sourceBroker());
@@ -770,20 +767,20 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
if (getOwnerRequest != null) {
getOwnerRequest.complete(data.dstBroker());
}
- CompletableFuture<Integer> ownFuture = null;
+
if (isTargetBroker(data.dstBroker())) {
pulsar.getNamespaceService()
.onNamespaceBundleOwned(LoadManagerShared.getNamespaceBundle(pulsar, serviceUnit));
lastOwnEventHandledAt = System.currentTimeMillis();
- ownFuture = CompletableFuture.completedFuture(null);
+ stateChangeListeners.notify(serviceUnit, data, null);
+ log(null, serviceUnit, data, null);
} else if ((data.force() || isTransferCommand(data)) && isTargetBroker(data.sourceBroker())) {
- ownFuture = closeServiceUnit(serviceUnit, false);
+ stateChangeListeners.notifyOnCompletion(
+ closeServiceUnit(serviceUnit, false), serviceUnit, data)
+ .whenComplete((__, e) -> log(e, serviceUnit, data, null));
} else {
- ownFuture = CompletableFuture.completedFuture(null);
+ stateChangeListeners.notify(serviceUnit, data, null);
}
-
- stateChangeListeners.notifyOnCompletion(ownFuture, serviceUnit, data)
- .whenComplete((__, e) -> log(e, serviceUnit, data, null));
}
private void handleAssignEvent(String serviceUnit, ServiceUnitStateData data) {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java
index 2c2d6653627..b8873ebb6bb 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java
@@ -147,6 +147,8 @@ public class ExtensibleLoadManagerImplTest extends MockedPulsarServiceBaseTest {
private final String defaultTestNamespace = "public/test";
+ private LookupService lookupService;
+
@BeforeClass
@Override
public void setup() throws Exception {
@@ -195,6 +197,7 @@ public class ExtensibleLoadManagerImplTest extends MockedPulsarServiceBaseTest {
admin.namespaces().createNamespace(defaultTestNamespace);
admin.namespaces().setNamespaceReplicationClusters(defaultTestNamespace,
Sets.newHashSet(this.conf.getClusterName()));
+ lookupService = (LookupService) FieldUtils.readDeclaredField(pulsarClient, "lookup", true);
}
}
@@ -208,9 +211,10 @@ public class ExtensibleLoadManagerImplTest extends MockedPulsarServiceBaseTest {
}
@BeforeMethod(alwaysRun = true)
- protected void initializeState() throws PulsarAdminException {
+ protected void initializeState() throws PulsarAdminException, IllegalAccessException {
admin.namespaces().unload(defaultTestNamespace);
reset(primaryLoadManager, secondaryLoadManager);
+ FieldUtils.writeDeclaredField(pulsarClient, "lookup", lookupService, true);
}
@Test
@@ -506,8 +510,7 @@ public class ExtensibleLoadManagerImplTest extends MockedPulsarServiceBaseTest {
private LookupService spyLookupService(AtomicInteger lookupCount, TopicName topicName)
throws IllegalAccessException {
- var lookup = spy((LookupService)
- FieldUtils.readDeclaredField(pulsarClient, "lookup", true));
+ var lookup = spy(lookupService);
FieldUtils.writeDeclaredField(pulsarClient, "lookup", lookup, true);
doAnswer(invocationOnMock -> {
lookupCount.incrementAndGet();
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 abdbe2f872c..794ef318bba 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
@@ -812,11 +812,14 @@ public class ClientCnx extends PulsarHandler {
: closeProducer.getAssignedBrokerServiceUrl());
log.info("[{}] Broker notification of Closed producer: {}. Redirecting to {}.",
remoteAddress, closeProducer.getProducerId(), uri);
- producer.getConnectionHandler().connectionClosed(this, 0L, Optional.of(uri));
+ producer.getConnectionHandler().connectionClosed(
+ this, Optional.of(0L), Optional.of(uri));
} catch (URISyntaxException e) {
log.error("[{}] Invalid redirect url {}/{} for {}", remoteAddress,
- closeProducer.getAssignedBrokerServiceUrl(),
- closeProducer.getAssignedBrokerServiceUrlTls(),
+ closeProducer.hasAssignedBrokerServiceUrl()
+ ? closeProducer.getAssignedBrokerServiceUrl() : "",
+ closeProducer.hasAssignedBrokerServiceUrlTls()
+ ? closeProducer.getAssignedBrokerServiceUrlTls() : "",
closeProducer.getRequestId());
producer.connectionClosed(this);
}
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 90210803242..7361c27155c 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
@@ -170,7 +170,7 @@ public class ConnectionHandler {
connectionClosed(cnx, null, Optional.empty());
}
- public void connectionClosed(ClientCnx cnx, Long initialConnectionDelayMs, Optional<URI> hostUrl) {
+ public void connectionClosed(ClientCnx cnx, Optional<Long> initialConnectionDelayMs, Optional<URI> hostUrl) {
lastConnectionClosedTimestamp = System.currentTimeMillis();
duringConnect.set(false);
state.client.getCnxPool().releaseConnection(cnx);
@@ -180,7 +180,7 @@ public class ConnectionHandler {
state.topic, state.getHandlerName(), state.getState());
return;
}
- long delayMs = initialConnectionDelayMs != null ? initialConnectionDelayMs.longValue() : backoff.next();
+ long delayMs = initialConnectionDelayMs.orElse(backoff.next());
state.setState(State.Connecting);
log.info("[{}] [{}] Closed connection {} -- Will try again in {} s",
state.topic, state.getHandlerName(), cnx.channel(),
(pulsar) 04/04: resolved comment
Posted by he...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
heesung pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit c30eae1296d3ce3d23811f39dac29da9dfd13bed
Author: Heesung Sohn <he...@streamnative.io>
AuthorDate: Sat Nov 4 14:58:43 2023 -0700
resolved comment
---
.../src/main/java/org/apache/pulsar/client/impl/ClientCnx.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
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 794ef318bba..ba79f1b8247 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
@@ -814,13 +814,13 @@ public class ClientCnx extends PulsarHandler {
remoteAddress, closeProducer.getProducerId(), uri);
producer.getConnectionHandler().connectionClosed(
this, Optional.of(0L), Optional.of(uri));
- } catch (URISyntaxException e) {
+ } catch (Throwable e) {
log.error("[{}] Invalid redirect url {}/{} for {}", remoteAddress,
closeProducer.hasAssignedBrokerServiceUrl()
? closeProducer.getAssignedBrokerServiceUrl() : "",
closeProducer.hasAssignedBrokerServiceUrlTls()
? closeProducer.getAssignedBrokerServiceUrlTls() : "",
- closeProducer.getRequestId());
+ closeProducer.getRequestId(), e);
producer.connectionClosed(this);
}
} else {