You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2025/07/17 10:38:44 UTC
(pulsar) branch branch-3.3 updated (84c56edae10 -> ab559c6b349)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 84c56edae10 [improve][broker]Rmove unused imports
new a197f46b85d [improve][broker] Improve the log when namespace bundle is not available (#24434)
new 5dbe026acd1 [fix][proxy] Fix proxy OOM by replacing TopicName with a simple conversion method (#24465)
new 2ee969a044b [fix][client] NPE in MultiTopicsConsumerImpl.negativeAcknowledge (#24476)
new 128f705878a [fix][proxy] Fix default value of connectionMaxIdleSeconds in Pulsar Proxy (#24529)
new c42ba7ffd09 [fix][client] Fix issue in auto releasing of idle connection with topics pattern consumer (#24528)
new ab559c6b349 [improve][misc] Optimize topic list hashing so that potentially large String allocation is avoided (#24525)
The 6 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:
.../pulsar/broker/namespace/NamespaceService.java | 13 +-
.../pulsar/broker/service/BrokerService.java | 12 +-
.../apache/pulsar/broker/MultiBrokerBaseTest.java | 14 ++
.../PersistentDispatcherFailoverConsumerTest.java | 3 +-
.../service/PersistentTopicConcurrentTest.java | 3 +-
.../pulsar/broker/service/PersistentTopicTest.java | 3 +-
.../pulsar/broker/service/ServerCnxTest.java | 6 +-
.../impl/AutoCloseUselessClientConSupports.java | 2 +-
...eUselessClientConTopicsPatternConsumerTest.java | 107 +++++++++++++++
.../org/apache/pulsar/client/impl/ClientCnx.java | 3 +
.../pulsar/client/impl/ClientCnxIdleState.java | 7 +-
.../client/impl/MultiTopicsConsumerImpl.java | 10 ++
.../pulsar/client/impl/ClientCnxIdleStateTest.java | 61 +++++++++
.../apache/pulsar/client/impl/ClientCnxTest.java | 11 ++
.../pulsar/common/api/raw/MessageParser.java | 19 ++-
.../org/apache/pulsar/common/naming/TopicName.java | 75 ++++++++++
.../org/apache/pulsar/common/topics/TopicList.java | 20 ++-
.../apache/pulsar/common/naming/TopicNameTest.java | 34 ++++-
.../apache/pulsar/common/topics/TopicListTest.java | 19 ++-
.../pulsar/proxy/server/LookupProxyHandler.java | 6 +-
.../pulsar/proxy/server/ParserProxyHandler.java | 14 +-
.../pulsar/proxy/server/ProxyConnection.java | 4 +-
.../impl/AutoCloseUselessClientConProxyTest.java | 151 +++++++++++++++++++++
.../org/apache/pulsar/proxy/server/ProxyTest.java | 30 ++++
24 files changed, 579 insertions(+), 48 deletions(-)
create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/client/impl/AutoCloseUselessClientConTopicsPatternConsumerTest.java
create mode 100644 pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxIdleStateTest.java
create mode 100644 pulsar-proxy/src/test/java/org/apache/pulsar/client/impl/AutoCloseUselessClientConProxyTest.java
(pulsar) 02/06: [fix][proxy] Fix proxy OOM by replacing TopicName with a simple conversion method (#24465)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 5dbe026acd16927f50443b578fe9a65bb0ed3cb1
Author: Yunze Xu <xy...@163.com>
AuthorDate: Wed Jul 2 09:51:16 2025 +0800
[fix][proxy] Fix proxy OOM by replacing TopicName with a simple conversion method (#24465)
(cherry picked from commit ae050b9c44ef0be474cedcf78c4dd6ba31cf6cc8)
---
.../pulsar/common/api/raw/MessageParser.java | 19 +++++-
.../org/apache/pulsar/common/naming/TopicName.java | 75 ++++++++++++++++++++++
.../apache/pulsar/common/naming/TopicNameTest.java | 34 +++++++++-
.../pulsar/proxy/server/LookupProxyHandler.java | 6 +-
.../pulsar/proxy/server/ParserProxyHandler.java | 14 ++--
5 files changed, 133 insertions(+), 15 deletions(-)
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/api/raw/MessageParser.java b/pulsar-common/src/main/java/org/apache/pulsar/common/api/raw/MessageParser.java
index af516fa7534..0e9aae4603d 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/api/raw/MessageParser.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/api/raw/MessageParser.java
@@ -55,11 +55,17 @@ public class MessageParser {
void process(RawMessage message) throws IOException;
}
+ @Deprecated
+ public static void parseMessage(TopicName topicName, long ledgerId, long entryId, ByteBuf headersAndPayload,
+ MessageProcessor processor, int maxMessageSize) throws IOException {
+ parseMessage(topicName.toString(), ledgerId, entryId, headersAndPayload, processor, maxMessageSize);
+ }
+
/**
* Parse a raw Pulsar entry payload and extract all the individual message that may be included in the batch. The
* provided {@link MessageProcessor} will be invoked for each individual message.
*/
- public static void parseMessage(TopicName topicName, long ledgerId, long entryId, ByteBuf headersAndPayload,
+ public static void parseMessage(String topicName, long ledgerId, long entryId, ByteBuf headersAndPayload,
MessageProcessor processor, int maxMessageSize) throws IOException {
ByteBuf payload = headersAndPayload;
ByteBuf uncompressedPayload = null;
@@ -117,7 +123,7 @@ public class MessageParser {
}
}
- public static boolean verifyChecksum(TopicName topic, ByteBuf headersAndPayload, long ledgerId, long entryId) {
+ public static boolean verifyChecksum(String topic, ByteBuf headersAndPayload, long ledgerId, long entryId) {
if (hasChecksum(headersAndPayload)) {
int checksum = readChecksum(headersAndPayload);
int computedChecksum = computeChecksum(headersAndPayload);
@@ -132,7 +138,14 @@ public class MessageParser {
return true;
}
- public static ByteBuf uncompressPayloadIfNeeded(TopicName topic, MessageMetadata msgMetadata,
+ @Deprecated
+ public static ByteBuf uncompressPayloadIfNeeded(TopicName topicName, MessageMetadata msgMetadata,
+ ByteBuf payload, long ledgerId, long entryId, int maxMessageSize) {
+ return uncompressPayloadIfNeeded(topicName.toString(), msgMetadata, payload, ledgerId, entryId,
+ maxMessageSize);
+ }
+
+ public static ByteBuf uncompressPayloadIfNeeded(String topic, MessageMetadata msgMetadata,
ByteBuf payload, long ledgerId, long entryId, int maxMessageSize) {
CompressionCodec codec = CompressionCodecProvider.getCompressionCodec(msgMetadata.getCompression());
int uncompressedSize = msgMetadata.getUncompressedSize();
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java b/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java
index aadb9d15af6..30c7da14459 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java
@@ -20,6 +20,7 @@ package org.apache.pulsar.common.naming;
import com.google.common.base.Splitter;
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
@@ -442,4 +443,78 @@ public class TopicName implements ServiceUnitId {
public boolean isV2() {
return cluster == null;
}
+
+ /**
+ * Convert a topic name to a full topic name.
+ * In Pulsar, a full topic name is "<domain>://<tenant>/<namespace>/<local-topic>" (v2) or
+ * "<domain>://<tenant>/<cluster>/<namespace>/<local-topic>" (v1). However, for convenient, it's allowed for clients
+ * to pass a short topic name with v2 format:
+ * - "<local-topic>", which represents "persistent://public/default/<local-topic>"
+ * - "<tenant>/<namespace>/<local-topic>, which represents "persistent://<tenant>/<namespace>/<local-topic>"
+ *
+ * @param topic the topic name from client
+ * @return the full topic name.
+ */
+ public static String toFullTopicName(String topic) {
+ final int index = topic.indexOf("://");
+ if (index >= 0) {
+ TopicDomain.getEnum(topic.substring(0, index));
+ final List<String> parts = splitBySlash(topic.substring(index + "://".length()), 4);
+ if (parts.size() != 3 && parts.size() != 4) {
+ throw new IllegalArgumentException(topic + " is invalid");
+ }
+ if (parts.size() == 3) {
+ NamespaceName.validateNamespaceName(parts.get(0), parts.get(1));
+ if (StringUtils.isBlank(parts.get(2))) {
+ throw new IllegalArgumentException(topic + " has blank local topic");
+ }
+ } else {
+ NamespaceName.validateNamespaceName(parts.get(0), parts.get(1), parts.get(2));
+ if (StringUtils.isBlank(parts.get(3))) {
+ throw new IllegalArgumentException(topic + " has blank local topic");
+ }
+ }
+ return topic; // it's a valid full topic name
+ } else {
+ List<String> parts = splitBySlash(topic, 0);
+ if (parts.size() != 1 && parts.size() != 3) {
+ throw new IllegalArgumentException(topic + " is invalid");
+ }
+ if (parts.size() == 1) {
+ if (StringUtils.isBlank(parts.get(0))) {
+ throw new IllegalArgumentException(topic + " has blank local topic");
+ }
+ return "persistent://public/default/" + parts.get(0);
+ } else {
+ NamespaceName.validateNamespaceName(parts.get(0), parts.get(1));
+ if (StringUtils.isBlank(parts.get(2))) {
+ throw new IllegalArgumentException(topic + " has blank local topic");
+ }
+ return "persistent://" + topic;
+ }
+ }
+ }
+
+ private static List<String> splitBySlash(String topic, int limit) {
+ final List<String> tokens = new ArrayList<>(3);
+ final int loopCount = (limit <= 0) ? Integer.MAX_VALUE : limit - 1;
+ int beginIndex = 0;
+ for (int i = 0; i < loopCount; i++) {
+ final int endIndex = topic.indexOf('/', beginIndex);
+ if (endIndex < 0) {
+ tokens.add(topic.substring(beginIndex));
+ return tokens;
+ } else if (endIndex > beginIndex) {
+ tokens.add(topic.substring(beginIndex, endIndex));
+ } else {
+ throw new IllegalArgumentException("Invalid topic name " + topic);
+ }
+ beginIndex = endIndex + 1;
+ }
+ if (beginIndex >= topic.length()) {
+ throw new IllegalArgumentException("Invalid topic name " + topic);
+ }
+ tokens.add(topic.substring(beginIndex));
+ return tokens;
+ }
}
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/naming/TopicNameTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/naming/TopicNameTest.java
index 27eb82d15af..bb4798fca46 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/naming/TopicNameTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/naming/TopicNameTest.java
@@ -22,6 +22,7 @@ import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotEquals;
import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertThrows;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
import org.apache.pulsar.common.util.Codec;
@@ -52,9 +53,8 @@ public class TopicNameTest {
assertEquals(TopicName.get("persistent://tenant/cluster/namespace/topic").toString(),
"persistent://tenant/cluster/namespace/topic");
-
- assertNotEquals(TopicName.get("persistent://tenant/cluster/namespace/topic"),
- "persistent://tenant/cluster/namespace/topic");
+ assertEquals(TopicName.toFullTopicName("persistent://tenant/cluster/namespace/topic"),
+ "persistent://tenant/cluster/namespace/topic");
assertEquals(TopicName.get("persistent://tenant/cluster/namespace/topic").getDomain(),
TopicDomain.persistent);
@@ -103,6 +103,7 @@ public class TopicNameTest {
} catch (IllegalArgumentException e) {
// Ok
}
+ assertThrows(IllegalArgumentException.class, () -> TopicName.toFullTopicName("://tenant.namespace:my-topic"));
try {
TopicName.get("://tenant.namespace");
@@ -110,6 +111,7 @@ public class TopicNameTest {
} catch (IllegalArgumentException e) {
// Ok
}
+ assertThrows(IllegalArgumentException.class, () -> TopicName.toFullTopicName("://tenant.namespace"));
try {
TopicName.get("invalid://tenant/cluster/namespace/topic");
@@ -117,6 +119,8 @@ public class TopicNameTest {
} catch (IllegalArgumentException e) {
// Ok
}
+ assertThrows(IllegalArgumentException.class,
+ () -> TopicName.toFullTopicName("invalid://tenant/cluster/namespace/topic"));
try {
TopicName.get("tenant/cluster/namespace/topic");
@@ -124,6 +128,7 @@ public class TopicNameTest {
} catch (IllegalArgumentException e) {
// Ok
}
+ assertThrows(IllegalArgumentException.class, () -> TopicName.toFullTopicName("tenant/cluster/namespace/topic"));
try {
TopicName.get("persistent:///cluster/namespace/mydest-1");
@@ -131,6 +136,8 @@ public class TopicNameTest {
} catch (IllegalArgumentException e) {
// Ok
}
+ assertThrows(IllegalArgumentException.class,
+ () -> TopicName.toFullTopicName("persistent:///cluster/namespace/mydest-1"));
try {
TopicName.get("persistent://pulsar//namespace/mydest-1");
@@ -138,6 +145,8 @@ public class TopicNameTest {
} catch (IllegalArgumentException e) {
// Ok
}
+ assertThrows(IllegalArgumentException.class,
+ () -> TopicName.toFullTopicName("persistent://pulsar//namespace/mydest-1"));
try {
TopicName.get("persistent://pulsar/cluster//mydest-1");
@@ -145,6 +154,8 @@ public class TopicNameTest {
} catch (IllegalArgumentException e) {
// Ok
}
+ assertThrows(IllegalArgumentException.class,
+ () -> TopicName.toFullTopicName("persistent://pulsar/cluster//mydest-1"));
try {
TopicName.get("persistent://pulsar/cluster/namespace/");
@@ -152,6 +163,8 @@ public class TopicNameTest {
} catch (IllegalArgumentException e) {
// Ok
}
+ assertThrows(IllegalArgumentException.class,
+ () -> TopicName.toFullTopicName("persistent://pulsar/cluster/namespace/"));
try {
TopicName.get("://pulsar/cluster/namespace/");
@@ -159,6 +172,7 @@ public class TopicNameTest {
} catch (IllegalArgumentException e) {
// Ok
}
+ assertThrows(IllegalArgumentException.class, () -> TopicName.toFullTopicName("://pulsar/cluster/namespace/"));
assertEquals(TopicName.get("persistent://tenant/cluster/namespace/topic")
.getPersistenceNamingEncoding(), "tenant/cluster/namespace/persistent/topic");
@@ -169,6 +183,7 @@ public class TopicNameTest {
} catch (IllegalArgumentException e) {
// Ok
}
+ assertThrows(IllegalArgumentException.class, () -> TopicName.toFullTopicName("://tenant.namespace"));
try {
TopicName.get("://tenant/cluster/namespace");
@@ -176,6 +191,7 @@ public class TopicNameTest {
} catch (IllegalArgumentException e) {
// Ok
}
+ assertThrows(IllegalArgumentException.class, () -> TopicName.toFullTopicName("://tenant//cluster/namespace"));
try {
TopicName.get(" ");
@@ -183,6 +199,7 @@ public class TopicNameTest {
} catch (IllegalArgumentException e) {
// Ok
}
+ assertThrows(IllegalArgumentException.class, () -> TopicName.toFullTopicName(" "));
TopicName nameWithSlash = TopicName.get("persistent://tenant/cluster/namespace/ns-abc/table/1");
assertEquals(nameWithSlash.getEncodedLocalName(), Codec.encode("ns-abc/table/1"));
@@ -344,4 +361,15 @@ public class TopicNameTest {
assertNotEquals(tp2.toString(), tp1.toString());
assertEquals(tp2.toString(), "persistent://tenant1/namespace1/tp1-partition-0-DLQ-partition-0");
}
+
+ @Test
+ public void testToFullTopicName() {
+ // There is no constraint for local topic name
+ assertEquals("persistent://public/default/tp???xx=", TopicName.toFullTopicName("tp???xx="));
+ assertEquals("persistent://tenant/ns/tp???xx=", TopicName.toFullTopicName("tenant/ns/tp???xx="));
+ assertEquals("persistent://tenant/ns/test", TopicName.toFullTopicName("persistent://tenant/ns/test"));
+ assertThrows(IllegalArgumentException.class, () -> TopicName.toFullTopicName("ns/topic"));
+ // v1 format is not supported when the domain is not included
+ assertThrows(IllegalArgumentException.class, () -> TopicName.toFullTopicName("tenant/cluster/ns/topic"));
+ }
}
diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/LookupProxyHandler.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/LookupProxyHandler.java
index 75109883f98..c38e2ba08c1 100644
--- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/LookupProxyHandler.java
+++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/LookupProxyHandler.java
@@ -221,7 +221,7 @@ public class LookupProxyHandler {
**/
private void handlePartitionMetadataResponse(CommandPartitionedTopicMetadata partitionMetadata,
long clientRequestId) {
- TopicName topicName = TopicName.get(partitionMetadata.getTopic());
+ String topicName = TopicName.toFullTopicName(partitionMetadata.getTopic());
String serviceUrl = getBrokerServiceUrl(clientRequestId);
if (serviceUrl == null) {
@@ -235,7 +235,7 @@ public class LookupProxyHandler {
if (log.isDebugEnabled()) {
log.debug("Getting connections to '{}' for Looking up topic '{}' with clientReq Id '{}'", addr,
- topicName.getPartitionedTopicName(), clientRequestId);
+ topicName, clientRequestId);
}
proxyConnection.getConnectionPool().getConnection(addr).thenAccept(clientCnx -> {
// Connected to backend broker
@@ -245,7 +245,7 @@ public class LookupProxyHandler {
partitionMetadata.isMetadataAutoCreationEnabled());
clientCnx.newLookup(command, requestId).whenComplete((r, t) -> {
if (t != null) {
- log.warn("[{}] failed to get Partitioned metadata : {}", topicName.toString(),
+ log.warn("[{}] failed to get Partitioned metadata : {}", topicName,
t.getMessage(), t);
PulsarClientException pce = PulsarClientException.unwrap(t);
writeAndFlush(Commands.newLookupErrorResponse(clientCnx.revertClientExToErrorCode(pce),
diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ParserProxyHandler.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ParserProxyHandler.java
index 22957c9599f..3a98311eb15 100644
--- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ParserProxyHandler.java
+++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ParserProxyHandler.java
@@ -101,7 +101,8 @@ public class ParserProxyHandler extends ChannelInboundHandlerAdapter {
private final BaseCommand cmd = new BaseCommand();
public void channelRead(ChannelHandlerContext ctx, Object msg) {
- TopicName topicName;
+ String key;
+ String topicName;
List<RawMessage> messages = new ArrayList<>();
ByteBuf buffer = (ByteBuf) (msg);
@@ -130,8 +131,8 @@ public class ParserProxyHandler extends ChannelInboundHandlerAdapter {
logging(ctx.channel(), cmd.getType(), "", null);
break;
}
- topicName = TopicName.get(ParserProxyHandler.producerHashMap.get(cmd.getSend().getProducerId() + ","
- + ctx.channel().id()));
+ topicName = TopicName.toFullTopicName(ParserProxyHandler.producerHashMap.get(
+ cmd.getSend().getProducerId() + "," + ctx.channel().id()));
MutableLong msgBytes = new MutableLong(0);
MessageParser.parseMessage(topicName, -1L,
-1L, buffer, (message) -> {
@@ -139,7 +140,7 @@ public class ParserProxyHandler extends ChannelInboundHandlerAdapter {
msgBytes.add(message.getData().readableBytes());
}, maxMessageSize);
// update topic stats
- TopicStats topicStats = this.service.getTopicStats().computeIfAbsent(topicName.toString(),
+ TopicStats topicStats = this.service.getTopicStats().computeIfAbsent(topicName,
topic -> new TopicStats());
topicStats.getMsgInRate().recordMultipleEvents(messages.size(), msgBytes.longValue());
logging(ctx.channel(), cmd.getType(), "", messages);
@@ -158,8 +159,9 @@ public class ParserProxyHandler extends ChannelInboundHandlerAdapter {
logging(ctx.channel(), cmd.getType(), "", null);
break;
}
- topicName = TopicName.get(ParserProxyHandler.consumerHashMap.get(cmd.getMessage().getConsumerId()
- + "," + peerChannelId));
+ topicName = TopicName.toFullTopicName(ParserProxyHandler.consumerHashMap.get(
+ cmd.getMessage().getConsumerId() + "," + peerChannelId));
+
msgBytes = new MutableLong(0);
MessageParser.parseMessage(topicName, -1L,
-1L, buffer, (message) -> {
(pulsar) 04/06: [fix][proxy] Fix default value of connectionMaxIdleSeconds in Pulsar Proxy (#24529)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 128f705878a511284d849c2e6fa742bf2a7d9b0b
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Thu Jul 17 07:24:23 2025 +0300
[fix][proxy] Fix default value of connectionMaxIdleSeconds in Pulsar Proxy (#24529)
(cherry picked from commit a87d47710b5495ebdbcbb07f61798f9af697e552)
---
.../pulsar/proxy/server/ProxyConnection.java | 4 +--
.../org/apache/pulsar/proxy/server/ProxyTest.java | 30 ++++++++++++++++++++++
2 files changed, 32 insertions(+), 2 deletions(-)
diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java
index 824df7ee6bc..77065c956d4 100644
--- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java
+++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java
@@ -727,6 +727,8 @@ public class ProxyConnection extends PulsarHandler {
ProxyConfiguration proxyConfig = service.getConfiguration();
initialConf.setServiceUrl(
proxyConfig.isTlsEnabledWithBroker() ? service.getServiceUrlTls() : service.getServiceUrl());
+ /** The proxy service does not need to automatically clean up idling connections, so set to false. **/
+ initialConf.setConnectionMaxIdleSeconds(-1);
// Apply all arbitrary configuration. This must be called before setting any fields annotated as
// @Secret on the ClientConfigurationData object because of the way they are serialized.
@@ -735,8 +737,6 @@ public class ProxyConnection extends PulsarHandler {
.filterAndMapProperties(proxyConfig.getProperties(), "brokerClient_");
ClientConfigurationData clientConf = ConfigurationDataUtils
.loadData(overrides, initialConf, ClientConfigurationData.class);
- /** The proxy service does not need to automatically clean up invalid connections, so set false. **/
- initialConf.setConnectionMaxIdleSeconds(-1);
clientConf.setAuthentication(this.getClientAuthentication());
if (proxyConfig.isTlsEnabledWithBroker()) {
clientConf.setUseTls(true);
diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyTest.java
index e2bdb0ff9a9..b5e34b2acde 100644
--- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyTest.java
+++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyTest.java
@@ -41,6 +41,8 @@ import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.ToString;
import org.apache.avro.reflect.Nullable;
+import org.apache.commons.lang3.mutable.MutableBoolean;
+import org.apache.commons.lang3.reflect.FieldUtils;
import org.apache.pulsar.PulsarVersion;
import org.apache.pulsar.broker.BrokerTestUtil;
import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
@@ -178,6 +180,34 @@ public class ProxyTest extends MockedPulsarServiceBaseTest {
}
}
+ @Test
+ public void testProxyConnectionClientConfig() throws Exception {
+ @Cleanup
+ PulsarClient client = PulsarClient.builder().serviceUrl(proxyService.getServiceUrl())
+ .build();
+
+ @Cleanup
+ Producer<byte[]> producer = client.newProducer()
+ .topic("persistent://sample/test/local/producer-topic2")
+ .create();
+
+ MutableBoolean found = new MutableBoolean(false);
+ proxyService.getClientCnxs().forEach(proxyConnection -> {
+ if (proxyConnection.getConnectionPool() != null) {
+ try {
+ found.setTrue();
+ assertEquals(-1,
+ FieldUtils.readDeclaredField(proxyConnection.getConnectionPool(),
+ "connectionMaxIdleSeconds",
+ true));
+ } catch (IllegalAccessException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ });
+ assertTrue(found.isTrue(), "No proxy connection found with connectionMaxIdleSeconds set to -1");
+ }
+
@Test
public void testProducerConsumer() throws Exception {
@Cleanup
(pulsar) 03/06: [fix][client] NPE in MultiTopicsConsumerImpl.negativeAcknowledge (#24476)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 2ee969a044b0e9b4ade6383d8255fe222ee6da29
Author: Penghui Li <pe...@apache.org>
AuthorDate: Wed Jul 2 18:58:39 2025 -0700
[fix][client] NPE in MultiTopicsConsumerImpl.negativeAcknowledge (#24476)
(cherry picked from commit 89167308709cf72d3f2731bad1bb30384f80abb4)
---
.../org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java | 10 ++++++++++
1 file changed, 10 insertions(+)
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
index f30dc48dab7..62d4e9c63de 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
@@ -566,6 +566,11 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
@Override
public void negativeAcknowledge(MessageId messageId) {
+ if (getState() != State.Ready) {
+ log.warn("[{}] [{}] Cannot negative acknowledge message {} - consumer is not ready (state: {})",
+ topic, subscription, messageId, getState());
+ return;
+ }
checkArgument(messageId instanceof TopicMessageId);
ConsumerImpl<T> consumer = consumers.get(((TopicMessageId) messageId).getOwnerTopic());
consumer.negativeAcknowledge(messageId);
@@ -574,6 +579,11 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
@Override
public void negativeAcknowledge(Message<?> message) {
+ if (getState() != State.Ready) {
+ log.warn("[{}] [{}] Cannot negative acknowledge message {} - consumer is not ready (state: {})",
+ topic, subscription, message.getMessageId(), getState());
+ return;
+ }
MessageId messageId = message.getMessageId();
checkArgument(messageId instanceof TopicMessageId);
ConsumerImpl<T> consumer = consumers.get(((TopicMessageId) messageId).getOwnerTopic());
(pulsar) 06/06: [improve][misc] Optimize topic list hashing so that potentially large String allocation is avoided (#24525)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit ab559c6b349fda017c387fe073acc5f7a0bfa5d8
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Thu Jul 17 10:18:57 2025 +0300
[improve][misc] Optimize topic list hashing so that potentially large String allocation is avoided (#24525)
(cherry picked from commit 06f424c4823399ae66608ad9fd0e2133a1055b1a)
---
.../org/apache/pulsar/common/topics/TopicList.java | 20 ++++++++++++++------
.../apache/pulsar/common/topics/TopicListTest.java | 19 ++++++++++++-------
2 files changed, 26 insertions(+), 13 deletions(-)
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/topics/TopicList.java b/pulsar-common/src/main/java/org/apache/pulsar/common/topics/TopicList.java
index 7a5659c33b5..1e0bde1b0c5 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/topics/TopicList.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/topics/TopicList.java
@@ -18,8 +18,10 @@
*/
package org.apache.pulsar.common.topics;
+import com.google.common.hash.Hasher;
import com.google.common.hash.Hashing;
import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
@@ -46,6 +48,7 @@ public class TopicList {
Pattern topicsPattern = Pattern.compile(regex);
return filterTopics(original, topicsPattern);
}
+
public static List<String> filterTopics(List<String> original, Pattern topicsPattern) {
final Pattern shortenedTopicsPattern = Pattern.compile(removeTopicDomainScheme(topicsPattern.toString()));
@@ -68,14 +71,19 @@ public class TopicList {
}
public static String calculateHash(List<String> topics) {
- return Hashing.crc32c().hashBytes(topics.stream()
- .sorted()
- .collect(Collectors.joining(","))
- .getBytes(StandardCharsets.UTF_8)).toString();
+ Hasher hasher = Hashing.crc32c().newHasher();
+ String[] sortedTopics = topics.toArray(new String[topics.size()]);
+ Arrays.sort(sortedTopics);
+ for (int i = 0; i < sortedTopics.length; i++) {
+ hasher.putString(sortedTopics[i], StandardCharsets.UTF_8);
+ // Skip the delimiter for the last item so that the hash format is compatible with previous versions
+ if (i < sortedTopics.length - 1) {
+ hasher.putByte((byte) ',');
+ }
+ }
+ return hasher.hash().toString();
}
-
-
// get topics, which are contained in list1, and not in list2
public static Set<String> minus(Collection<String> list1, Collection<String> list2) {
HashSet<String> s1 = new HashSet<>(list1);
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/topics/TopicListTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/topics/TopicListTest.java
index b3a7536ebff..843fe43966a 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/topics/TopicListTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/topics/TopicListTest.java
@@ -18,19 +18,18 @@
*/
package org.apache.pulsar.common.topics;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotEquals;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
import com.google.common.collect.Lists;
-import org.testng.annotations.Test;
-
import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
import java.util.Set;
import java.util.regex.Pattern;
import java.util.stream.Stream;
-
-import static org.testng.Assert.assertEquals;
-import static org.testng.Assert.assertNotEquals;
-import static org.testng.Assert.assertTrue;
-import static org.testng.Assert.fail;
+import org.testng.annotations.Test;
public class TopicListTest {
@@ -102,10 +101,16 @@ public class TopicListTest {
String hash1 = TopicList.calculateHash(Arrays.asList(topicName3, topicName2, topicName1));
String hash2 = TopicList.calculateHash(Arrays.asList(topicName1, topicName3, topicName2));
assertEquals(hash1, hash2, "Hash must not depend on order of topics in the list");
+ assertEquals(hash1, "90d4a04a", "Hash must be equal to the expected value");
String hash3 = TopicList.calculateHash(Arrays.asList(topicName1, topicName2));
assertNotEquals(hash1, hash3, "Different list must have different hashes");
+ String hash4 = TopicList.calculateHash(Arrays.asList(topicName1));
+ assertEquals(hash4, "0d0602ed", "Hash must be equal to the expected value");
+
+ String hash5 = TopicList.calculateHash(Collections.emptyList());
+ assertEquals(hash5, "00000000", "Hash of empty list must be 0");
}
@Test
(pulsar) 05/06: [fix][client] Fix issue in auto releasing of idle connection with topics pattern consumer (#24528)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit c42ba7ffd0991d9f9732178137b5d159de6c7d4e
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Thu Jul 17 10:18:41 2025 +0300
[fix][client] Fix issue in auto releasing of idle connection with topics pattern consumer (#24528)
(cherry picked from commit 8be127fe3834851008061da0de22e591b9c7b130)
---
.../apache/pulsar/broker/MultiBrokerBaseTest.java | 14 ++
.../impl/AutoCloseUselessClientConSupports.java | 2 +-
...eUselessClientConTopicsPatternConsumerTest.java | 107 +++++++++++++++
.../org/apache/pulsar/client/impl/ClientCnx.java | 3 +
.../pulsar/client/impl/ClientCnxIdleState.java | 7 +-
.../pulsar/client/impl/ClientCnxIdleStateTest.java | 61 +++++++++
.../apache/pulsar/client/impl/ClientCnxTest.java | 11 ++
.../impl/AutoCloseUselessClientConProxyTest.java | 151 +++++++++++++++++++++
8 files changed, 354 insertions(+), 2 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/MultiBrokerBaseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/MultiBrokerBaseTest.java
index 6e4d7893adb..2a3e705dca1 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/MultiBrokerBaseTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/MultiBrokerBaseTest.java
@@ -50,6 +50,11 @@ public abstract class MultiBrokerBaseTest extends MockedPulsarServiceBaseTest {
super.internalSetup();
additionalBrokersSetup();
pulsarResourcesSetup();
+ additionalSetup();
+ }
+
+ protected void additionalSetup() throws Exception {
+ // override this method to add any additional setup logic
}
protected void pulsarResourcesSetup() throws PulsarAdminException {
@@ -91,9 +96,18 @@ public abstract class MultiBrokerBaseTest extends MockedPulsarServiceBaseTest {
@Override
public final void cleanup() throws Exception {
additionalBrokersCleanup();
+ try {
+ additionalCleanup();
+ } catch (Exception e) {
+ log.warn("Exception during additional cleanup", e);
+ }
super.internalCleanup();
}
+ protected void additionalCleanup() throws Exception {
+ // override this method to add any additional cleanup logic
+ }
+
protected void additionalBrokersCleanup() {
if (additionalBrokerAdmins != null) {
for (PulsarAdmin additionalBrokerAdmin : additionalBrokerAdmins) {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/AutoCloseUselessClientConSupports.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/AutoCloseUselessClientConSupports.java
index c9f478969a6..662538504cd 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/AutoCloseUselessClientConSupports.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/AutoCloseUselessClientConSupports.java
@@ -39,7 +39,7 @@ import org.apache.pulsar.client.api.transaction.Transaction;
import org.awaitility.Awaitility;
import org.testng.Assert;
-public class AutoCloseUselessClientConSupports extends MultiBrokerBaseTest {
+public abstract class AutoCloseUselessClientConSupports extends MultiBrokerBaseTest {
protected int BROKER_COUNT = 5;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/AutoCloseUselessClientConTopicsPatternConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/AutoCloseUselessClientConTopicsPatternConsumerTest.java
new file mode 100644
index 00000000000..981612d3930
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/AutoCloseUselessClientConTopicsPatternConsumerTest.java
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import lombok.Cleanup;
+import org.apache.pulsar.broker.BrokerTestUtil;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.InjectedClientCnxClientBuilder;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.impl.metrics.InstrumentProvider;
+import org.testng.annotations.Test;
+
+@Test
+public class AutoCloseUselessClientConTopicsPatternConsumerTest extends AutoCloseUselessClientConSupports {
+ private static final String TOPIC_NAME = BrokerTestUtil.newUniqueName("pattern_");
+ private static final String TOPIC_FULL_NAME = "persistent://public/default/" + TOPIC_NAME;
+ private static final String TOPIC_PATTERN = "persistent://public/default/pattern_.*";
+
+ @Override
+ protected void pulsarResourcesSetup() throws PulsarAdminException {
+ super.pulsarResourcesSetup();
+ admin.topics().createNonPartitionedTopic(TOPIC_FULL_NAME);
+ }
+
+ @Test
+ public void testConnectionAutoReleaseWhileUsingTopicsPatternConsumer() throws Exception {
+ ClientBuilderImpl clientBuilder = (ClientBuilderImpl) PulsarClient.builder()
+ .serviceUrl(lookupUrl.toString())
+ .statsInterval(0, TimeUnit.SECONDS)
+ .connectionsPerBroker(Integer.MAX_VALUE);
+ AtomicInteger connectionCreationCounter = new AtomicInteger(0);
+ PulsarClientImpl pulsarClient = InjectedClientCnxClientBuilder.create(clientBuilder, (conf, eventLoopGroup) -> {
+ connectionCreationCounter.incrementAndGet();
+ return new ClientCnx(InstrumentProvider.NOOP, conf, eventLoopGroup);
+ });
+
+ @Cleanup
+ Consumer consumer = pulsarClient.newConsumer(Schema.BYTES)
+ .topicsPattern(TOPIC_PATTERN)
+ .isAckReceiptEnabled(true)
+ .subscriptionName("my-subscription-x")
+ .subscribe();
+ @Cleanup
+ Producer producer = pulsarClient.newProducer(Schema.BYTES)
+ .topic(TOPIC_NAME)
+ .create();
+
+ Consumer consumer2 = pulsarClient.newConsumer(Schema.BYTES)
+ .topicsPattern(TOPIC_PATTERN)
+ .subscriptionName("my-subscription-y")
+ .subscribe();
+
+ // check that there are more than 3 connections
+ // at least 3 connections are required:
+ // 1 for "producer", 1 for "consumer", and 1 for the topic watcher of "consumer"
+ // additional connections will be created for the second consumer and its topic watcher
+ // there's also a connection for topic lookup
+ assertThat(pulsarClient.getCnxPool().getPoolSize()).isGreaterThan(3);
+
+ int connectionsCreatedBefore = connectionCreationCounter.get();
+
+ // trigger releasing of unused client connections
+ trigReleaseConnection(pulsarClient);
+
+ // close consumer2 that creates an extra connection due to connectionsPerBroker set to Integer.MAX_VALUE
+ consumer2.close();
+
+ // trigger releasing of unused client connections
+ trigReleaseConnection(pulsarClient);
+
+ // verify that the number of connections is 3 now, which is the expected number of connections
+ assertThat(pulsarClient.getCnxPool().getPoolSize()).isEqualTo(3);
+
+ // Ensure all things still works
+ ensureProducerAndConsumerWorks(producer, consumer);
+
+ // Verify that the number of connections did not increase after the work was completed
+ assertThat(pulsarClient.getCnxPool().getPoolSize()).isEqualTo(3);
+
+ assertThat(connectionCreationCounter.get())
+ .as("No new connections should be created after releasing unused connections since that would "
+ + "mean that an used connection was released.")
+ .isEqualTo(connectionsCreatedBefore);
+ }
+}
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 8b5964b15fc..4e7a2b42ddf 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
@@ -1508,6 +1508,9 @@ public class ClientCnx extends PulsarHandler {
if (!transactionMetaStoreHandlers.isEmpty()) {
return false;
}
+ if (!topicListWatchers.isEmpty()) {
+ return false;
+ }
return true;
}
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnxIdleState.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnxIdleState.java
index 59d3309f248..a3bff56d298 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnxIdleState.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnxIdleState.java
@@ -31,7 +31,7 @@ public class ClientCnxIdleState {
private final long createTime;
/** The time when marks the connection is idle. **/
- private long idleMarkTime;
+ private volatile long idleMarkTime;
public ClientCnxIdleState(ClientCnx clientCnx){
this.clientCnx = clientCnx;
@@ -171,6 +171,11 @@ public class ClientCnxIdleState {
return;
}
if (isIdle()) {
+ // check if the connection is still idle, if not, mark it as using
+ if (!clientCnx.idleCheck() && compareAndSetIdleStat(State.IDLE, State.USING)) {
+ idleMarkTime = 0;
+ return;
+ }
if (maxIdleSeconds * 1000 + idleMarkTime < System.currentTimeMillis()) {
tryMarkReleasing();
}
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxIdleStateTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxIdleStateTest.java
new file mode 100644
index 00000000000..e3d88a3823d
--- /dev/null
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxIdleStateTest.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl;
+
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.testng.Assert.assertTrue;
+import java.util.concurrent.TimeUnit;
+import org.testng.annotations.Test;
+
+public class ClientCnxIdleStateTest {
+
+ @Test
+ public void testShouldNotReleaseConnectionIfIdleCheckFails() throws InterruptedException {
+ ClientCnx clientCnx = mock(ClientCnx.class);
+ ClientCnxIdleState idleState = new ClientCnxIdleState(clientCnx);
+ int maxIdleSeconds = 1;
+
+ // initially, return true for idle check
+ doReturn(true).when(clientCnx).idleCheck();
+
+ // do the first idle detection
+ idleState.doIdleDetect(maxIdleSeconds);
+
+ // the state should be IDLE since the idle check passed
+ assertTrue(idleState.isIdle());
+
+ // Wait for more than maxIdleSeconds
+ Thread.sleep(TimeUnit.SECONDS.toMillis(maxIdleSeconds) + 1);
+
+ // now return false for idle check
+ doReturn(false).when(clientCnx).idleCheck();
+
+ // do the second idle detection
+ idleState.doIdleDetect(maxIdleSeconds);
+
+ // the state should now be USING since the idle check failed
+ assertTrue(idleState.isUsing());
+
+ // verify that idleCheck was called twice
+ verify(clientCnx, times(2)).idleCheck();
+ }
+}
\ No newline at end of file
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxTest.java
index e7986ec57a2..86644708eba 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxTest.java
@@ -285,6 +285,17 @@ public class ClientCnxTest {
eventLoop.shutdownGracefully();
}
+ @Test
+ public void testIdleCheckWithTopicListWatcher() {
+ ClientCnx cnx =
+ new ClientCnx(InstrumentProvider.NOOP, new ClientConfigurationData(), mock(EventLoopGroup.class));
+ // idle check should return true initially
+ assertTrue(cnx.idleCheck());
+ cnx.registerTopicListWatcher(0, mock(TopicListWatcher.class));
+ // idle check should now return false since there's a registered watcher
+ assertFalse(cnx.idleCheck());
+ }
+
@Test
public void testNoWatchersWhenNoServerSupport() {
withConnection("testNoWatchersWhenNoServerSupport", cnx -> {
diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/client/impl/AutoCloseUselessClientConProxyTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/client/impl/AutoCloseUselessClientConProxyTest.java
new file mode 100644
index 00000000000..cec3392b385
--- /dev/null
+++ b/pulsar-proxy/src/test/java/org/apache/pulsar/client/impl/AutoCloseUselessClientConProxyTest.java
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.doReturn;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import lombok.Cleanup;
+import org.apache.pulsar.broker.BrokerTestUtil;
+import org.apache.pulsar.broker.authentication.AuthenticationService;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.InjectedClientCnxClientBuilder;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.impl.auth.AuthenticationDisabled;
+import org.apache.pulsar.client.impl.metrics.InstrumentProvider;
+import org.apache.pulsar.common.configuration.PulsarConfigurationLoader;
+import org.apache.pulsar.metadata.impl.ZKMetadataStore;
+import org.apache.pulsar.proxy.server.ProxyConfiguration;
+import org.apache.pulsar.proxy.server.ProxyService;
+import org.testng.annotations.Test;
+
+@Test
+public class AutoCloseUselessClientConProxyTest extends AutoCloseUselessClientConSupports {
+ private static final String TOPIC_NAME = BrokerTestUtil.newUniqueName("pattern_");
+ private static final String TOPIC_FULL_NAME = "persistent://public/default/" + TOPIC_NAME;
+ private static final String TOPIC_PATTERN = "persistent://public/default/pattern_.*";
+ private ProxyService proxyService;
+ private ProxyConfiguration proxyConfig;
+ private PulsarClient proxiedClient;
+ private AtomicInteger connectionCreationCounter = new AtomicInteger(0);
+
+ @Override
+ protected void additionalSetup() throws Exception {
+ proxyConfig = new ProxyConfiguration();
+ proxyConfig.setServicePort(Optional.ofNullable(0));
+ proxyConfig.setBrokerProxyAllowedTargetPorts("*");
+ proxyConfig.setMetadataStoreUrl("dummy");
+ proxyConfig.setClusterName(configClusterName);
+ startProxyService();
+ // use the same port for subsequent restarts
+ proxyConfig.setServicePort(proxyService.getListenPort());
+
+ ClientBuilderImpl clientBuilder = (ClientBuilderImpl) PulsarClient.builder()
+ .serviceUrl("pulsar://localhost:" + proxyService.getListenPort().get())
+ .connectionsPerBroker(Integer.MAX_VALUE) // effectively uses a different connection for each usage
+ .statsInterval(0, TimeUnit.SECONDS);
+ proxiedClient = InjectedClientCnxClientBuilder.create(clientBuilder, (conf, eventLoopGroup) -> {
+ connectionCreationCounter.incrementAndGet();
+ return new ClientCnx(InstrumentProvider.NOOP, conf, eventLoopGroup);
+ });
+ registerCloseable(proxiedClient);
+ }
+
+ private void startProxyService() throws Exception {
+ proxyService = BrokerTestUtil.spyWithoutRecordingInvocations(new ProxyService(proxyConfig,
+ new AuthenticationService(PulsarConfigurationLoader.convertFrom(proxyConfig)),
+ AuthenticationDisabled.INSTANCE));
+ doReturn(registerCloseable(new ZKMetadataStore(mockZooKeeper))).when(proxyService).createLocalMetadataStore();
+ doReturn(registerCloseable(new ZKMetadataStore(mockZooKeeperGlobal))).when(proxyService)
+ .createConfigurationMetadataStore();
+ proxyService.start();
+ registerCloseable(proxyService);
+ }
+
+ @Override
+ protected void pulsarResourcesSetup() throws PulsarAdminException {
+ super.pulsarResourcesSetup();
+ admin.topics().createNonPartitionedTopic(TOPIC_FULL_NAME);
+ }
+
+ @Test
+ public void testConnectionAutoReleaseWhileUsingTopicsPatternConsumerAndProxy() throws Exception {
+ PulsarClientImpl pulsarClient = (PulsarClientImpl) proxiedClient;
+ @Cleanup
+ Consumer consumer = pulsarClient.newConsumer(Schema.BYTES)
+ .topicsPattern(TOPIC_PATTERN)
+ .isAckReceiptEnabled(true)
+ .subscriptionName("my-subscription-x")
+ .subscribe();
+ @Cleanup
+ Producer producer = pulsarClient.newProducer(Schema.BYTES)
+ .topic(TOPIC_NAME)
+ .create();
+
+ Consumer consumer2 = pulsarClient.newConsumer(Schema.BYTES)
+ .topicsPattern(TOPIC_PATTERN)
+ .subscriptionName("my-subscription-y")
+ .subscribe();
+
+ int poolSizeAfterCreatingConsumersAndProducer = pulsarClient.getCnxPool().getPoolSize();
+ // check that there are more than 3 connections
+ // at least 3 connections are required:
+ // 1 for "producer", 1 for "consumer", and 1 for the topic watcher of "consumer"
+ // additional connections will be created for the second consumer and its topic watcher
+ // there's also a connection for topic lookup
+ assertThat(poolSizeAfterCreatingConsumersAndProducer).isGreaterThan(3);
+
+ int connectionsCreatedBefore = connectionCreationCounter.get();
+
+ // trigger releasing of unused client connections
+ trigReleaseConnection(pulsarClient);
+
+ // verify that the number of connections is still more than 3, but less than the pool size after creating
+ // consumers and producer
+ // since the lookup connection to the proxy should be closed now
+ assertThat(pulsarClient.getCnxPool().getPoolSize())
+ .isGreaterThan(3)
+ .isLessThan(poolSizeAfterCreatingConsumersAndProducer);
+
+ // close consumer2 that creates an extra connection due to connectionsPerBroker set to Integer.MAX_VALUE
+ consumer2.close();
+
+ // trigger releasing of unused client connections
+ trigReleaseConnection(pulsarClient);
+
+ // verify that the number of connections is 3 now, which is the expected number of connections
+ assertThat(pulsarClient.getCnxPool().getPoolSize()).isEqualTo(3);
+
+ // Ensure all things still works
+ ensureProducerAndConsumerWorks(producer, consumer);
+
+ // Verify that the number of connections did not increase after the work was completed
+ assertThat(pulsarClient.getCnxPool().getPoolSize()).isEqualTo(3);
+
+ assertThat(connectionCreationCounter.get())
+ .as("No new connections should be created after releasing unused connections since that would "
+ + "mean that an used connection was released.")
+ .isEqualTo(connectionsCreatedBefore);
+ }
+}
(pulsar) 01/06: [improve][broker] Improve the log when namespace bundle is not available (#24434)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit a197f46b85d5e9bf119a3dcaeb90f6df1de8f787
Author: Yunze Xu <xy...@163.com>
AuthorDate: Sat Jun 21 20:14:44 2025 +0800
[improve][broker] Improve the log when namespace bundle is not available (#24434)
(cherry picked from commit 19b4a05f888da35a584cf7ff6d594dd9df5e03d1)
---
.../apache/pulsar/broker/namespace/NamespaceService.java | 13 ++++++++-----
.../org/apache/pulsar/broker/service/BrokerService.java | 12 ++++++------
.../service/PersistentDispatcherFailoverConsumerTest.java | 3 ++-
.../broker/service/PersistentTopicConcurrentTest.java | 3 ++-
.../apache/pulsar/broker/service/PersistentTopicTest.java | 3 ++-
.../org/apache/pulsar/broker/service/ServerCnxTest.java | 6 ++++--
6 files changed, 24 insertions(+), 16 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
index 0e55fe24030..a347c05824a 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
@@ -1283,13 +1283,16 @@ public class NamespaceService implements AutoCloseable {
}
public CompletableFuture<Boolean> checkTopicOwnership(TopicName topicName) {
- // TODO: Add unit tests cover it.
+ return getBundleAsync(topicName).thenCompose(bundle -> checkBundleOwnership(topicName, bundle));
+ }
+
+ public CompletableFuture<Boolean> checkBundleOwnership(TopicName topicName, NamespaceBundle bundle) {
if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar)) {
- return getBundleAsync(topicName)
- .thenCompose(bundle -> loadManager.get().checkOwnershipAsync(Optional.of(topicName), bundle));
+ // TODO: Add unit tests cover it.
+ return loadManager.get().checkOwnershipAsync(Optional.of(topicName), bundle);
+ } else {
+ return ownershipCache.checkOwnershipAsync(bundle);
}
- return getBundleAsync(topicName)
- .thenCompose(ownershipCache::checkOwnershipAsync);
}
public CompletableFuture<Void> removeOwnedServiceUnitAsync(NamespaceBundle nsBundle) {
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 0b056c8fb86..17373e8502b 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
@@ -2297,19 +2297,19 @@ public class BrokerService implements Closeable {
public CompletableFuture<Void> checkTopicNsOwnership(final String topic) {
TopicName topicName = TopicName.get(topic);
+ final var namespaceService = pulsar.getNamespaceService();
- return pulsar.getNamespaceService().checkTopicOwnership(topicName)
- .thenCompose(ownedByThisInstance -> {
+ return namespaceService.getBundleAsync(topicName).thenCompose(bundle ->
+ namespaceService.checkBundleOwnership(topicName, bundle).thenCompose(ownedByThisInstance -> {
if (ownedByThisInstance) {
return CompletableFuture.completedFuture(null);
} else {
- String msg = String.format("Namespace bundle for topic (%s) not served by this instance:%s. "
- + "Please redo the lookup. Request is denied: namespace=%s",
- topic, pulsar.getBrokerId(), topicName.getNamespace());
+ String msg = String.format("Namespace bundle (%s) for topic (%s) not served by this instance:"
+ + "%s. Please redo the lookup.", bundle, topic, pulsar.getBrokerId());
log.warn(msg);
return FutureUtil.failedFuture(new ServiceUnitNotReadyException(msg));
}
- });
+ }));
}
public CompletableFuture<Integer> unloadServiceUnit(NamespaceBundle serviceUnit,
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java
index a5babaf26c0..412b12f258c 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java
@@ -161,7 +161,8 @@ public class PersistentDispatcherFailoverConsumerTest {
NamespaceService nsSvc = pulsarTestContext.getPulsarService().getNamespaceService();
doReturn(true).when(nsSvc).isServiceUnitOwned(any(NamespaceBundle.class));
doReturn(true).when(nsSvc).isServiceUnitActive(any(TopicName.class));
- doReturn(CompletableFuture.completedFuture(true)).when(nsSvc).checkTopicOwnership(any(TopicName.class));
+ doReturn(CompletableFuture.completedFuture(mock(NamespaceBundle.class))).when(nsSvc).getBundleAsync(any());
+ doReturn(CompletableFuture.completedFuture(true)).when(nsSvc).checkBundleOwnership(any(), any());
setupMLAsyncCallbackMocks();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicConcurrentTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicConcurrentTest.java
index cbbb8808f3d..2d350297369 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicConcurrentTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicConcurrentTest.java
@@ -105,7 +105,8 @@ public class PersistentTopicConcurrentTest extends MockedBookKeeperTestCase {
doReturn(nsSvc).when(pulsar).getNamespaceService();
doReturn(true).when(nsSvc).isServiceUnitOwned(any(NamespaceBundle.class));
doReturn(true).when(nsSvc).isServiceUnitActive(any(TopicName.class));
- doReturn(CompletableFuture.completedFuture(true)).when(nsSvc).checkTopicOwnership(any(TopicName.class));
+ doReturn(CompletableFuture.completedFuture(mock(NamespaceBundle.class))).when(nsSvc).getBundleAsync(any());
+ doReturn(CompletableFuture.completedFuture(true)).when(nsSvc).checkBundleOwnership(any(), any());
final List<Position> addedEntries = new ArrayList<>();
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 c7ff07bb357..19fd1e72c76 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
@@ -220,7 +220,8 @@ public class PersistentTopicTest extends MockedBookKeeperTestCase {
doReturn(true).when(nsSvc).isServiceUnitOwned(any());
doReturn(true).when(nsSvc).isServiceUnitActive(any());
doReturn(CompletableFuture.completedFuture(true)).when(nsSvc).isServiceUnitActiveAsync(any());
- doReturn(CompletableFuture.completedFuture(true)).when(nsSvc).checkTopicOwnership(any());
+ doReturn(CompletableFuture.completedFuture(mock(NamespaceBundle.class))).when(nsSvc).getBundleAsync(any());
+ doReturn(CompletableFuture.completedFuture(true)).when(nsSvc).checkBundleOwnership(any(), any());
setupMLAsyncCallbackMocks();
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java
index 6f127877e8a..a012748e8b8 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java
@@ -142,6 +142,7 @@ import org.apache.pulsar.common.api.proto.ProtocolVersion;
import org.apache.pulsar.common.api.proto.ServerError;
import org.apache.pulsar.common.api.proto.Subscription;
import org.apache.pulsar.common.api.proto.TxnAction;
+import org.apache.pulsar.common.naming.NamespaceBundle;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.AuthAction;
@@ -223,11 +224,12 @@ public class ServerCnxTest {
brokerService = pulsarTestContext.getBrokerService();
namespaceService = pulsar.getNamespaceService();
- doReturn(CompletableFuture.completedFuture(null)).when(namespaceService).getBundleAsync(any());
+ doReturn(CompletableFuture.completedFuture(mock(NamespaceBundle.class))).when(namespaceService)
+ .getBundleAsync(any());
+ doReturn(CompletableFuture.completedFuture(true)).when(namespaceService).checkBundleOwnership(any(), any());
doReturn(true).when(namespaceService).isServiceUnitOwned(any());
doReturn(true).when(namespaceService).isServiceUnitActive(any());
doReturn(CompletableFuture.completedFuture(true)).when(namespaceService).isServiceUnitActiveAsync(any());
- doReturn(CompletableFuture.completedFuture(true)).when(namespaceService).checkTopicOwnership(any());
doReturn(CompletableFuture.completedFuture(topics)).when(namespaceService).getListOfTopics(
NamespaceName.get("use", "ns-abc"), CommandGetTopicsOfNamespace.Mode.ALL);
doReturn(CompletableFuture.completedFuture(topics)).when(namespaceService).getListOfUserTopics(