merlimat closed pull request #1286: Compaction uses GetLastMessageId URL: https://github.com/apache/incubator-pulsar/pull/1286
This is a PR merged from a forked repository. As GitHub hides the original diff on merge, it is displayed below for the sake of provenance: As this is a foreign pull request (from a fork), the diff is supplied below (as it won't show otherwise due to GitHub magic): diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/client/api/RawReader.java b/pulsar-broker/src/main/java/org/apache/pulsar/client/api/RawReader.java index b5a4a6575..15b03bee3 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/client/api/RawReader.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/client/api/RawReader.java @@ -20,6 +20,7 @@ import java.util.Map; import java.util.concurrent.CompletableFuture; +import org.apache.pulsar.client.api.MessageId; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.impl.PulsarClientImpl; import org.apache.pulsar.client.impl.RawReaderImpl; @@ -60,6 +61,11 @@ */ CompletableFuture<Void> acknowledgeCumulativeAsync(MessageId messageId, Map<String,Long> properties); + /** + * Get the last message id available immediately available for reading + */ + CompletableFuture<MessageId> getLastMessageIdAsync(); + /** * Close the raw reader. */ 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 04aa61039..a048a5bb2 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 @@ -78,6 +78,11 @@ public RawReaderImpl(PulsarClientImpl client, String topic, String subscription, return consumer.closeAsync(); } + @Override + public CompletableFuture<MessageId> getLastMessageIdAsync() { + return consumer.getLastMessageIdAsync(); + } + static class RawConsumerImpl extends ConsumerImpl { final BlockingQueue<RawMessageAndCnx> incomingRawMessages; final Queue<CompletableFuture<RawMessage>> pendingRawReceives; diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/TwoPhaseCompactor.java b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/TwoPhaseCompactor.java index 957e1970c..10dc8b129 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/TwoPhaseCompactor.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/TwoPhaseCompactor.java @@ -75,15 +75,22 @@ public TwoPhaseCompactor(ServiceConfiguration conf, private CompletableFuture<PhaseOneResult> phaseOne(RawReader reader) { Map<String,MessageId> latestForKey = new HashMap<>(); - CompletableFuture<PhaseOneResult> loopPromise = new CompletableFuture<>(); - phaseOneLoop(reader, Optional.empty(), Optional.empty(), latestForKey, loopPromise); + + reader.getLastMessageIdAsync().whenComplete( + (lastMessageId, exception) -> { + if (exception != null) { + loopPromise.completeExceptionally(exception); + } else { + phaseOneLoop(reader, Optional.empty(), lastMessageId, latestForKey, loopPromise); + } + }); return loopPromise; } private void phaseOneLoop(RawReader reader, Optional<MessageId> firstMessageId, - Optional<MessageId> lastMessageId, + MessageId lastMessageId, Map<String,MessageId> latestForKey, CompletableFuture<PhaseOneResult> loopPromise) { if (loopPromise.isDone()) { @@ -91,34 +98,30 @@ private void phaseOneLoop(RawReader reader, } CompletableFuture<RawMessage> future = reader.readNextAsync(); scheduleTimeout(future); - future.whenComplete( + future.whenCompleteAsync( (m, exception) -> { try { if (exception != null) { - if (exception instanceof TimeoutException - && firstMessageId.isPresent()) { - loopPromise.complete(new PhaseOneResult(firstMessageId.get(), - lastMessageId.get(), - latestForKey)); - } else { - loopPromise.completeExceptionally(exception); - } + loopPromise.completeExceptionally(exception); return; } - MessageId id = m.getMessageId(); String key = extractKey(m); latestForKey.put(key, id); - phaseOneLoop(reader, - Optional.of(firstMessageId.orElse(id)), - Optional.of(id), - latestForKey, loopPromise); + if (id.compareTo(lastMessageId) == 0) { + loopPromise.complete(new PhaseOneResult(firstMessageId.orElse(id), + id, latestForKey)); + } else { + phaseOneLoop(reader, + Optional.of(firstMessageId.orElse(id)), + lastMessageId, + latestForKey, loopPromise); + } } finally { m.close(); } - }); - + }, scheduler); } private void scheduleTimeout(CompletableFuture<RawMessage> future) { @@ -168,7 +171,7 @@ private void scheduleTimeout(CompletableFuture<RawMessage> future) { private void phaseTwoLoop(RawReader reader, MessageId to, Map<String, MessageId> latestForKey, LedgerHandle lh, Semaphore outstanding, CompletableFuture<Void> promise) { - reader.readNextAsync().whenComplete( + reader.readNextAsync().whenCompleteAsync( (m, exception) -> { try { if (exception != null) { @@ -205,7 +208,7 @@ private void phaseTwoLoop(RawReader reader, MessageId to, Map<String, MessageId> } finally { m.close(); } - }); + }, scheduler); } private CompletableFuture<LedgerHandle> createLedger(BookKeeper bk) { 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 19cdc3155..53da2eefa 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 @@ -131,16 +131,16 @@ public void testRawReader() throws Exception { Set<String> keys = publishMessages(topic, numKeys); RawReader reader = RawReader.create(pulsarClient, topic, subscription).get(); - try { - while (true) { // should break out with TimeoutException - try (RawMessage m = reader.readNextAsync().get(1, TimeUnit.SECONDS)) { - Assert.assertTrue(keys.remove(extractKey(m))); + + MessageId lastMessageId = reader.getLastMessageIdAsync().get(); + while (true) { + try (RawMessage m = reader.readNextAsync().get()) { + Assert.assertTrue(keys.remove(extractKey(m))); + if (lastMessageId.compareTo(m.getMessageId()) == 0) { + break; } } - } catch (TimeoutException te) { - // ok } - Assert.assertTrue(keys.isEmpty()); } @@ -153,28 +153,27 @@ public void testSeekToStart() throws Exception { Set<String> readKeys = new HashSet<>(); RawReader reader = RawReader.create(pulsarClient, topic, subscription).get(); - try { - while (true) { // should break out with TimeoutException - try (RawMessage m = reader.readNextAsync().get(1, TimeUnit.SECONDS)) { - readKeys.add(extractKey(m)); + MessageId lastMessageId = reader.getLastMessageIdAsync().get(); + while (true) { + try (RawMessage m = reader.readNextAsync().get()) { + readKeys.add(extractKey(m)); + if (lastMessageId.compareTo(m.getMessageId()) == 0) { + break; } } - } catch (TimeoutException te) { - // ok } Assert.assertEquals(readKeys.size(), numKeys); // seek to start, read all keys again, // assert that we read all keys we had read previously reader.seekAsync(MessageId.earliest).get(); - try { - while (true) { // should break out with TimeoutException - try (RawMessage m = reader.readNextAsync().get(1, TimeUnit.SECONDS)) { - Assert.assertTrue(readKeys.remove(extractKey(m))); + while (true) { + try (RawMessage m = reader.readNextAsync().get()) { + Assert.assertTrue(readKeys.remove(extractKey(m))); + if (lastMessageId.compareTo(m.getMessageId()) == 0) { + break; } } - } catch (TimeoutException te) { - // ok } Assert.assertTrue(readKeys.isEmpty()); } @@ -190,34 +189,34 @@ public void testSeekToMiddle() throws Exception { RawReader reader = RawReader.create(pulsarClient, topic, subscription).get(); int i = 0; MessageId seekTo = null; - try { - while (true) { // should break out with TimeoutException - try (RawMessage m = reader.readNextAsync().get(1, TimeUnit.SECONDS)) { - i++; - if (i > numKeys/2) { - if (seekTo == null) { - seekTo = m.getMessageId(); - } - readKeys.add(extractKey(m)); + MessageId lastMessageId = reader.getLastMessageIdAsync().get(); + + while (true) { + try (RawMessage m = reader.readNextAsync().get()) { + i++; + if (i > numKeys/2) { + if (seekTo == null) { + seekTo = m.getMessageId(); } + readKeys.add(extractKey(m)); + } + if (lastMessageId.compareTo(m.getMessageId()) == 0) { + break; } } - } catch (TimeoutException te) { - // ok } Assert.assertEquals(readKeys.size(), numKeys/2); // seek to middle, read all keys again, // assert that we read all keys we had read previously reader.seekAsync(seekTo).get(); - try { - while (true) { // should break out with TimeoutException - try (RawMessage m = reader.readNextAsync().get(1, TimeUnit.SECONDS)) { - Assert.assertTrue(readKeys.remove(extractKey(m))); + while (true) { // should break out with TimeoutException + try (RawMessage m = reader.readNextAsync().get()) { + Assert.assertTrue(readKeys.remove(extractKey(m))); + if (lastMessageId.compareTo(m.getMessageId()) == 0) { + break; } } - } catch (TimeoutException te) { - // ok } Assert.assertTrue(readKeys.isEmpty()); } @@ -280,18 +279,18 @@ public void accept(RawMessage m) { } } }; - try { - while (true) { // should break out with TimeoutException - try (RawMessage m = reader.readNextAsync().get(1, TimeUnit.SECONDS)) { - if (RawBatchConverter.isBatch(m)) { - RawBatchConverter.explodeBatch(m).forEach(consumer); - } else { - consumer.accept(m); - } + MessageId lastMessageId = reader.getLastMessageIdAsync().get(); + while (true) { + try (RawMessage m = reader.readNextAsync().get()) { + if (RawBatchConverter.isBatch(m)) { + RawBatchConverter.explodeBatch(m).forEach(consumer); + } else { + consumer.accept(m); + } + if (lastMessageId.compareTo(m.getMessageId()) == 0) { + break; } } - } catch (TimeoutException te) { - // ok } Assert.assertTrue(keys.isEmpty()); } @@ -304,24 +303,23 @@ public void testAcknowledgeWithProperties() throws Exception { Set<String> keys = publishMessages(topic, numKeys); - MessageId lastMessageId = null; RawReader reader = RawReader.create(pulsarClient, topic, subscription).get(); - try { - while (true) { // should break out with TimeoutException - try (RawMessage m = reader.readNextAsync().get(1, TimeUnit.SECONDS)) { - lastMessageId = m.getMessageId(); - Assert.assertTrue(keys.remove(extractKey(m))); + MessageId lastMessageId = reader.getLastMessageIdAsync().get(); + + while (true) { + try (RawMessage m = reader.readNextAsync().get()) { + Assert.assertTrue(keys.remove(extractKey(m))); + + if (lastMessageId.compareTo(m.getMessageId()) == 0) { + break; } } - } catch (TimeoutException te) { - // ok } - Assert.assertTrue(keys.isEmpty()); Map<String,Long> properties = new HashMap<>(); properties.put("foobar", 0xdeadbeefdecaL); - reader.acknowledgeCumulativeAsync(lastMessageId, properties).get(5, TimeUnit.SECONDS); + reader.acknowledgeCumulativeAsync(lastMessageId, properties).get(); PersistentTopic topicRef = (PersistentTopic) pulsar.getBrokerService().getTopicReference(topic); ManagedLedger ledger = topicRef.getManagedLedger(); @@ -349,12 +347,12 @@ public void testReadCancellationOnClose() throws Exception { } for (int i = 0; i < numKeys/2; i++) { - futures.remove(0).get(5, TimeUnit.SECONDS); // complete successfully + futures.remove(0).get(); // complete successfully } reader.closeAsync().get(); while (!futures.isEmpty()) { try { - futures.remove(0).get(5, TimeUnit.SECONDS); + futures.remove(0).get(); Assert.fail("Should have been cancelled"); } catch (CancellationException ee) { // correct behaviour diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java index e9c781764..4e526cc16 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java @@ -1315,7 +1315,7 @@ public boolean hasMessageAvailable() throws PulsarClientException { return booleanFuture; } - private CompletableFuture<MessageId> getLastMessageIdAsync() { + CompletableFuture<MessageId> getLastMessageIdAsync() { if (getState() == State.Closing || getState() == State.Closed) { return FutureUtil .failedFuture(new PulsarClientException.AlreadyClosedException("Consumer was already closed")); ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services