IllegalArgumentException("inconsistent range") from ConcurrentSkipListMap

fix

Author: maxlisongsong

bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/ReplicationWorker.java

@@ -146,6 +146,7 @@ public class ReplicationWorker implements Runnable {
     private final Counter numNotAdheringPlacementLedgersReplicated;
     private final Map<String, Counter> exceptionCounters;
     final LoadingCache<Long, AtomicInteger> replicationFailedLedgers;
+    @VisibleForTesting
     final LoadingCache<Long, ConcurrentSkipListSet<Long>> unableToReadEntriesForReplication;
 
     /**
@@ -322,8 +323,7 @@ private void logBKExceptionAndReleaseLedger(BKException e, long ledgerIdToReplic
             .releaseUnderreplicatedLedger(ledgerIdToReplicate);
         getExceptionCounter(e.getClass().getSimpleName()).inc();
     }
-
-    private boolean tryReadingFaultyEntries(LedgerHandle lh, LedgerFragment ledgerFragment) {
+    public boolean tryReadingFaultyEntries(LedgerHandle lh, LedgerFragment ledgerFragment) {
         long ledgerId = lh.getId();
         ConcurrentSkipListSet<Long> entriesUnableToReadForThisLedger = unableToReadEntriesForReplication
                 .getIfPresent(ledgerId);
@@ -332,6 +332,9 @@ private boolean tryReadingFaultyEntries(LedgerHandle lh, LedgerFragment ledgerFr
         }
         long firstEntryIdOfFragment = ledgerFragment.getFirstEntryId();
         long lastEntryIdOfFragment = ledgerFragment.getLastKnownEntryId();
+        if (firstEntryIdOfFragment > lastEntryIdOfFragment) {
+            return true;
+        }
         NavigableSet<Long> entriesOfThisFragmentUnableToRead = entriesUnableToReadForThisLedger
                 .subSet(firstEntryIdOfFragment, true, lastEntryIdOfFragment, true);
         if (entriesOfThisFragmentUnableToRead.isEmpty()) {

bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/TestReplicationWorker.java

@@ -37,6 +37,7 @@
 import java.net.URI;
 import java.net.UnknownHostException;
 import java.util.Enumeration;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map.Entry;
 import java.util.Objects;
@@ -57,6 +58,7 @@
 import org.apache.bookkeeper.client.ClientUtil;
 import org.apache.bookkeeper.client.EnsemblePlacementPolicy;
 import org.apache.bookkeeper.client.LedgerEntry;
+import org.apache.bookkeeper.client.LedgerFragment;
 import org.apache.bookkeeper.client.LedgerHandle;
 import org.apache.bookkeeper.client.RackawareEnsemblePlacementPolicy;
 import org.apache.bookkeeper.client.ZoneawareEnsemblePlacementPolicy;
@@ -1296,6 +1298,33 @@ public void testReplicationStats() throws Exception {
                 RackawareEnsemblePlacementPolicy.class, checkReplicationStats);
     }
 
+    /**
+     * Test the tryReadingFaultyEntries function, where firstEntryId > lastKnownEntryId
+     * and unableToReadEntriesForReplication not empty, there is a IllegalArgumentException.
+     * java.lang.IllegalArgumentException: inconsistent range
+     */
+    @Test
+    public void testTryReadingFaultyEntriesFromEmptyFragment() throws Exception {
+        LedgerHandle lh = bkc.createLedger(3, 3, BookKeeper.DigestType.CRC32,
+                TESTPASSWD);
+
+        HashSet<Integer> bookieIndexes = new HashSet<>();
+        bookieIndexes.add(0);
+        bookieIndexes.add(1);
+        bookieIndexes.add(2);
+        LedgerFragment fragment = new LedgerFragment(lh, 2, 1, bookieIndexes);
+
+        ReplicationWorker rw = new ReplicationWorker(baseConf);
+        rw.start();
+        rw.unableToReadEntriesForReplication.getUnchecked(lh.getId()).add(0L);
+
+        try {
+            rw.tryReadingFaultyEntries(lh, fragment);
+        } catch (IllegalArgumentException e) {
+            fail("should fail with exception: " + e);
+        }
+    }
+
     private void testRepairedNotAdheringPlacementPolicyLedgerFragments(
             Class<? extends EnsemblePlacementPolicy> placementPolicyClass,
             BiConsumer<Boolean, ReplicationWorker> checkReplicationStats) throws Exception {