CASSANDRA-21134: Address review feedback for background write Direct I/O

Fail fast on unsupported background_write_disk_access_mode values, drop
"scrub" from the DIO operation docs (it's UNSUPPORTED_CORRECTNESS), and
tighten AntiCompactionTest/CompactionsTest to assert Direct I/O actually
engages rather than passing spuriously.

Author: samueldlightfoot

conf/cassandra.yaml

@@ -694,7 +694,7 @@ commitlog_disk_access_mode: legacy
 # compaction_read_disk_access_mode: auto
 
 # Set the disk access mode for writing compressed SSTables during background operations
-# (compaction, streaming, scrub, cleanup, repair, etc.). The allowed values are:
+# (compaction, streaming, cleanup, repair, etc.). The allowed values are:
 # - standard: use buffered I/O (default)
 # - direct: use direct I/O, bypassing the OS page cache
 # Note: Only applies to compressed tables. Uncompressed tables always use buffered I/O.

src/java/org/apache/cassandra/config/Config.java

@@ -362,7 +362,7 @@ public MemtableOptions()
 
     public DataStorageSpec.IntKibibytesBound compressed_read_ahead_buffer_size = new DataStorageSpec.IntKibibytesBound("256KiB");
 
-    // Direct IO for background SSTable writes (compaction, streaming, scrub, cleanup, etc.)
+    // Direct IO for background SSTable writes (compaction, streaming, cleanup, etc.)
     // When 'direct' is set, background writes bypass the OS page cache using O_DIRECT.
     // Memtable flushes always use buffered I/O regardless of this setting.
     // Default is 'standard' (buffered I/O) - users must opt-in to Direct IO

src/java/org/apache/cassandra/config/DatabaseDescriptor.java

@@ -3481,6 +3481,11 @@ public static void initializeBackgroundWriteDiskAccessMode()
                 }
             }
         }
+        else if (providedMode != DiskAccessMode.standard)
+        {
+            throw new ConfigurationException("Unsupported disk access mode for background_write_disk_access_mode " +
+                                             "(options: standard/direct/auto): " + providedMode, false);
+        }
 
         backgroundWriteDiskAccessMode = providedMode;
     }

test/unit/org/apache/cassandra/db/compaction/AntiCompactionTest.java

@@ -265,6 +265,7 @@ public void testAntiCompactionWithCompressedTableAndDirectWrites() throws Throwa
             store.truncateBlocking();
 
             populateDeterministic(store, ts);
+            DirectIoTestUtils.activatedDirectWriteOperations().clear();
             SSTableStats directStats = DirectIoTestUtils.withDirectWrites(() ->
                 antiCompactRanges(store, atEndpoint(range(0, 4), NO_RANGES)));
             assertEquals(2, directStats.numLiveSSTables);

test/unit/org/apache/cassandra/db/compaction/CompactionsTest.java

@@ -86,6 +86,7 @@
 import org.apache.cassandra.io.sstable.format.SSTableFormat.Components;
 import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.io.sstable.metadata.StatsMetadata;
+import org.apache.cassandra.io.util.DirectIoTestUtils;
 import org.apache.cassandra.schema.CompactionParams;
 import org.apache.cassandra.schema.CompressionParams;
 import org.apache.cassandra.schema.KeyspaceParams;
@@ -491,8 +492,16 @@ public CompactionAwareWriter getCompactionAwareWriter(ColumnFamilyStore cfs,
                 return new MaxSSTableSizeWriter(cfs, directories, transaction, nonExpiredSSTables, maxSSTableSize, 0);
             }
         };
+        DirectIoTestUtils.activatedDirectWriteOperations().clear();
         task.execute(CompactionManager.instance.active);
 
+        if (backgroundWriteDiskAccessMode == DiskAccessMode.direct)
+        {
+            Set<OperationType> activated = DirectIoTestUtils.activatedDirectWriteOperations();
+            assertTrue("DirectCompressedSequentialWriter must have engaged for COMPACTION; activated=" + activated,
+                       activated.contains(OperationType.COMPACTION));
+        }
+
         Set<SSTableReader> result = cfs.getLiveSSTables();
         assertTrue("expected segment rotation to produce >= 2 SSTables, got " + result.size(), result.size() >= 2);