CASSANALYTICS-180: TokenPartitioner fails to detect range gap in reader

CHANGES.txt

@@ -1,5 +1,6 @@
 0.5.0
 -----
+ * TokenPartitioner fails to detect range gap in reader (CASSANALYTICS-180)
  * CDC reader stats silently dropped in SidecarCdcBuilder (CASSANALYTICS-191)
  * Add CapturePublishedSchema metric to SidecarCdcStats (CASSANALYTICS-189)
  * Expand list of architecture that supports unaligned access in FastByteOperations (CASSANALYTICS-188)

cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/partitioner/TokenPartitioner.java

@@ -37,9 +37,7 @@
 import com.google.common.collect.BoundType;
 import com.google.common.collect.Range;
 import com.google.common.collect.RangeMap;
-import com.google.common.collect.RangeSet;
 import com.google.common.collect.TreeRangeMap;
-import com.google.common.collect.TreeRangeSet;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -205,17 +203,10 @@ private void validateRangesDoNotOverlap()
 
     private void validateCompleteRangeCoverage()
     {
-        RangeSet<BigInteger> missingRangeSet = TreeRangeSet.create();
-        missingRangeSet.add(Range.closed(ring.partitioner().minToken(),
-                                         ring.partitioner().maxToken()));
-
-        partitionMap.asMapOfRanges().keySet().forEach(missingRangeSet::remove);
-
-        List<Range<BigInteger>> missingRanges = missingRangeSet.asRanges().stream()
-                                                                          .filter(Range::isEmpty)
-                                                                          .collect(Collectors.toList());
+        List<Range<BigInteger>> missingRanges = RangeUtils.findUncoveredRingRanges(ring.partitioner(),
+                                                                                  partitionMap.asMapOfRanges().keySet());
         Preconditions.checkState(missingRanges.isEmpty(),
-                                 "There should be no missing ranges, but found " + missingRanges.toString());
+                                 "There should be no missing ranges, but found %s", missingRanges);
     }
 
     private void validateMapSizes()

cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/RangeUtils.java

@@ -21,14 +21,18 @@
 
 import java.math.BigInteger;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ArrayListMultimap;
 import com.google.common.collect.BoundType;
 import com.google.common.collect.Multimap;
 import com.google.common.collect.Range;
+import com.google.common.collect.RangeSet;
+import com.google.common.collect.TreeRangeSet;
 
 import org.apache.cassandra.bridge.TokenRange;
 import org.apache.cassandra.spark.data.model.TokenOwner;
@@ -70,6 +74,56 @@ public static boolean isOpenClosedRange(Range<?> range)
         return range.lowerBoundType() == BoundType.OPEN && range.upperBoundType() == BoundType.CLOSED;
     }
 
+    /**
+     * Finds the sub-ranges of the whole token ring that none of the {@code coveringRanges} covers, i.e. the gaps in
+     * the ring coverage. An empty result means the ring is covered in its entirety.
+     * <p>
+     * The ring is expressed as {@code (minToken, maxToken]}, matching the open-closed notation used for token
+     * ranges everywhere else. Expressing it as {@code [minToken, maxToken]} instead would report minToken as a
+     * spurious single-token gap, because open-closed ranges never cover their own lower endpoint.
+     *
+     * @param partitioner the partitioner whose ring bounds are expected to be covered
+     * @param coveringRanges the ranges expected to cover the ring; they may overlap and need not be sorted
+     * @return the uncovered sub-ranges of the ring, in ascending order
+     */
+    public static List<Range<BigInteger>> findUncoveredRingRanges(Partitioner partitioner,
+                                                                  Collection<Range<BigInteger>> coveringRanges)
+    {
+        return findUncoveredRanges(Range.openClosed(partitioner.minToken(), partitioner.maxToken()), coveringRanges);
+    }
+
+    /**
+     * Finds the sub-ranges of {@code fullRange} that none of the {@code coveringRanges} covers, i.e. the gaps in
+     * the coverage. An empty result means {@code coveringRanges} covers {@code fullRange} in its entirety.
+     * <p>
+     * Both the input and the output honor Guava's bound types, so {@code fullRange} must be expressed using the
+     * same notation as the covering ranges. That makes the method easy to misuse, which is why it is not exposed
+     * beyond this class: callers go through {@link #findUncoveredRingRanges}, which owns the notation so that the
+     * bound types are not decided at each call site. Empty ranges are never reported, as a range that contains no
+     * token cannot be a gap.
+     *
+     * @param fullRange the non-empty range expected to be fully covered
+     * @param coveringRanges the ranges expected to cover {@code fullRange}; they may overlap and need not be sorted
+     * @return the uncovered sub-ranges of {@code fullRange}, in ascending order
+     */
+    @VisibleForTesting
+    static List<Range<BigInteger>> findUncoveredRanges(Range<BigInteger> fullRange,
+                                                       Collection<Range<BigInteger>> coveringRanges)
+    {
+        // An empty fullRange has nothing to cover, so every input would trivially look fully covered.
+        // Reject it rather than report a false "no gaps".
+        Preconditions.checkArgument(!fullRange.isEmpty(), "fullRange must not be empty");
+
+        RangeSet<BigInteger> uncovered = TreeRangeSet.create();
+        uncovered.add(fullRange);
+        // Not removeAll: on Guava 16.0.1 it only accepts a RangeSet, not a Collection, and it is itself just this
+        // loop of remove() calls, so there is nothing to gain by wrapping the input in a second range set
+        coveringRanges.forEach(uncovered::remove);
+        // TreeRangeSet coalesces connected ranges and never retains empty ones,
+        // so everything left is a real, non-empty gap
+        return new ArrayList<>(uncovered.asRanges());
+    }
+
     /**
      * Splits the given range into equal-sized small ranges. Number of splits can be controlled by
      * nrSplits. If nrSplits are smaller than size of the range, split size would be set to 1, which is

cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/partitioner/TokenPartitionerValidationTest.java

@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.cassandra.spark.data.partitioner;
+
+import java.math.BigInteger;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Range;
+import org.junit.jupiter.api.Test;
+
+import org.apache.cassandra.spark.data.ReplicationFactor;
+import org.apache.cassandra.spark.utils.RangeUtils;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+public class TokenPartitionerValidationTest
+{
+    private static final Partitioner PARTITIONER = Partitioner.Murmur3Partitioner;
+
+    @Test
+    public void testValidationDetectsRangeGap()
+    {
+        List<Range<BigInteger>> subRanges = RangeUtils.split(wholeRing(), 4);
+
+        assertThatThrownBy(() -> new TokenPartitioner(withGapAt(subRanges, 2), ring()))
+        .isInstanceOf(IllegalStateException.class)
+        .hasMessageContaining("There should be no missing ranges")
+        .hasMessageContaining(gapPunchedInto(subRanges.get(2)).toString());
+    }
+
+    @Test
+    public void testValidationDetectsRangeGapAtRingLowerEdge()
+    {
+        // Guards the bound type at minToken from both sides: minToken itself is owned by no sub-range and must not
+        // be reported, yet a gap starting immediately above it must still be caught. The gap is punched into the
+        // first sub-range rather than dropping it, so that the partition count stays put and validateMapSizes
+        // cannot fail first with an unrelated message.
+        List<Range<BigInteger>> subRanges = RangeUtils.split(wholeRing(), 4);
+
+        assertThatThrownBy(() -> new TokenPartitioner(withGapAt(subRanges, 0), ring()))
+        .isInstanceOf(IllegalStateException.class)
+        .hasMessageContaining("There should be no missing ranges")
+        .hasMessageContaining(gapPunchedInto(subRanges.get(0)).toString());
+    }
+
+    @Test
+    public void testValidationAcceptsCompleteRangeCoverage()
+    {
+        // minToken is deliberately left uncovered: the sub-ranges are open-closed, so it belongs to none of them.
+        // Validation must not report it as a gap, otherwise every job fails on a healthy ring.
+        TokenPartitioner partitioner = new TokenPartitioner(RangeUtils.split(wholeRing(), 4), ring());
+        assertThat(partitioner.numPartitions()).isEqualTo(4);
+    }
+
+    private static Range<BigInteger> wholeRing()
+    {
+        return Range.openClosed(PARTITIONER.minToken(), PARTITIONER.maxToken());
+    }
+
+    /**
+     * Punches a real, non-empty gap into the sub-range at {@code gapIndex} by moving its lower endpoint up, so that
+     * the returned ranges leave exactly {@link #gapPunchedInto} uncovered.
+     */
+    private static List<Range<BigInteger>> withGapAt(List<Range<BigInteger>> gapFreeRanges, int gapIndex)
+    {
+        List<Range<BigInteger>> ranges = new ArrayList<>(gapFreeRanges);
+        Range<BigInteger> covered = ranges.get(gapIndex);
+        ranges.set(gapIndex, Range.openClosed(gapPunchedInto(covered).upperEndpoint(), covered.upperEndpoint()));
+        return ranges;
+    }
+
+    /**
+     * @return the sub-range that {@link #withGapAt} leaves uncovered when it punches a gap into {@code range}
+     */
+    private static Range<BigInteger> gapPunchedInto(Range<BigInteger> range)
+    {
+        return Range.openClosed(range.lowerEndpoint(), range.lowerEndpoint().add(BigInteger.TEN));
+    }
+
+    private static CassandraRing ring()
+    {
+        List<CassandraInstance> instances = Arrays.asList(new CassandraInstance("0", "local0-i1", "DEV"),
+                                                          new CassandraInstance("100", "local0-i2", "DEV"),
+                                                          new CassandraInstance("200", "local0-i3", "DEV"));
+        return new CassandraRing(PARTITIONER,
+                                 "test",
+                                 new ReplicationFactor(ImmutableMap.of("class", "NetworkTopologyStrategy", "DEV", "3")),
+                                 instances);
+    }
+}

cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/utils/RangeUtilsTest.java

@@ -230,6 +230,169 @@ void testSplitNotSatisfyNrSplits()
         assertThat(RangeUtils.split(range, nrSplits)).isEqualTo(expectedResult);
     }
 
+    @Test
+    void testFindUncoveredRangesWithCompleteCoverage()
+    {
+        Range<BigInteger> fullRange = Range.openClosed(BigInteger.ZERO, BigInteger.valueOf(30));
+        List<Range<BigInteger>> covering = Arrays.asList(
+        Range.openClosed(BigInteger.ZERO, BigInteger.TEN),
+        Range.openClosed(BigInteger.TEN, BigInteger.valueOf(20)),
+        Range.openClosed(BigInteger.valueOf(20), BigInteger.valueOf(30))
+        );
+        assertThat(RangeUtils.findUncoveredRanges(fullRange, covering)).isEmpty();
+    }
+
+    @Test
+    void testFindUncoveredRangesDetectsGap()
+    {
+        // leaves (10, 20] uncovered -- a real, non-empty gap
+        Range<BigInteger> fullRange = Range.openClosed(BigInteger.ZERO, BigInteger.valueOf(30));
+        List<Range<BigInteger>> covering = Arrays.asList(
+        Range.openClosed(BigInteger.ZERO, BigInteger.TEN),
+        Range.openClosed(BigInteger.valueOf(20), BigInteger.valueOf(30))
+        );
+        assertThat(RangeUtils.findUncoveredRanges(fullRange, covering))
+        .containsExactly(Range.openClosed(BigInteger.TEN, BigInteger.valueOf(20)));
+    }
+
+    @Test
+    void testFindUncoveredRangesDetectsMultipleGapsInAscendingOrder()
+    {
+        Range<BigInteger> fullRange = Range.openClosed(BigInteger.ZERO, BigInteger.valueOf(40));
+        List<Range<BigInteger>> covering = Arrays.asList(
+        Range.openClosed(BigInteger.valueOf(20), BigInteger.valueOf(30)),
+        Range.openClosed(BigInteger.ZERO, BigInteger.TEN)
+        );
+        assertThat(RangeUtils.findUncoveredRanges(fullRange, covering))
+        .containsExactly(Range.openClosed(BigInteger.TEN, BigInteger.valueOf(20)),
+                         Range.openClosed(BigInteger.valueOf(30), BigInteger.valueOf(40)));
+    }
+
+    @Test
+    void testFindUncoveredRangesToleratesOverlappingAndUnsortedInput()
+    {
+        Range<BigInteger> fullRange = Range.openClosed(BigInteger.ZERO, BigInteger.valueOf(30));
+        List<Range<BigInteger>> covering = Arrays.asList(
+        Range.openClosed(BigInteger.valueOf(15), BigInteger.valueOf(30)),
+        Range.openClosed(BigInteger.ZERO, BigInteger.valueOf(20))
+        );
+        assertThat(RangeUtils.findUncoveredRanges(fullRange, covering)).isEmpty();
+    }
+
+    @Test
+    void testFindUncoveredRangesToleratesEmptyCoveringRanges()
+    {
+        // (5, 5] is degenerate: it covers no token, so removing it is a no-op and the ranges either side still cover
+        Range<BigInteger> fullRange = Range.openClosed(BigInteger.ZERO, BigInteger.TEN);
+        List<Range<BigInteger>> covering = Arrays.asList(
+        Range.openClosed(BigInteger.ZERO, BigInteger.valueOf(5)),
+        Range.openClosed(BigInteger.valueOf(5), BigInteger.valueOf(5)),
+        Range.openClosed(BigInteger.valueOf(5), BigInteger.TEN)
+        );
+        assertThat(RangeUtils.findUncoveredRanges(fullRange, covering)).isEmpty();
+    }
+
+    @Test
+    void testFindUncoveredRangesRejectsEmptyFullRange()
+    {
+        // Nothing can cover an empty range, so reporting it as fully covered would be a false negative
+        Range<BigInteger> emptyRange = Range.openClosed(BigInteger.TEN, BigInteger.TEN);
+        assertThatThrownBy(() -> RangeUtils.findUncoveredRanges(emptyRange, Collections.emptyList()))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("fullRange must not be empty");
+    }
+
+    @Test
+    void testFindUncoveredRingRangesDetectsGap()
+    {
+        for (Partitioner partitioner : Partitioner.values())
+        {
+            List<Range<BigInteger>> ring = RangeUtils.split(wholeRing(partitioner), 4);
+            assertThat(RangeUtils.findUncoveredRingRanges(partitioner, withGapAt(ring, 2)))
+            .as("A gap in the %s ring should be reported", partitioner)
+            .containsExactly(gapPunchedInto(ring.get(2)));
+        }
+    }
+
+    @Test
+    void testFindUncoveredRingRangesDetectsGapAtRingLowerEdge()
+    {
+        // The bound type at minToken is the whole subtlety of findUncoveredRingRanges: it must not report minToken
+        // itself as a gap, yet it must still report a gap that starts immediately above minToken
+        for (Partitioner partitioner : Partitioner.values())
+        {
+            List<Range<BigInteger>> ring = RangeUtils.split(wholeRing(partitioner), 4);
+            assertThat(RangeUtils.findUncoveredRingRanges(partitioner, withGapAt(ring, 0)))
+            .as("A gap at the lower edge of the %s ring should be reported", partitioner)
+            .containsExactly(gapPunchedInto(ring.get(0)));
+        }
+    }
+
+    @Test
+    void testFindUncoveredRingRangesDetectsMissingFirstSubRange()
+    {
+        // Dropping the first sub-range leaves everything above minToken up to its upper endpoint uncovered
+        for (Partitioner partitioner : Partitioner.values())
+        {
+            List<Range<BigInteger>> ring = RangeUtils.split(wholeRing(partitioner), 4);
+            assertThat(RangeUtils.findUncoveredRingRanges(partitioner, ring.subList(1, ring.size())))
+            .as("A missing first sub-range of the %s ring should be reported", partitioner)
+            .containsExactly(Range.openClosed(partitioner.minToken(), ring.get(0).upperEndpoint()));
+        }
+    }
+
+    @Test
+    void testFindUncoveredRingRangesDetectsMissingLastSubRange()
+    {
+        // The counterpart of the above at the other end of the ring, where maxToken is inclusive
+        for (Partitioner partitioner : Partitioner.values())
+        {
+            List<Range<BigInteger>> ring = RangeUtils.split(wholeRing(partitioner), 4);
+            assertThat(RangeUtils.findUncoveredRingRanges(partitioner, ring.subList(0, ring.size() - 1)))
+            .as("A missing last sub-range of the %s ring should be reported", partitioner)
+            .containsExactly(ring.get(ring.size() - 1));
+        }
+    }
+
+    @Test
+    void testFindUncoveredRingRangesOverWholeRingIsGapFree()
+    {
+        // A gap-free split of the whole ring leaves minToken uncovered, because the sub-ranges are open-closed.
+        // findUncoveredRingRanges owns that bound-type decision, so no caller can get it wrong: it must not
+        // report [minToken, minToken] as a gap.
+        for (Partitioner partitioner : Partitioner.values())
+        {
+            assertThat(RangeUtils.findUncoveredRingRanges(partitioner, RangeUtils.split(wholeRing(partitioner), 16)))
+            .as("Splitting the whole %s ring should leave no gap", partitioner)
+            .isEmpty();
+        }
+    }
+
+    private static Range<BigInteger> wholeRing(Partitioner partitioner)
+    {
+        return Range.openClosed(partitioner.minToken(), partitioner.maxToken());
+    }
+
+    /**
+     * Punches a real, non-empty gap into the sub-range at {@code gapIndex} by moving its lower endpoint up, so that
+     * the returned ranges leave exactly {@link #gapPunchedInto} uncovered.
+     */
+    private static List<Range<BigInteger>> withGapAt(List<Range<BigInteger>> gapFreeRanges, int gapIndex)
+    {
+        List<Range<BigInteger>> ranges = new ArrayList<>(gapFreeRanges);
+        Range<BigInteger> covered = ranges.get(gapIndex);
+        ranges.set(gapIndex, Range.openClosed(gapPunchedInto(covered).upperEndpoint(), covered.upperEndpoint()));
+        return ranges;
+    }
+
+    /**
+     * @return the sub-range that {@link #withGapAt} leaves uncovered when it punches a gap into {@code range}
+     */
+    private static Range<BigInteger> gapPunchedInto(Range<BigInteger> range)
+    {
+        return Range.openClosed(range.lowerEndpoint(), range.lowerEndpoint().add(BigInteger.TEN));
+    }
+
     private static void assertTokenRanges(int nodes, int replicationFactor, String[]... ranges)
     {
         assertThat(nodes).isEqualTo(ranges.length);