merlimat commented on a change in pull request #1205: Algorithm to find start
point of compacted ledger
URL: https://github.com/apache/incubator-pulsar/pull/1205#discussion_r167300931
##########
File path:
pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java
##########
@@ -18,10 +18,109 @@
*/
package org.apache.pulsar.compaction;
+import com.github.benmanes.caffeine.cache.AsyncLoadingCache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.google.common.collect.ComparisonChain;
+
+import java.util.NoSuchElementException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.client.LedgerEntry;
import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.pulsar.client.api.RawMessage;
+import org.apache.pulsar.client.impl.RawMessageImpl;
+import org.apache.pulsar.common.api.proto.PulsarApi.MessageIdData;
+
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class CompactedTopicImpl implements CompactedTopic {
+ final static long NEWER_THAN_COMPACTED = -0xfeed0fbaL;
+
@Override
public void newCompactedLedger(Position p, long compactedLedgerId) {}
+
+ static CompletableFuture<Long> findStartPoint(PositionImpl p,
+ long lastEntryId,
+
AsyncLoadingCache<Long,MessageIdData> cache) {
+ CompletableFuture<Long> promise = new CompletableFuture<>();
+ findStartPointLoop(p, 0, lastEntryId, promise, cache);
+ return promise;
+ }
+
+ private static void findStartPointLoop(PositionImpl p, long start, long
end,
+ CompletableFuture<Long> promise,
+
AsyncLoadingCache<Long,MessageIdData> cache) {
+ long midpoint = start + ((end - start) / 2);
+
+ CompletableFuture<MessageIdData> startEntry = cache.get(start);
+ CompletableFuture<MessageIdData> middleEntry = cache.get(midpoint);
+ CompletableFuture<MessageIdData> endEntry = cache.get(end);
+
+ CompletableFuture.allOf(startEntry, middleEntry,
endEntry).whenComplete(
+ (v, exception) -> {
+ if (exception != null) {
+ promise.completeExceptionally(exception);
+ }
+ try {
+ if (comparePositionAndMessageId(p, startEntry.get()) <
0) {
Review comment:
Also, if doing like :
```java
CompletableFuture.allOf(startEntry, middleEntry, endEntry).thenRun(() -> {
// Success case
}).exceptionally(ex -> {
promise.completeExceptionally(exception);
return null;
});
```
The `exceptionally` block will also be executed if there were any exception
in the `thenRun()` code, in addition to the async operation failing.
----------------------------------------------------------------
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
