CASSANDRA-21273 - Add operation type and id to SSTable deletion log m…

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

@@ -386,7 +386,7 @@ private void removeCopiedSSTables(Set<MovedSSTable> movedSSTables)
         logger.debug("Removing copied SSTables which were left in data directories after failed SSTable import.");
         for (MovedSSTable movedSSTable : movedSSTables)
         {
-            // no logging here as for moveSSTablesBack case above as logging is done in delete method
+            logger.info("Deleting sstable: {}, operation type: SSTable import cleanup", movedSSTable.newDescriptor);
             movedSSTable.newDescriptor.getFormat().delete(movedSSTable.newDescriptor);
         }
     }

src/java/org/apache/cassandra/db/lifecycle/LogTransaction.java

@@ -379,6 +379,8 @@ public static class SSTableTidier implements Runnable
         private final Object lock;
         private final Ref<LogTransaction> parentRef;
         private final Counter totalDiskSpaceUsed;
+        private final OperationType operationType;
+        private final TimeUUID operationId;
 
         public SSTableTidier(SSTableReader referent, boolean wasNew, LogTransaction parent)
         {
@@ -387,6 +389,8 @@ public SSTableTidier(SSTableReader referent, boolean wasNew, LogTransaction pare
             this.wasNew = wasNew;
             this.lock = parent.lock;
             this.parentRef = parent.selfRef.tryRef();
+            this.operationType = parent.type();
+            this.operationId = parent.id();
 
             if (this.parentRef == null)
                 throw new IllegalStateException("Transaction already completed");
@@ -417,6 +421,7 @@ public void run()
                     if (!desc.fileFor(Components.DATA).exists() && !wasNew)
                         logger.error("SSTableTidier ran with no existing data file for an sstable that was not new");
 
+                    logger.info("Deleting sstable: {}, operation type: {}, id: {}", desc, operationType, operationId);
                     desc.getFormat().delete(desc);
                 }
                 catch (Throwable t)

src/java/org/apache/cassandra/io/sstable/format/big/BigFormat.java

@@ -321,7 +321,7 @@ public void deleteOrphanedComponents(Descriptor descriptor, Set<Component> compo
 
     private void delete(Descriptor desc, List<Component> components)
     {
-        logger.info("Deleting sstable: {}", desc);
+        logger.trace("Deleting sstable: {}", desc);
 
         if (components.remove(DATA))
             components.add(0, DATA); // DATA component should be first

src/java/org/apache/cassandra/io/sstable/format/bti/BtiFormat.java

@@ -214,7 +214,7 @@ public void deleteOrphanedComponents(Descriptor descriptor, Set<Component> compo
 
     private void delete(Descriptor desc, List<Component> components)
     {
-        logger.info("Deleting sstable: {}", desc);
+        logger.trace("Deleting sstable: {}", desc);
 
         if (components.remove(SSTableFormat.Components.DATA))
             components.add(0, SSTableFormat.Components.DATA); // DATA component should be first

test/unit/org/apache/cassandra/db/lifecycle/LogTransactionTest.java

@@ -39,6 +39,7 @@
 import com.google.common.collect.Sets;
 
 import org.junit.Test;
+import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.Util;
 import org.apache.cassandra.config.DatabaseDescriptor;
@@ -75,6 +76,9 @@
 import org.apache.cassandra.utils.concurrent.AbstractTransactionalTest;
 import org.apache.cassandra.utils.concurrent.Transactional;
 
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.ListAppender;
+
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
@@ -341,6 +345,45 @@ public void testCommitOnlyOld() throws Throwable
         assertFiles(dataFolder.path(), new HashSet<>());
     }
 
+    @Test
+    public void testDeletionLogContainsOperationContext() throws Throwable
+    {
+        ch.qos.logback.classic.Logger logger = (ch.qos.logback.classic.Logger) LoggerFactory.getLogger(LogTransaction.class);
+        ListAppender<ILoggingEvent> listAppender = new ListAppender<>();
+        try
+        {
+            listAppender.start();
+            logger.addAppender(listAppender);
+
+            ColumnFamilyStore cfs = MockSchema.newCFS(KEYSPACE);
+            File dataFolder = new Directories(cfs.metadata()).getDirectoryForNewSSTables();
+            SSTableReader sstable = sstable(dataFolder, cfs, 0, 128);
+
+            LogTransaction log = new LogTransaction(OperationType.COMPACTION);
+
+            LogTransaction.SSTableTidier tidier = log.obsoleted(sstable);
+            assertNotNull(tidier);
+
+            String id = log.id().toString();
+
+            log.finish();
+            sstable.markObsolete(tidier);
+            sstable.selfRef().release();
+
+            LogTransaction.waitForDeletions();
+
+            boolean found = listAppender.list.stream()
+                                             .anyMatch(e -> e.getFormattedMessage().contains("Deleting sstable:")
+                                                            && e.getFormattedMessage().contains("operation type: Compaction")
+                                                            && e.getFormattedMessage().contains(id));
+            assertTrue(found);
+        }
+        finally
+        {
+            logger.detachAppender(listAppender);
+        }
+    }
+
     @Test
     public void testCommitMultipleFolders() throws Throwable
     {