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
