From 70e09fe40f56d4c4ed09119dbcba52e872c0f8a1 Mon Sep 17 00:00:00 2001 From: Ivan Kelly Date: Tue, 9 Jan 2018 17:01:20 +0100 Subject: [PATCH 1/2] RawReader allows futures to be cancelled RawReader#readNextAsync returns a Future to the user, which includes a retained ByteBuf. If the user is no longer interested in the result of the read call, it should be able to cancel to ensure that the ByteBuf is released. If there are multiple read requests outstanding, and one of them is cancelled, all read requests that were made after the cancelled request will also be cancelled. --- .../apache/pulsar/client/impl/RawReaderImpl.java | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java index 8057220acd50d..f66685f6a73cd 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java @@ -115,7 +115,10 @@ void tryCompletePending() { if (future == null) { assert(messageAndCnx == null); } else { - future.complete(messageAndCnx.msg); + if (!future.complete(messageAndCnx.msg)) { + messageAndCnx.msg.close(); + closeAsync(); + } ClientCnx currentCnx = cnx(); if (currentCnx == messageAndCnx.cnx) { @@ -137,10 +140,14 @@ private void reset() { while (!pendingRawReceives.isEmpty()) { toError.add(pendingRawReceives.remove()); } + RawMessageAndCnx m = incomingRawMessages.poll(); + while (m != null) { + m.msg.close(); + m = incomingRawMessages.poll(); + } incomingRawMessages.clear(); } - toError.forEach((f) -> f.completeExceptionally( - new PulsarClientException.ConsumerBusyException("Sought while reading"))); + toError.forEach((f) -> f.cancel(false)); } @Override From 97e26ee5224360def58e98b6c9a4f9d7b235e650 Mon Sep 17 00:00:00 2001 From: Ivan Kelly Date: Wed, 10 Jan 2018 14:55:11 +0100 Subject: [PATCH 2/2] Test that all outstanding futures are cancelled on close --- .../pulsar/client/impl/RawReaderTest.java | 28 +++++++++++++++++++ 1 file changed, 28 insertions(+) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawReaderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawReaderTest.java index 34eff5d31725c..cef3e428a278d 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawReaderTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawReaderTest.java @@ -31,6 +31,7 @@ import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.CancellationException; import java.util.concurrent.Future; import java.util.concurrent.TimeoutException; import java.util.concurrent.TimeUnit; @@ -333,4 +334,31 @@ public void testAcknowledgeWithProperties() throws Exception { Assert.assertEquals(ledger.openCursor(subscription).getProperties().get("foobar"), Long.valueOf(0xdeadbeefdecaL)); } + + @Test + public void testReadCancellationOnClose() throws Exception { + int numKeys = 10; + + String topic = "persistent://my-property/use/my-ns/my-raw-topic"; + publishMessages(topic, numKeys/2); + + RawReader reader = RawReader.create(pulsarClient, topic, subscription).get(); + List> futures = new ArrayList<>(); + for (int i = 0; i < numKeys; i++) { + futures.add(reader.readNextAsync()); + } + + for (int i = 0; i < numKeys/2; i++) { + futures.remove(0).get(5, TimeUnit.SECONDS); // complete successfully + } + reader.closeAsync().get(); + while (!futures.isEmpty()) { + try { + futures.remove(0).get(5, TimeUnit.SECONDS); + Assert.fail("Should have been cancelled"); + } catch (CancellationException ee) { + // correct behaviour + } + } + } }