Skip to content

Commit bb11a99

Browse files
lhotarilhotari
committed
[fix][client] Prevent retry topic and dead letter topic producer leaks when sending of message fails (#23824)
(cherry picked from commit 04e89fe)
1 parent af97eaa commit bb11a99

File tree

5 files changed

+423
-164
lines changed

5 files changed

+423
-164
lines changed

Diff for: ‎pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeadLetterTopicTest.java

+93
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
*/
1919
package org.apache.pulsar.client.api;
2020

21+
import static org.assertj.core.api.Assertions.assertThat;
2122
import static org.testng.Assert.assertEquals;
2223
import static org.testng.Assert.assertNotNull;
2324
import static org.testng.Assert.assertNull;
@@ -40,9 +41,11 @@
4041
import lombok.Cleanup;
4142
import lombok.Data;
4243
import org.apache.avro.reflect.Nullable;
44+
import org.apache.pulsar.broker.BrokerTestUtil;
4345
import org.apache.pulsar.client.api.schema.GenericRecord;
4446
import org.apache.pulsar.client.impl.ConsumerBuilderImpl;
4547
import org.apache.pulsar.client.util.RetryMessageUtil;
48+
import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
4649
import org.awaitility.Awaitility;
4750
import org.slf4j.Logger;
4851
import org.slf4j.LoggerFactory;
@@ -1167,4 +1170,94 @@ public void testDeadLetterPolicyDeserialize() throws Exception {
11671170
consumerBuilder.loadConf(config);
11681171
assertEquals(((ConsumerBuilderImpl)consumerBuilder).getConf().getDeadLetterPolicy(), policy);
11691172
}
1173+
1174+
@Data
1175+
static class Payload {
1176+
String number;
1177+
1178+
public Payload() {
1179+
1180+
}
1181+
1182+
public Payload(String number) {
1183+
this.number = number;
1184+
}
1185+
}
1186+
1187+
@Data
1188+
static class PayloadIncompatible {
1189+
long number;
1190+
1191+
public PayloadIncompatible() {
1192+
1193+
}
1194+
1195+
public PayloadIncompatible(long number) {
1196+
this.number = number;
1197+
}
1198+
}
1199+
1200+
// reproduce issue reported in https://github.com/apache/pulsar/issues/20635#issuecomment-1709616321
1201+
@Test
1202+
public void testCloseDeadLetterTopicProducerOnExceptionToPreventProducerLeak() throws Exception {
1203+
String namespace = BrokerTestUtil.newUniqueName("my-property/my-ns");
1204+
admin.namespaces().createNamespace(namespace);
1205+
// don't enforce schema validation
1206+
admin.namespaces().setSchemaValidationEnforced(namespace, false);
1207+
// set schema compatibility strategy to always compatible
1208+
admin.namespaces().setSchemaCompatibilityStrategy(namespace, SchemaCompatibilityStrategy.ALWAYS_COMPATIBLE);
1209+
1210+
Schema<Payload> schema = Schema.AVRO(Payload.class);
1211+
Schema<PayloadIncompatible> schemaIncompatible = Schema.AVRO(PayloadIncompatible.class);
1212+
String topic = BrokerTestUtil.newUniqueName("persistent://" + namespace
1213+
+ "/testCloseDeadLetterTopicProducerOnExceptionToPreventProducerLeak");
1214+
String dlqTopic = topic + "-DLQ";
1215+
1216+
// create topics
1217+
admin.topics().createNonPartitionedTopic(topic);
1218+
admin.topics().createNonPartitionedTopic(dlqTopic);
1219+
1220+
AtomicInteger nackCounter = new AtomicInteger(0);
1221+
Consumer<Payload> payloadConsumer = null;
1222+
try {
1223+
payloadConsumer = pulsarClient.newConsumer(schema).topic(topic)
1224+
.subscriptionType(SubscriptionType.Shared).subscriptionName("sub")
1225+
.ackTimeout(1, TimeUnit.SECONDS)
1226+
.negativeAckRedeliveryDelay(1, TimeUnit.MILLISECONDS)
1227+
.deadLetterPolicy(DeadLetterPolicy.builder().maxRedeliverCount(3).deadLetterTopic(dlqTopic).build())
1228+
.messageListener((c, msg) -> {
1229+
if (nackCounter.incrementAndGet() < 10) {
1230+
c.negativeAcknowledge(msg);
1231+
}
1232+
}).subscribe();
1233+
1234+
// send a message to the topic with the incompatible schema
1235+
PayloadIncompatible payloadIncompatible = new PayloadIncompatible(123);
1236+
try (Producer<PayloadIncompatible> producer = pulsarClient.newProducer(schemaIncompatible).topic(topic)
1237+
.create()) {
1238+
producer.send(payloadIncompatible);
1239+
}
1240+
1241+
Thread.sleep(2000L);
1242+
1243+
assertThat(pulsar.getBrokerService().getTopicReference(dlqTopic).get().getProducers().size())
1244+
.describedAs("producer count of dlq topic %s should be <= 1 so that it doesn't leak producers",
1245+
dlqTopic)
1246+
.isLessThanOrEqualTo(1);
1247+
1248+
} finally {
1249+
if (payloadConsumer != null) {
1250+
try {
1251+
payloadConsumer.close();
1252+
} catch (PulsarClientException e) {
1253+
// ignore
1254+
}
1255+
}
1256+
}
1257+
1258+
assertThat(pulsar.getBrokerService().getTopicReference(dlqTopic).get().getProducers().size())
1259+
.describedAs("producer count of dlq topic %s should be 0 here",
1260+
dlqTopic)
1261+
.isEqualTo(0);
1262+
}
11701263
}

Diff for: ‎pulsar-broker/src/test/java/org/apache/pulsar/client/api/RetryTopicTest.java

+120-46
Original file line numberDiff line numberDiff line change
@@ -18,12 +18,12 @@
1818
*/
1919
package org.apache.pulsar.client.api;
2020

21+
import static org.assertj.core.api.Assertions.assertThat;
2122
import static org.testng.Assert.assertEquals;
2223
import static org.testng.Assert.assertFalse;
2324
import static org.testng.Assert.assertNull;
2425
import static org.testng.Assert.assertTrue;
2526
import static org.testng.Assert.fail;
26-
import java.lang.reflect.Field;
2727
import java.util.HashMap;
2828
import java.util.HashSet;
2929
import java.util.List;
@@ -36,11 +36,10 @@
3636
import lombok.Data;
3737
import org.apache.avro.AvroRuntimeException;
3838
import org.apache.avro.reflect.Nullable;
39+
import org.apache.pulsar.broker.BrokerTestUtil;
3940
import org.apache.pulsar.client.api.schema.GenericRecord;
40-
import org.apache.pulsar.client.impl.ConsumerImpl;
41-
import org.apache.pulsar.client.impl.MultiTopicsConsumerImpl;
4241
import org.apache.pulsar.client.util.RetryMessageUtil;
43-
import org.reflections.ReflectionUtils;
42+
import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
4443
import org.slf4j.Logger;
4544
import org.slf4j.LoggerFactory;
4645
import org.testng.annotations.AfterMethod;
@@ -617,10 +616,12 @@ public void testRetryTopicByCustomTopicName() throws Exception {
617616

618617
@Test(timeOut = 30000L)
619618
public void testRetryTopicException() throws Exception {
620-
final String topic = "persistent://my-property/my-ns/retry-topic";
619+
String retryLetterTopic = BrokerTestUtil.newUniqueName("persistent://my-property/my-ns/retry-topic");
620+
final String topic = BrokerTestUtil.newUniqueName("persistent://my-property/my-ns/input-topic");
621621
final int maxRedeliveryCount = 2;
622622
final int sendMessages = 1;
623623
// subscribe before publish
624+
@Cleanup
624625
Consumer<byte[]> consumer = pulsarClient.newConsumer(Schema.BYTES)
625626
.topic(topic)
626627
.subscriptionName("my-subscription")
@@ -629,7 +630,7 @@ public void testRetryTopicException() throws Exception {
629630
.receiverQueueSize(100)
630631
.deadLetterPolicy(DeadLetterPolicy.builder()
631632
.maxRedeliverCount(maxRedeliveryCount)
632-
.retryLetterTopic("persistent://my-property/my-ns/my-subscription-custom-Retry")
633+
.retryLetterTopic(retryLetterTopic)
633634
.build())
634635
.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
635636
.subscribe();
@@ -642,30 +643,16 @@ public void testRetryTopicException() throws Exception {
642643
}
643644
producer.close();
644645

645-
// mock a retry producer exception when reconsumelater is called
646-
MultiTopicsConsumerImpl<byte[]> multiTopicsConsumer = (MultiTopicsConsumerImpl<byte[]>) consumer;
647-
List<ConsumerImpl<byte[]>> consumers = multiTopicsConsumer.getConsumers();
648-
for (ConsumerImpl<byte[]> c : consumers) {
649-
Set<Field> deadLetterPolicyField =
650-
ReflectionUtils.getAllFields(c.getClass(), ReflectionUtils.withName("deadLetterPolicy"));
651-
652-
if (deadLetterPolicyField.size() != 0) {
653-
Field field = deadLetterPolicyField.iterator().next();
654-
field.setAccessible(true);
655-
DeadLetterPolicy deadLetterPolicy = (DeadLetterPolicy) field.get(c);
656-
deadLetterPolicy.setRetryLetterTopic("#persistent://invlaid-topic#");
657-
}
658-
}
646+
admin.topics().terminateTopic(retryLetterTopic);
647+
659648
Message<byte[]> message = consumer.receive();
660649
log.info("consumer received message : {} {}", message.getMessageId(), new String(message.getData()));
661650
try {
662651
consumer.reconsumeLater(message, 1, TimeUnit.SECONDS);
663-
} catch (PulsarClientException.InvalidTopicNameException e) {
664-
assertEquals(e.getClass(), PulsarClientException.InvalidTopicNameException.class);
665-
} catch (Exception e) {
666-
fail("exception should be PulsarClientException.InvalidTopicNameException");
652+
fail("exception should be PulsarClientException.TopicTerminatedException");
653+
} catch (PulsarClientException.TopicTerminatedException e) {
654+
// ok
667655
}
668-
consumer.close();
669656
}
670657

671658

@@ -718,10 +705,12 @@ public void testRetryProducerWillCloseByConsumer() throws Exception {
718705

719706
@Test(timeOut = 30000L)
720707
public void testRetryTopicExceptionWithConcurrent() throws Exception {
721-
final String topic = "persistent://my-property/my-ns/retry-topic";
708+
String retryLetterTopic = BrokerTestUtil.newUniqueName("persistent://my-property/my-ns/retry-topic");
709+
final String topic = BrokerTestUtil.newUniqueName("persistent://my-property/my-ns/input-topic");
722710
final int maxRedeliveryCount = 2;
723711
final int sendMessages = 10;
724712
// subscribe before publish
713+
@Cleanup
725714
Consumer<byte[]> consumer = pulsarClient.newConsumer(Schema.BYTES)
726715
.topic(topic)
727716
.subscriptionName("my-subscription")
@@ -730,7 +719,7 @@ public void testRetryTopicExceptionWithConcurrent() throws Exception {
730719
.receiverQueueSize(100)
731720
.deadLetterPolicy(DeadLetterPolicy.builder()
732721
.maxRedeliverCount(maxRedeliveryCount)
733-
.retryLetterTopic("persistent://my-property/my-ns/my-subscription-custom-Retry")
722+
.retryLetterTopic(retryLetterTopic)
734723
.build())
735724
.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
736725
.subscribe();
@@ -739,24 +728,11 @@ public void testRetryTopicExceptionWithConcurrent() throws Exception {
739728
.topic(topic)
740729
.create();
741730
for (int i = 0; i < sendMessages; i++) {
742-
producer.newMessage().key("1").value(String.format("Hello Pulsar [%d]", i).getBytes()).send();
731+
producer.send(String.format("Hello Pulsar [%d]", i).getBytes());
743732
}
744733
producer.close();
745734

746-
// mock a retry producer exception when reconsumelater is called
747-
MultiTopicsConsumerImpl<byte[]> multiTopicsConsumer = (MultiTopicsConsumerImpl<byte[]>) consumer;
748-
List<ConsumerImpl<byte[]>> consumers = multiTopicsConsumer.getConsumers();
749-
for (ConsumerImpl<byte[]> c : consumers) {
750-
Set<Field> deadLetterPolicyField =
751-
ReflectionUtils.getAllFields(c.getClass(), ReflectionUtils.withName("deadLetterPolicy"));
752-
753-
if (deadLetterPolicyField.size() != 0) {
754-
Field field = deadLetterPolicyField.iterator().next();
755-
field.setAccessible(true);
756-
DeadLetterPolicy deadLetterPolicy = (DeadLetterPolicy) field.get(c);
757-
deadLetterPolicy.setRetryLetterTopic("#persistent://invalid-topic#");
758-
}
759-
}
735+
admin.topics().terminateTopic(retryLetterTopic);
760736

761737
List<Message<byte[]>> messages = Lists.newArrayList();
762738
for (int i = 0; i < sendMessages; i++) {
@@ -769,16 +745,114 @@ public void testRetryTopicExceptionWithConcurrent() throws Exception {
769745
new Thread(() -> {
770746
try {
771747
consumer.reconsumeLater(message, 1, TimeUnit.SECONDS);
772-
} catch (Exception ignore) {
773-
774-
} finally {
748+
} catch (PulsarClientException.TopicTerminatedException e) {
749+
// ok
775750
latch.countDown();
751+
} catch (PulsarClientException e) {
752+
// unexpected exception
753+
fail("unexpected exception", e);
776754
}
777755
}).start();
778756
}
779757

780-
latch.await();
758+
latch.await(sendMessages, TimeUnit.SECONDS);
781759
consumer.close();
782760
}
783761

762+
@Data
763+
static class Payload {
764+
String number;
765+
766+
public Payload() {
767+
768+
}
769+
770+
public Payload(String number) {
771+
this.number = number;
772+
}
773+
}
774+
775+
@Data
776+
static class PayloadIncompatible {
777+
long number;
778+
779+
public PayloadIncompatible() {
780+
781+
}
782+
783+
public PayloadIncompatible(long number) {
784+
this.number = number;
785+
}
786+
}
787+
788+
// reproduce similar issue as reported in https://github.com/apache/pulsar/issues/20635#issuecomment-1709616321
789+
// but for retry topic
790+
@Test
791+
public void testCloseRetryLetterTopicProducerOnExceptionToPreventProducerLeak() throws Exception {
792+
String namespace = BrokerTestUtil.newUniqueName("my-property/my-ns");
793+
admin.namespaces().createNamespace(namespace);
794+
// don't enforce schema validation
795+
admin.namespaces().setSchemaValidationEnforced(namespace, false);
796+
// set schema compatibility strategy to always compatible
797+
admin.namespaces().setSchemaCompatibilityStrategy(namespace, SchemaCompatibilityStrategy.ALWAYS_COMPATIBLE);
798+
799+
Schema<Payload> schema = Schema.AVRO(Payload.class);
800+
Schema<PayloadIncompatible> schemaIncompatible = Schema.AVRO(
801+
PayloadIncompatible.class);
802+
String topic = BrokerTestUtil.newUniqueName("persistent://" + namespace
803+
+ "/testCloseDeadLetterTopicProducerOnExceptionToPreventProducerLeak");
804+
String dlqTopic = topic + "-DLQ";
805+
String retryTopic = topic + "-RETRY";
806+
807+
// create topics
808+
admin.topics().createNonPartitionedTopic(topic);
809+
admin.topics().createNonPartitionedTopic(dlqTopic);
810+
admin.topics().createNonPartitionedTopic(retryTopic);
811+
812+
Consumer<Payload> payloadConsumer = null;
813+
try {
814+
payloadConsumer = pulsarClient.newConsumer(schema).topic(topic)
815+
.subscriptionType(SubscriptionType.Shared).subscriptionName("sub")
816+
.ackTimeout(1, TimeUnit.SECONDS)
817+
.negativeAckRedeliveryDelay(1, TimeUnit.MILLISECONDS)
818+
.enableRetry(true)
819+
.deadLetterPolicy(DeadLetterPolicy.builder().retryLetterTopic(retryTopic).maxRedeliverCount(3)
820+
.deadLetterTopic(dlqTopic).build())
821+
.messageListener((c, msg) -> {
822+
try {
823+
c.reconsumeLater(msg, 1, TimeUnit.MILLISECONDS);
824+
} catch (PulsarClientException e) {
825+
throw new RuntimeException(e);
826+
}
827+
}).subscribe();
828+
829+
// send a message to the topic with the incompatible schema
830+
PayloadIncompatible payloadIncompatible = new PayloadIncompatible(123);
831+
try (Producer<PayloadIncompatible> producer = pulsarClient.newProducer(schemaIncompatible).topic(topic)
832+
.create()) {
833+
producer.send(payloadIncompatible);
834+
}
835+
836+
Thread.sleep(2000L);
837+
838+
assertThat(pulsar.getBrokerService().getTopicReference(retryTopic).get().getProducers().size())
839+
.describedAs("producer count of retry topic %s should be <= 1 so that it doesn't leak producers",
840+
retryTopic)
841+
.isLessThanOrEqualTo(1);
842+
843+
} finally {
844+
if (payloadConsumer != null) {
845+
try {
846+
payloadConsumer.close();
847+
} catch (PulsarClientException e) {
848+
// ignore
849+
}
850+
}
851+
}
852+
853+
assertThat(pulsar.getBrokerService().getTopicReference(retryTopic).get().getProducers().size())
854+
.describedAs("producer count of retry topic %s should be 0 here",
855+
retryTopic)
856+
.isEqualTo(0);
857+
}
784858
}

0 commit comments

Comments
 (0)