This is an automated email from the ASF dual-hosted git repository.
poorbarcode pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new 49b83d8d7e8 [fix][client]Broker-side producer handle leak if closes a
producer which state is regitering schema (#25725)
49b83d8d7e8 is described below
commit 49b83d8d7e8b9720ca3891d66871cd2a1cd74e3c
Author: fengyubiao <[email protected]>
AuthorDate: Sat May 9 14:32:09 2026 +0800
[fix][client]Broker-side producer handle leak if closes a producer which
state is regitering schema (#25725)
(cherry picked from commit 4f0766708ad4d6ce027938414d6fc8e4d764343f)
---
.../SchemaCompatibilityCheckTest.java | 70 ++++++++++++++++++++++
.../apache/pulsar/client/impl/ProducerImpl.java | 2 +-
2 files changed, 71 insertions(+), 1 deletion(-)
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/schema/compatibility/SchemaCompatibilityCheckTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/schema/compatibility/SchemaCompatibilityCheckTest.java
index b18e2c179b3..94d27f81bec 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/schema/compatibility/SchemaCompatibilityCheckTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/schema/compatibility/SchemaCompatibilityCheckTest.java
@@ -26,21 +26,31 @@ import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
import com.google.common.collect.Sets;
import java.util.Collections;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadLocalRandom;
import java.util.stream.Collectors;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.broker.BrokerTestUtil;
import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.ConsumerBuilder;
+import org.apache.pulsar.client.api.InjectedClientCnxClientBuilder;
import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.ProducerBuilder;
+import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SchemaSerializationException;
import org.apache.pulsar.client.api.schema.SchemaDefinition;
+import org.apache.pulsar.client.impl.ClientBuilderImpl;
+import org.apache.pulsar.client.impl.ClientCnx;
+import org.apache.pulsar.client.impl.metrics.InstrumentProvider;
import org.apache.pulsar.client.impl.schema.SchemaInfoImpl;
+import org.apache.pulsar.common.api.proto.CommandGetOrCreateSchemaResponse;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicDomain;
import org.apache.pulsar.common.naming.TopicName;
@@ -51,6 +61,7 @@ import org.apache.pulsar.common.policies.data.TenantInfo;
import org.apache.pulsar.common.schema.SchemaInfo;
import org.apache.pulsar.common.schema.SchemaType;
import org.apache.pulsar.schema.Schemas;
+import org.awaitility.Awaitility;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
@@ -630,6 +641,65 @@ public class SchemaCompatibilityCheckTest extends
MockedPulsarServiceBaseTest {
}
}
+
+ @Test
+ public void testCloseProducerWhenRegisteringNewSchema() throws Exception {
+ final String ns = BrokerTestUtil.newUniqueName(PUBLIC_TENANT + "/ns");
+ final String topic = "persistent://" + BrokerTestUtil.newUniqueName(ns
+ "/tp");
+ admin.namespaces().createNamespace(ns);
+ admin.namespaces().setSchemaCompatibilityStrategy(ns,
SchemaCompatibilityStrategy.ALWAYS_INCOMPATIBLE);
+ Awaitility.await().untilAsserted(() -> {
+ assertEquals(admin.namespaces().getSchemaCompatibilityStrategy(ns),
+ SchemaCompatibilityStrategy.ALWAYS_INCOMPATIBLE);
+ });
+
+ // Injection: Let the handling response of registering schema delay,
then we have enough time to close producer
+ // when it's state is registering schema.
+ CountDownLatch handleErrorSignal = new CountDownLatch(1);
+ ClientBuilderImpl clientBuilder = (ClientBuilderImpl)
PulsarClient.builder().serviceUrl(lookupUrl.toString());
+ PulsarClient injectedReplClient =
InjectedClientCnxClientBuilder.create(clientBuilder,
+ (conf, eventLoopGroup) -> {
+ return new ClientCnx(InstrumentProvider.NOOP, conf,
eventLoopGroup) {
+
+ @Override
+ protected void
handleGetOrCreateSchemaResponse(CommandGetOrCreateSchemaResponse response) {
+ if (response.hasErrorCode()) {
+ try {
+ handleErrorSignal.await();
+ } catch (InterruptedException e) {
+ // Nothing to do.
+ }
+ }
+ super.handleGetOrCreateSchemaResponse(response);
+ }
+ };
+ });
+
+ Producer<byte[]> producer =
injectedReplClient.newProducer(Schema.AUTO_PRODUCE_BYTES()).topic(topic).create();
+ // Registers a consumer to avoid client to close idle connections.
+ Consumer consumer =
injectedReplClient.newConsumer(Schema.AUTO_CONSUME()).subscriptionName("s1")
+ .topic(topic).subscribe();
+ PersistentTopic persistentTopic =
+ (PersistentTopic) pulsar.getBrokerService().getTopic(topic,
false).join().get();
+ assertEquals(persistentTopic.getProducers().size(), 1);
+ producer.newMessage(Schema.AVRO(Schemas.PersonOne.class)).value(new
Schemas.PersonOne(1)).send();
+ CompletableFuture<MessageId> send2 =
producer.newMessage(Schema.AVRO(Schemas.PersonTwo.class))
+ .value(new Schemas.PersonTwo(2, "2")).sendAsync();
+ producer.close();
+ Awaitility.await().untilAsserted(() -> {
+ assertTrue(send2.isDone());
+ assertTrue(send2.isCompletedExceptionally());
+ // Since the producer was closed, the topic should maintain 0
producers.
+ assertEquals(persistentTopic.getProducers().size(), 0);
+ });
+ handleErrorSignal.countDown();
+
+ // cleanup.
+ consumer.close();
+ injectedReplClient.close();
+ admin.topics().unload(topic);
+ }
+
public static String randomName(int numChars) {
StringBuilder sb = new StringBuilder();
for (int i = 0; i < numChars; i++) {
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
index e2a32c5bd7c..c8f6540f23f 100644
---
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
@@ -1220,7 +1220,7 @@ public class ProducerImpl<T> extends ProducerBase<T>
implements TimerTask, Conne
closeProducerTasks();
ClientCnx cnx = cnx();
- if (cnx == null || currentState != State.Ready) {
+ if (cnx == null || (currentState != State.Ready && currentState !=
State.RegisteringSchema)) {
log.info("[{}] [{}] Closed Producer (not connected)", topic,
producerName);
closeAndClearPendingMessages();
return CompletableFuture.completedFuture(null);
