Fix HH to compact with correct gcBefore

patch by Alexey Zotov; reviewed by jbellis for CASSANDRA-4772

CHANGES.txt

@@ -1,4 +1,6 @@
 1.1.6
+ * Fix HH to compact with correct gcBefore, which avoids wiping out
+   undelivered hints (CASSANDRA-4772)
  * LCS will merge up to 32 L0 sstables as intended (CASSANDRA-4778)
  * NTS will default unconfigured DC replicas to zero (CASSANDRA-4675)
  * use default consistency level in counter validation if none is

src/java/org/apache/cassandra/db/HintedHandOffManager.java

@@ -29,6 +29,7 @@ import java.util.concurrent.*;
 import javax.management.MBeanServer;
 import javax.management.ObjectName;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableSortedSet;
 import org.apache.cassandra.io.sstable.Descriptor;
 import org.apache.cassandra.io.sstable.SSTable;
@@ -189,14 +190,15 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean
         StorageService.optionalTasks.execute(runnable);
     }
 
-    private Future<?> compact() throws ExecutionException, InterruptedException
+    @VisibleForTesting
+    protected Future<?> compact() throws ExecutionException, InterruptedException
     {
         final ColumnFamilyStore hintStore = Table.open(Table.SYSTEM_TABLE).getColumnFamilyStore(HINTS_CF);
         hintStore.forceBlockingFlush();
         ArrayList<Descriptor> descriptors = new ArrayList<Descriptor>();
         for (SSTable sstable : hintStore.getSSTables())
             descriptors.add(sstable.descriptor);
-        return CompactionManager.instance.submitUserDefined(hintStore, descriptors, Integer.MAX_VALUE);
+        return CompactionManager.instance.submitUserDefined(hintStore, descriptors, (int) (System.currentTimeMillis() / 1000));
     }
 
     private static boolean pagingFinished(ColumnFamily hintColumnFamily, ByteBuffer startColumn)

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

@@ -0,0 +1,65 @@
+package org.apache.cassandra.db;
+
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Test;
+
+import org.apache.cassandra.SchemaLoader;
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.HintedHandOffManager;
+import org.apache.cassandra.db.RowMutation;
+import org.apache.cassandra.db.Table;
+import org.apache.cassandra.db.compaction.CompactionManager;
+import org.apache.cassandra.db.compaction.SizeTieredCompactionStrategy;
+import org.apache.cassandra.db.filter.QueryPath;
+import org.apache.cassandra.dht.IPartitioner;
+import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.FBUtilities;
+
+import static junit.framework.Assert.assertEquals;
+
+public class HintedHandOffTest extends SchemaLoader
+{
+
+    public static final String TABLE4 = "Keyspace4";
+    public static final String STANDARD1_CF = "Standard1";
+    public static final String COLUMN1 = "column1";
+
+    // Test compaction of hints column family. It shouldn't remove all columns on compaction.
+    @Test
+    public void testCompactionOfHintsCF() throws Exception
+    {
+        // prepare hints column family
+        Table systemTable = Table.open("system");
+        ColumnFamilyStore hintStore = systemTable.getColumnFamilyStore(HintedHandOffManager.HINTS_CF);
+        hintStore.clearUnsafe();
+        hintStore.metadata.gcGraceSeconds(36000); // 10 hours
+        hintStore.setCompactionStrategyClass(SizeTieredCompactionStrategy.class.getCanonicalName());
+        hintStore.disableAutoCompaction();
+
+        // insert 1 hint
+        RowMutation rm = new RowMutation(TABLE4, ByteBufferUtil.bytes(1));
+        rm.add(new QueryPath(STANDARD1_CF,
+                             null,
+                             ByteBufferUtil.bytes(String.valueOf(COLUMN1))),
+               ByteBufferUtil.EMPTY_BYTE_BUFFER,
+               System.currentTimeMillis());
+
+        RowMutation.hintFor(rm, ByteBufferUtil.bytes("foo")).apply();
+
+        // flush data to disk
+        hintStore.forceBlockingFlush();
+        assertEquals(1, hintStore.getSSTables().size());
+
+        // submit compaction
+        FBUtilities.waitOnFuture(HintedHandOffManager.instance.compact());
+        while (CompactionManager.instance.getPendingTasks() > 0 || CompactionManager.instance.getActiveCompactions() > 0)
+            TimeUnit.SECONDS.sleep(1);
+
+        // single row should not be removed because of gc_grace_seconds
+        // is 10 hours and there are no any tombstones in sstable
+        assertEquals(1, hintStore.getSSTables().size());
+    }
+}