This is an automated email from the ASF dual-hosted git repository.
poorbarcode pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 9deda7a1ed8 [fix][client]Broker-side producer handle leak if closes a
producer which state is regitering schema (#25725)
9deda7a1ed8 is described below
commit 9deda7a1ed85c83013944c9f9b8ed24bbbb4d371
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)
---
.../broker/service/OneWayReplicatorTest.java | 4 ++
.../SchemaCompatibilityCheckTest.java | 68 ++++++++++++++++++++++
.../apache/pulsar/client/impl/ProducerImpl.java | 2 +-
3 files changed, 73 insertions(+), 1 deletion(-)
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java
index 06c4411aae0..ddfef968a3b 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java
@@ -40,6 +40,8 @@ import io.netty.channel.Channel;
import io.netty.util.concurrent.FastThreadLocalThread;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
@@ -97,9 +99,11 @@ import
org.apache.pulsar.client.api.SubscriptionInitialPosition;
import org.apache.pulsar.client.api.schema.GenericRecord;
import org.apache.pulsar.client.impl.ClientBuilderImpl;
import org.apache.pulsar.client.impl.ClientCnx;
+import org.apache.pulsar.client.impl.MessageImpl;
import org.apache.pulsar.client.impl.ProducerBuilderImpl;
import org.apache.pulsar.client.impl.ProducerImpl;
import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.TypedMessageBuilderImpl;
import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
import org.apache.pulsar.client.impl.metrics.InstrumentProvider;
import org.apache.pulsar.common.api.proto.CommandSendReceipt;
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 b59ae89c0bf..6d95ae74582 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
@@ -29,15 +29,20 @@ import com.google.common.collect.Sets;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
+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;
@@ -45,7 +50,11 @@ import org.apache.pulsar.client.api.PulsarClientException;
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;
@@ -57,6 +66,7 @@ import org.apache.pulsar.common.schema.SchemaInfo;
import org.apache.pulsar.common.schema.SchemaType;
import org.apache.pulsar.schema.MockExternalJsonSchema;
import org.apache.pulsar.schema.Schemas;
+import org.awaitility.Awaitility;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
@@ -637,6 +647,64 @@ 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);
+ }
+
@Test
public void testExternalSchemaTypeCompatibility() throws Exception {
String namespace = "test-namespace-" + randomName(16);
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 4119fc8b6ed..4ec3c46c858 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
@@ -1240,7 +1240,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);
