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_r167300225
########## 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: Nit: you could use `.join()` instead of `get()`, that throws unchecked exception, since in this case we're not expecting any. ---------------------------------------------------------------- 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