You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by te...@apache.org on 2024/01/08 13:00:34 UTC
(pulsar) branch master updated: [fix][broker] Fix potential NPE when disabling the liveness check (#21840)
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new aae0e9dfd55 [fix][broker] Fix potential NPE when disabling the liveness check (#21840)
aae0e9dfd55 is described below
commit aae0e9dfd5513d88d5246732fa6b5c7e394f29d6
Author: Qiang Zhao <ma...@apache.org>
AuthorDate: Mon Jan 8 21:00:27 2024 +0800
[fix][broker] Fix potential NPE when disabling the liveness check (#21840)
Co-authored-by: Jiwe Guo <te...@apache.org>
---
.../service/AbstractDispatcherSingleActiveConsumer.java | 2 +-
.../org/apache/pulsar/broker/service/AbstractTopic.java | 2 +-
.../java/org/apache/pulsar/broker/service/ServerCnx.java | 13 +++++++------
.../java/org/apache/pulsar/broker/service/TransportCnx.java | 4 ++--
.../apache/pulsar/broker/service/PersistentTopicTest.java | 2 +-
5 files changed, 12 insertions(+), 11 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java
index 4a8a805f16b..7726eb814a0 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java
@@ -170,7 +170,7 @@ public abstract class AbstractDispatcherSingleActiveConsumer extends AbstractBas
Consumer actConsumer = ACTIVE_CONSUMER_UPDATER.get(this);
if (actConsumer != null) {
return actConsumer.cnx().checkConnectionLiveness().thenCompose(actConsumerStillAlive -> {
- if (actConsumerStillAlive == null || actConsumerStillAlive) {
+ if (actConsumerStillAlive.isEmpty() || actConsumerStillAlive.get()) {
return FutureUtil.failedFuture(new ConsumerBusyException("Exclusive consumer is already"
+ " connected"));
} else {
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 2d32aac7d44..598480191d0 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
@@ -971,7 +971,7 @@ public abstract class AbstractTopic implements Topic, TopicPolicyListener<TopicP
// available. The producers related the connection that not available are automatically cleaned up.
if (!Objects.equals(oldProducer.getCnx(), newProducer.getCnx())) {
return oldProducer.getCnx().checkConnectionLiveness().thenCompose(previousIsActive -> {
- if (previousIsActive) {
+ if (previousIsActive.isEmpty() || previousIsActive.get()) {
return CompletableFuture.failedFuture(new BrokerServiceException.NamingException(
"Producer with name '" + newProducer.getProducerName()
+ "' is already connected to topic"));
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 467345e2c31..4e3eb9fd4ad 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
@@ -419,7 +419,7 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
// complete possible pending connection check future
if (connectionCheckInProgress != null && !connectionCheckInProgress.isDone()) {
- connectionCheckInProgress.complete(false);
+ connectionCheckInProgress.complete(Optional.of(false));
}
}
@@ -3454,16 +3454,17 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
return clientSourceAddressAndPort;
}
- CompletableFuture<Boolean> connectionCheckInProgress;
+ CompletableFuture<Optional<Boolean>> connectionCheckInProgress;
@Override
- public CompletableFuture<Boolean> checkConnectionLiveness() {
+ public CompletableFuture<Optional<Boolean>> checkConnectionLiveness() {
if (connectionLivenessCheckTimeoutMillis > 0) {
return NettyFutureUtil.toCompletableFuture(ctx.executor().submit(() -> {
if (connectionCheckInProgress != null) {
return connectionCheckInProgress;
} else {
- final CompletableFuture<Boolean> finalConnectionCheckInProgress = new CompletableFuture<>();
+ final CompletableFuture<Optional<Boolean>> finalConnectionCheckInProgress =
+ new CompletableFuture<>();
connectionCheckInProgress = finalConnectionCheckInProgress;
ctx.executor().schedule(() -> {
if (finalConnectionCheckInProgress == connectionCheckInProgress
@@ -3478,7 +3479,7 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
})).thenCompose(java.util.function.Function.identity());
} else {
// check is disabled
- return CompletableFuture.completedFuture((Boolean) null);
+ return CompletableFuture.completedFuture(Optional.empty());
}
}
@@ -3486,7 +3487,7 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
protected void messageReceived() {
super.messageReceived();
if (connectionCheckInProgress != null && !connectionCheckInProgress.isDone()) {
- connectionCheckInProgress.complete(true);
+ connectionCheckInProgress.complete(Optional.of(true));
connectionCheckInProgress = null;
}
}
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 6db085bd90a..eb2b318b7ea 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
@@ -82,9 +82,9 @@ public interface TransportCnx {
* by actively sending a Ping message to the client.
*
* @return a completable future where the result is true if the connection is alive, false otherwise. The result
- * is null if the connection liveness check is disabled.
+ * is empty if the connection liveness check is disabled.
*/
- CompletableFuture<Boolean> checkConnectionLiveness();
+ CompletableFuture<Optional<Boolean>> checkConnectionLiveness();
/**
* Increments the counter that controls the throttling of the connection by pausing reads.
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 04bf36eaa66..d5044276a5a 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
@@ -209,7 +209,7 @@ public class PersistentTopicTest extends MockedBookKeeperTestCase {
doReturn(eventLoopGroup.next()).when(channel).eventLoop();
doReturn(channel).when(ctx).channel();
doReturn(ctx).when(serverCnx).ctx();
- doReturn(CompletableFuture.completedFuture(true)).when(serverCnx).checkConnectionLiveness();
+ doReturn(CompletableFuture.completedFuture(Optional.of(true))).when(serverCnx).checkConnectionLiveness();
NamespaceService nsSvc = pulsarTestContext.getPulsarService().getNamespaceService();
NamespaceBundle bundle = mock(NamespaceBundle.class);