Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[fix][tableview] fixed ack failure in ReaderImpl due to null messageId #17728

Merged
merged 3 commits into from
Sep 22, 2022
Merged

[fix][tableview] fixed ack failure in ReaderImpl due to null messageId #17728

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
16e0f74
[fix][tableview] fixed ack failure in ReaderImpl due to null messageId
heesung-sn Sep 20, 2022
2267cf7
Fixed shouldSupportCancellingReadNextAsync test
heesung-sn Sep 20, 2022
ae8b009
Added CompletableFutureCancellationHandler in readNextAsync()
heesung-sn Sep 21, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 24 additions & 0 deletions pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MultiTopicsReaderTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<byte[]> reader = (MultiTopicsReaderImpl<byte[]>) pulsarClient.newReader()
.topic(topic)
.startMessageId(MessageId.earliest)
.readerName(subscription)
.create();
// given
CompletableFuture<Message<byte[]>> 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;

Expand Down
60 changes: 60 additions & 0 deletions pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TableViewTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand All @@ -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;
Expand All @@ -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;

/**
Expand Down Expand Up @@ -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}};
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

When I try the test on my laptop.
Looks like the test for the partitioned topic will always get passed without the fix.
The non-partitioned topic test works fine, I see the warning logs. After applying the fix, the test can get passed.

2022-09-20T20:05:46,640 - WARN  - [pulsar-client-internal-39-1:ReaderImpl@184] - [persistent://public/default/tableview-no-partition-ack-test][reader-57f0eede9d] acknowledge message null cumulative fail.
org.apache.pulsar.client.api.PulsarClientException$InvalidMessageException: Cannot handle message with null messageId
	at org.apache.pulsar.client.impl.ConsumerBase.validateMessageId(ConsumerBase.java:358) ~[classes/:?]
	at org.apache.pulsar.client.impl.ConsumerBase.acknowledgeCumulativeAsync(ConsumerBase.java:554) ~[classes/:?]
	at java.lang.invoke.MethodHandle.invokeWithArguments(MethodHandle.java:732) ~[?:?]
	at org.mockito.internal.util.reflection.InstrumentationMemberAccessor$Dispatcher$ByteBuddy$ktU1uSvS.invokeWithArguments(Unknown Source) ~[?:?]
	at org.mockito.internal.util.reflection.InstrumentationMemberAccessor.invoke(InstrumentationMemberAccessor.java:239) ~[mockito-core-3.12.4.jar:?]
	at org.mockito.internal.util.reflection.ModuleMemberAccessor.invoke(ModuleMemberAccessor.java:55) ~[mockito-core-3.12.4.jar:?]
	at org.mockito.internal.creation.bytebuddy.MockMethodAdvice.tryInvoke(MockMethodAdvice.java:333) ~[mockito-core-3.12.4.jar:?]
	at org.mockito.internal.creation.bytebuddy.MockMethodAdvice.access$500(MockMethodAdvice.java:60) ~[mockito-core-3.12.4.jar:?]
	at org.mockito.internal.creation.bytebuddy.MockMethodAdvice$RealMethodCall.invoke(MockMethodAdvice.java:253) ~[mockito-core-3.12.4.jar:?]
	at org.mockito.internal.invocation.InterceptedInvocation.callRealMethod(InterceptedInvocation.java:142) ~[mockito-core-3.12.4.jar:?]
	at org.mockito.internal.stubbing.answers.CallsRealMethods.answer(CallsRealMethods.java:45) ~[mockito-core-3.12.4.jar:?]
	at org.mockito.Answers.answer(Answers.java:99) ~[mockito-core-3.12.4.jar:?]
	at org.mockito.internal.handler.MockHandlerImpl.handle(MockHandlerImpl.java:110) ~[mockito-core-3.12.4.jar:?]
	at org.mockito.internal.handler.NullResultGuardian.handle(NullResultGuardian.java:29) ~[mockito-core-3.12.4.jar:?]
	at org.mockito.internal.handler.InvocationNotifierHandler.handle(InvocationNotifierHandler.java:34) ~[mockito-core-3.12.4.jar:?]
	at org.mockito.internal.creation.bytebuddy.MockMethodInterceptor.doIntercept(MockMethodInterceptor.java:82) ~[mockito-core-3.12.4.jar:?]
	at org.mockito.internal.creation.bytebuddy.MockMethodAdvice.handle(MockMethodAdvice.java:151) ~[mockito-core-3.12.4.jar:?]
	at org.apache.pulsar.client.impl.ConsumerBase.acknowledgeCumulativeAsync(ConsumerBase.java:554) ~[classes/:?]
	at org.apache.pulsar.client.impl.ReaderImpl.lambda$readNextAsync$3(ReaderImpl.java:183) ~[classes/:?]
	at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:863) ~[?:?]
	at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:841) ~[?:?]
	at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510) ~[?:?]
	at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2147) ~[?:?]
	at org.apache.pulsar.client.impl.ConsumerImpl.lambda$internalReceiveAsync$4(ConsumerImpl.java:487) ~[classes/:?]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) ~[?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) ~[?:?]
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) ~[netty-common-4.1.77.Final.jar:4.1.77.Final]
	at java.lang.Thread.run(Thread.java:833) ~[?:?]

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, partitioned topics should pass without this fix.

But we should see the test assert failure from the non-partitioned topic, without this fix.

Please let me know if you see different behavior.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@heesung-sn Yes, the same behavior.

}

@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<String> tv1 = pulsarClient.newTableViewBuilder(Schema.STRING)
.topic(topic)
.autoUpdatePartitionsInterval(5, TimeUnit.SECONDS)
.create();

ConsumerBase consumerBase;
if (partitionedTopic) {
MultiTopicsReaderImpl<String> reader =
((CompletableFuture<MultiTopicsReaderImpl<String>>) FieldUtils
.readDeclaredField(tv1, "reader", true)).get();
consumerBase = spy(reader.getMultiTopicsConsumer());
FieldUtils.writeDeclaredField(reader, "multiTopicsConsumer", consumerBase, true);
} else {
ReaderImpl<String> reader = ((CompletableFuture<ReaderImpl<String>>) FieldUtils
.readDeclaredField(tv1, "reader", true)).get();
consumerBase = spy(reader.getConsumer());
FieldUtils.writeDeclaredField(reader, "consumer", consumerBase, true);
}

@Cleanup
Producer<String> 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)));


}
}
8 changes: 7 additions & 1 deletion pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsReaderImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<T> implements Reader<T> {
Expand Down Expand Up @@ -146,7 +147,8 @@ public Message<T> readNext(int timeout, TimeUnit unit) throws PulsarClientExcept

@Override
public CompletableFuture<Message<T>> readNextAsync() {
return multiTopicsConsumer.receiveAsync().thenApply(msg -> {
CompletableFuture<Message<T>> originalFuture = multiTopicsConsumer.receiveAsync();
CompletableFuture<Message<T>> result = originalFuture.thenApply(msg -> {
multiTopicsConsumer.acknowledgeCumulativeAsync(msg)
.exceptionally(ex -> {
log.warn("[{}][{}] acknowledge message {} cumulative fail.", getTopic(),
Expand All @@ -155,6 +157,10 @@ public CompletableFuture<Message<T>> readNextAsync() {
});
return msg;
});
CompletableFutureCancellationHandler handler = new CompletableFutureCancellationHandler();
handler.attachToFuture(result);
handler.setCancelAction(() -> originalFuture.cancel(false));
return result;
}

@Override
Expand Down
24 changes: 14 additions & 10 deletions pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<T> implements Reader<T> {
Expand Down Expand Up @@ -177,17 +178,20 @@ public Message<T> readNext(int timeout, TimeUnit unit) throws PulsarClientExcept

@Override
public CompletableFuture<Message<T>> readNextAsync() {
CompletableFuture<Message<T>> 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;
});
}
CompletableFuture<Message<T>> originalFuture = consumer.receiveAsync();
CompletableFuture<Message<T>> result = originalFuture.thenApply(msg -> {
consumer.acknowledgeCumulativeAsync(msg)
.exceptionally(ex -> {
log.error("[{}][{}] acknowledge message {} cumulative fail.", getTopic(),
getConsumer().getSubscription(), msg.getMessageId(), ex);
return null;
});
return msg;
});
return receiveFuture;
CompletableFutureCancellationHandler handler = new CompletableFutureCancellationHandler();
handler.attachToFuture(result);
handler.setCancelAction(() -> originalFuture.cancel(false));
return result;
}

@Override
Expand Down