From 16e0f7440b18a917ac207e8e96cfc824fa0caceb Mon Sep 17 00:00:00 2001 From: Heesung Sohn Date: Mon, 19 Sep 2022 21:17:02 -0700 Subject: [PATCH 1/3] [fix][tableview] fixed ack failure in ReaderImpl due to null messageId --- .../pulsar/client/impl/TableViewTest.java | 60 +++++++++++++++++++ .../apache/pulsar/client/impl/ReaderImpl.java | 21 ++++--- 2 files changed, 70 insertions(+), 11 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TableViewTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TableViewTest.java index 20f510e97e2e6..8722f649212ac 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TableViewTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TableViewTest.java @@ -18,6 +18,10 @@ */ package org.apache.pulsar.client.impl; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; import static org.testng.Assert.assertEquals; import static org.testng.Assert.fail; @@ -29,7 +33,9 @@ import java.util.concurrent.TimeUnit; import lombok.Cleanup; import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.reflect.FieldUtils; import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest; +import org.apache.pulsar.client.api.MessageId; import org.apache.pulsar.client.api.MessageRoutingMode; import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.ProducerBuilder; @@ -42,6 +48,7 @@ import org.testng.Assert; import org.testng.annotations.AfterClass; import org.testng.annotations.BeforeClass; +import org.testng.annotations.DataProvider; import org.testng.annotations.Test; /** @@ -217,4 +224,57 @@ public void testPublishNullValue() throws Exception { assertEquals(tv1.size(), 1); assertEquals(tv.get("key2"), "value2"); } + + @DataProvider(name = "partitionedTopic") + public static Object[][] partitioned() { + return new Object[][] {{true}, {false}}; + } + + @Test(timeOut = 30 * 1000, dataProvider = "partitionedTopic") + public void testAck(boolean partitionedTopic) throws Exception { + String topic = null; + if (partitionedTopic) { + topic = "persistent://public/default/tableview-ack-test"; + admin.topics().createPartitionedTopic(topic, 3); + } else { + topic = "persistent://public/default/tableview-no-partition-ack-test"; + admin.topics().createNonPartitionedTopic(topic); + } + + @Cleanup + TableView tv1 = pulsarClient.newTableViewBuilder(Schema.STRING) + .topic(topic) + .autoUpdatePartitionsInterval(5, TimeUnit.SECONDS) + .create(); + + ConsumerBase consumerBase; + if (partitionedTopic) { + MultiTopicsReaderImpl reader = + ((CompletableFuture>) FieldUtils + .readDeclaredField(tv1, "reader", true)).get(); + consumerBase = spy(reader.getMultiTopicsConsumer()); + FieldUtils.writeDeclaredField(reader, "multiTopicsConsumer", consumerBase, true); + } else { + ReaderImpl reader = ((CompletableFuture>) FieldUtils + .readDeclaredField(tv1, "reader", true)).get(); + consumerBase = spy(reader.getConsumer()); + FieldUtils.writeDeclaredField(reader, "consumer", consumerBase, true); + } + + @Cleanup + Producer producer = pulsarClient.newProducer(Schema.STRING).topic(topic).create(); + + int msgCount = 20; + for (int i = 0; i < msgCount; i++) { + producer.newMessage().key("key:" + i).value("value" + i).send(); + } + + Awaitility.await() + .pollInterval(1, TimeUnit.SECONDS) + .atMost(Duration.ofMillis(5000)) + .untilAsserted(() + -> verify(consumerBase, times(msgCount)).acknowledgeCumulativeAsync(any(MessageId.class))); + + + } } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java index c4b5263736e1c..61f32654eda11 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java @@ -177,17 +177,16 @@ public Message readNext(int timeout, TimeUnit unit) throws PulsarClientExcept @Override public CompletableFuture> readNextAsync() { - CompletableFuture> receiveFuture = consumer.receiveAsync(); - receiveFuture.whenComplete((msg, t) -> { - if (msg != null) { - consumer.acknowledgeCumulativeAsync(msg).exceptionally(ex -> { - log.warn("[{}][{}] acknowledge message {} cumulative fail.", getTopic(), - getConsumer().getSubscription(), msg.getMessageId(), ex); - return null; - }); - } - }); - return receiveFuture; + return consumer.receiveAsync() + .whenComplete((msg, t) -> { + if (msg != null) { + consumer.acknowledgeCumulativeAsync(msg).exceptionally(ex -> { + log.error("[{}][{}] acknowledge message {} cumulative fail.", getTopic(), + getConsumer().getSubscription(), msg.getMessageId(), ex); + return null; + }); + } + }); } @Override From 2267cf7e5b51a832c23af868ae17a4d0d7396eda Mon Sep 17 00:00:00 2001 From: Heesung Sohn Date: Tue, 20 Sep 2022 09:05:35 -0700 Subject: [PATCH 2/3] Fixed shouldSupportCancellingReadNextAsync test --- .../apache/pulsar/client/impl/ReaderImpl.java | 19 +++++++++---------- .../pulsar/client/impl/ReaderImplTest.java | 13 +++++++++++-- 2 files changed, 20 insertions(+), 12 deletions(-) diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java index 61f32654eda11..da58db65c222c 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java @@ -177,16 +177,15 @@ public Message readNext(int timeout, TimeUnit unit) throws PulsarClientExcept @Override public CompletableFuture> readNextAsync() { - return consumer.receiveAsync() - .whenComplete((msg, t) -> { - if (msg != null) { - consumer.acknowledgeCumulativeAsync(msg).exceptionally(ex -> { - log.error("[{}][{}] acknowledge message {} cumulative fail.", getTopic(), - getConsumer().getSubscription(), msg.getMessageId(), ex); - return null; - }); - } - }); + return consumer.receiveAsync().thenApply(msg -> { + consumer.acknowledgeCumulativeAsync(msg) + .exceptionally(ex -> { + log.error("[{}][{}] acknowledge message {} cumulative fail.", getTopic(), + getConsumer().getSubscription(), msg.getMessageId(), ex); + return null; + }); + return msg; + }); } @Override diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ReaderImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ReaderImplTest.java index 0b7b388f23217..5a1a299f41d53 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ReaderImplTest.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ReaderImplTest.java @@ -21,8 +21,10 @@ import static org.testng.AssertJUnit.assertFalse; import static org.testng.AssertJUnit.assertTrue; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import org.apache.commons.lang3.reflect.FieldUtils; import org.apache.pulsar.client.api.Consumer; import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.Schema; @@ -64,13 +66,20 @@ public void clean() { } @Test - void shouldSupportCancellingReadNextAsync() { + void shouldSupportCancellingReadNextAsync() throws IllegalAccessException { // given - CompletableFuture> future = reader.readNextAsync(); + reader.readNextAsync(); Awaitility.await().untilAsserted(() -> { assertTrue(reader.getConsumer().hasNextPendingReceive()); }); + ConsumerBase consumer = (ConsumerBase) + FieldUtils.readDeclaredField(reader, "consumer", true); + ConcurrentLinkedQueue>> + pendingReceives = (ConcurrentLinkedQueue>>) + FieldUtils.readField(consumer, "pendingReceives", true); + CompletableFuture> future = pendingReceives.peek(); + // when future.cancel(false); From ae8b0098b78cc34c75d3808828bfdd1f9d6f6f17 Mon Sep 17 00:00:00 2001 From: Heesung Sohn Date: Wed, 21 Sep 2022 10:05:56 -0700 Subject: [PATCH 3/3] Added CompletableFutureCancellationHandler in readNextAsync() --- .../client/impl/MultiTopicsReaderTest.java | 24 +++++++++++++++++++ .../client/impl/MultiTopicsReaderImpl.java | 8 ++++++- .../apache/pulsar/client/impl/ReaderImpl.java | 8 ++++++- .../pulsar/client/impl/ReaderImplTest.java | 13 ++-------- 4 files changed, 40 insertions(+), 13 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MultiTopicsReaderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MultiTopicsReaderTest.java index 6b6bf9594836a..edb5f0cd88d7c 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MultiTopicsReaderTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MultiTopicsReaderTest.java @@ -61,6 +61,7 @@ import org.apache.pulsar.common.util.Murmur3_32Hash; import org.awaitility.Awaitility; import org.testng.Assert; +import org.testng.AssertJUnit; import org.testng.annotations.AfterClass; import org.testng.annotations.BeforeClass; import org.testng.annotations.Test; @@ -625,6 +626,29 @@ public void testKeyHashRangeReader() throws Exception { } + @Test + void shouldSupportCancellingReadNextAsync() throws Exception { + String topic = "persistent://my-property/my-ns/my-reader-topic" + UUID.randomUUID(); + admin.topics().createPartitionedTopic(topic, 3); + MultiTopicsReaderImpl reader = (MultiTopicsReaderImpl) pulsarClient.newReader() + .topic(topic) + .startMessageId(MessageId.earliest) + .readerName(subscription) + .create(); + // given + CompletableFuture> future = reader.readNextAsync(); + Awaitility.await().untilAsserted(() -> { + AssertJUnit.assertTrue(reader.getMultiTopicsConsumer().hasNextPendingReceive()); + }); + + // when + future.cancel(false); + + // then + AssertJUnit.assertFalse(reader.getMultiTopicsConsumer().hasNextPendingReceive()); + } + + private void testReadMessages(String topic, boolean enableBatch) throws Exception { int numKeys = 9; diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsReaderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsReaderImpl.java index 3ec95386cb83e..cbb921ed9d6aa 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsReaderImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsReaderImpl.java @@ -41,6 +41,7 @@ import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData; import org.apache.pulsar.client.impl.conf.ReaderConfigurationData; import org.apache.pulsar.client.util.ExecutorProvider; +import org.apache.pulsar.common.util.CompletableFutureCancellationHandler; @Slf4j public class MultiTopicsReaderImpl implements Reader { @@ -146,7 +147,8 @@ public Message readNext(int timeout, TimeUnit unit) throws PulsarClientExcept @Override public CompletableFuture> readNextAsync() { - return multiTopicsConsumer.receiveAsync().thenApply(msg -> { + CompletableFuture> originalFuture = multiTopicsConsumer.receiveAsync(); + CompletableFuture> result = originalFuture.thenApply(msg -> { multiTopicsConsumer.acknowledgeCumulativeAsync(msg) .exceptionally(ex -> { log.warn("[{}][{}] acknowledge message {} cumulative fail.", getTopic(), @@ -155,6 +157,10 @@ public CompletableFuture> readNextAsync() { }); return msg; }); + CompletableFutureCancellationHandler handler = new CompletableFutureCancellationHandler(); + handler.attachToFuture(result); + handler.setCancelAction(() -> originalFuture.cancel(false)); + return result; } @Override diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java index da58db65c222c..04f8706f21cad 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java @@ -42,6 +42,7 @@ import org.apache.pulsar.client.impl.conf.ReaderConfigurationData; import org.apache.pulsar.client.util.ExecutorProvider; import org.apache.pulsar.common.naming.TopicName; +import org.apache.pulsar.common.util.CompletableFutureCancellationHandler; @Slf4j public class ReaderImpl implements Reader { @@ -177,7 +178,8 @@ public Message readNext(int timeout, TimeUnit unit) throws PulsarClientExcept @Override public CompletableFuture> readNextAsync() { - return consumer.receiveAsync().thenApply(msg -> { + CompletableFuture> originalFuture = consumer.receiveAsync(); + CompletableFuture> result = originalFuture.thenApply(msg -> { consumer.acknowledgeCumulativeAsync(msg) .exceptionally(ex -> { log.error("[{}][{}] acknowledge message {} cumulative fail.", getTopic(), @@ -186,6 +188,10 @@ public CompletableFuture> readNextAsync() { }); return msg; }); + CompletableFutureCancellationHandler handler = new CompletableFutureCancellationHandler(); + handler.attachToFuture(result); + handler.setCancelAction(() -> originalFuture.cancel(false)); + return result; } @Override diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ReaderImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ReaderImplTest.java index 5a1a299f41d53..0b7b388f23217 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ReaderImplTest.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ReaderImplTest.java @@ -21,10 +21,8 @@ import static org.testng.AssertJUnit.assertFalse; import static org.testng.AssertJUnit.assertTrue; import java.util.concurrent.CompletableFuture; -import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; -import org.apache.commons.lang3.reflect.FieldUtils; import org.apache.pulsar.client.api.Consumer; import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.Schema; @@ -66,20 +64,13 @@ public void clean() { } @Test - void shouldSupportCancellingReadNextAsync() throws IllegalAccessException { + void shouldSupportCancellingReadNextAsync() { // given - reader.readNextAsync(); + CompletableFuture> future = reader.readNextAsync(); Awaitility.await().untilAsserted(() -> { assertTrue(reader.getConsumer().hasNextPendingReceive()); }); - ConsumerBase consumer = (ConsumerBase) - FieldUtils.readDeclaredField(reader, "consumer", true); - ConcurrentLinkedQueue>> - pendingReceives = (ConcurrentLinkedQueue>>) - FieldUtils.readField(consumer, "pendingReceives", true); - CompletableFuture> future = pendingReceives.peek(); - // when future.cancel(false);