From a78db628b0bcae6b1d30829b7510093ec4bca0ef Mon Sep 17 00:00:00 2001
From: Jacek Lewandowski <lewandowski.jacek@gmail.com>
Date: Thu, 25 Aug 2022 20:38:45 +0200
Subject: [PATCH] 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 +
 .../cassandra/db/compaction/Scrubber.java     | 12 +++++---
 .../org/apache/cassandra/db/ScrubTest.java    | 30 ++++++++++++++++++-
 3 files changed, 38 insertions(+), 5 deletions(-)

diff --git a/CHANGES.txt b/CHANGES.txt
index d7bf893397..d070392755 100644
--- a/CHANGES.txt
+++ b/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)
diff --git a/src/java/org/apache/cassandra/db/compaction/Scrubber.java b/src/java/org/apache/cassandra/db/compaction/Scrubber.java
index 463cc9cd34..b42093f2f2 100644
--- a/src/java/org/apache/cassandra/db/compaction/Scrubber.java
+++ b/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)
diff --git a/test/unit/org/apache/cassandra/db/ScrubTest.java b/test/unit/org/apache/cassandra/db/ScrubTest.java
index e8a3285c79..56e42c66b5 100644
--- a/test/unit/org/apache/cassandra/db/ScrubTest.java
+++ b/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
     {