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
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<Future<RawMessage>> 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
+            }
+        }
+    }
 }