Fix scrubber falling into infinite loop

Fixes scrubber falling into infinite loop when the last partition is broken in data file and compression is enabled.

Patch by Jacek Lewandowski, reviewed by Brandon Williams, for CASSANDRA-17862

CHANGES.txt

@@ -1,4 +1,5 @@
 3.0.28
+ * Fix scrubber falling into infinite loop when the last partition is broken (CASSANDRA-17862)
  * Improve libjemalloc resolution in bin/cassandra (CASSANDRA-15767)
  * Fix missing state resetting on CompressedRandomAccessReader read errors (CASSANDRA-17314)
  * Fix restarting of services on gossipping-only member (CASSANDRA-17752)

src/java/org/apache/cassandra/db/compaction/Scrubber.java

@@ -258,7 +258,8 @@ public class Scrubber implements Closeable
 
                             outputHandler.warn("Retry failed too. Skipping to next row (retry's stacktrace follows)", th2);
                             badRows++;
-                            seekToNextRow();
+                            if (!seekToNextRow())
+                                break;
                         }
                     }
                     else
@@ -268,7 +269,8 @@ public class Scrubber implements Closeable
                         outputHandler.warn("Row starting at position " + dataStart + " is unreadable; skipping to next");
                         badRows++;
                         if (currentIndexKey != null)
-                            seekToNextRow();
+                            if (!seekToNextRow())
+                                break;
                     }
                 }
             }
@@ -390,14 +392,14 @@ public class Scrubber implements Closeable
         return indexFile != null && !indexFile.isEOF();
     }
 
-    private void seekToNextRow()
+    private boolean seekToNextRow()
     {
         while(nextRowPositionFromIndex < dataFile.length())
         {
             try
             {
                 dataFile.seek(nextRowPositionFromIndex);
-                return;
+                return true;
             }
             catch (Throwable th)
             {
@@ -408,6 +410,8 @@ public class Scrubber implements Closeable
 
             updateIndexKey();
         }
+
+        return false;
     }
 
     private void saveOutOfOrderRow(DecoratedKey prevKey, DecoratedKey key, UnfilteredRowIterator iterator)

test/unit/org/apache/cassandra/db/ScrubTest.java

@@ -23,11 +23,13 @@ import java.nio.ByteBuffer;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.nio.file.StandardOpenOption;
 import java.util.*;
 import java.util.concurrent.ExecutionException;
 
 import org.apache.cassandra.db.lifecycle.LifecycleNewTracker;
 import org.apache.commons.lang3.StringUtils;
+import org.junit.After;
 import org.junit.BeforeClass;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -43,7 +45,6 @@ import org.apache.cassandra.db.compaction.OperationType;
 import org.apache.cassandra.db.compaction.Scrubber;
 import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
 import org.apache.cassandra.db.marshal.Int32Type;
-import org.apache.cassandra.db.marshal.LongType;
 import org.apache.cassandra.db.marshal.UUIDType;
 import org.apache.cassandra.db.partitions.Partition;
 import org.apache.cassandra.db.partitions.PartitionUpdate;
@@ -62,6 +63,7 @@ import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.schema.KeyspaceParams;
 import org.apache.cassandra.utils.ByteBufferUtil;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
@@ -107,6 +109,12 @@ public class ScrubTest
                                     SchemaLoader.compositeIndexCFMD(KEYSPACE, CF_INDEX2_BYTEORDERED, true).copy(ByteOrderedPartitioner.instance));
     }
 
+    @After
+    public void after()
+    {
+        ColumnFamilyStore.getIfExists(KEYSPACE, CF).truncateBlockingWithoutSnapshot();
+    }
+
     @Test
     public void testScrubOneRow() throws ExecutionException, InterruptedException
     {
@@ -125,6 +133,26 @@ public class ScrubTest
         assertOrderedAll(cfs, 1);
     }
 
+    @Test
+    public void testScrubLastBrokenPartition() throws ExecutionException, InterruptedException, IOException
+    {
+        CompactionManager.instance.disableAutoCompaction();
+        ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, CF);
+
+        // insert data and verify we get it back w/ range query
+        fillCF(cfs, 1);
+        assertOrderedAll(cfs, 1);
+
+        Set<SSTableReader> liveSSTables = cfs.getLiveSSTables();
+        assertThat(liveSSTables).hasSize(1);
+        Files.write(Paths.get(liveSSTables.iterator().next().getFilename()), new byte[10], StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING);
+
+        CompactionManager.instance.performScrub(cfs, true, true, false, 2);
+
+        // check data is still there
+        assertOrderedAll(cfs, 0);
+    }
+
     @Test
     public void testScrubCorruptedCounterRow() throws IOException, WriteTimeoutException
     {