CASSANDRA-21134: Fix O_DIRECT short read on early-open of compaction SSTables

openFinalEarly() publishes a reader right after sync() but before commit.
Under O_DIRECT the inherited sync flushes only whole blocks, so the last
chunk's sub-block tail stays buffered and the early reader short-reads past
EOF (CorruptBlockException). Override syncInternal() to also flush the
buffered tail, restoring the channel and buffer cursors afterwards so the
authoritative pad+truncate at commit still produces a byte-identical file.

Author: samueldlightfoot

src/java/org/apache/cassandra/io/compress/DirectCompressedSequentialWriter.java

@@ -148,6 +148,17 @@ protected void seekToChunkStart()
         // resetAndTruncate (the parent's reason for this seek) is unsupported under DIO.
     }
 
+    // openFinalEarly() publishes a reader right after sync(), before commit. The inherited sync flushes
+    // only whole blocks, so under O_DIRECT the last chunk's sub-block tail is still buffered and that
+    // reader short-reads past EOF — flush the tail here too.
+    @Override
+    protected void syncInternal()
+    {
+        doFlush(0);
+        flushBufferedTailForEarlyOpen();
+        syncDataOnlyInternal();
+    }
+
     @Override
     protected void writeChunk(ByteBuffer toWrite)
     {
@@ -229,6 +240,30 @@ private void flushCompleteBlocks()
         }
     }
 
+    // Restoring the channel and buffer cursors afterwards is what keeps the writer untouched, so the
+    // authoritative pad+truncate at commit still produces the identical file. The over-block padding is
+    // safe: reads are bounded by compressedFileLength and commit overwrites or truncates it away.
+    private void flushBufferedTailForEarlyOpen()
+    {
+        int logicalPos = writeBuffer.position();
+        if (logicalPos == 0)
+            return;
+
+        try
+        {
+            long resumeChannelPos = fchannel.position();
+            writeAlignedBlocks(BitUtil.align(logicalPos, blockSize));
+
+            fchannel.position(resumeChannelPos);
+            writeBuffer.limit(writeBuffer.capacity());
+            writeBuffer.position(logicalPos);
+        }
+        catch (IOException e)
+        {
+            throw new FSWriteError(e, getPath());
+        }
+    }
+
     private void flushFinalWithPadding()
     {
         int logicalPos = writeBuffer.position();

test/unit/org/apache/cassandra/io/compress/DirectCompressedSequentialWriterTest.java

@@ -810,6 +810,63 @@ public void testOnDiskFilePointerTracksActualDataSize() throws IOException
         }
     }
 
+    /**
+     * Covers the early-open read path: openFinalEarly() publishes a reader after sync() but before
+     * commit, and under O_DIRECT the last chunk's sub-block tail is still buffered then, so without the
+     * sync() tail-flush the reader short-reads past EOF. The other tests miss this window — they all read
+     * post-finish, once the tail is flushed. Tiny sub-block outputs (e.g. system/IndexInfo) are the prod shape.
+     */
+    @Test
+    public void testEarlyOpenAfterSyncReadsBackData() throws IOException
+    {
+        for (int size : new int[]{ 1, 64, 298, 1024, 4096, DEFAULT_CHUNK_LENGTH * 3 + 137 })
+            assertEarlyOpenReadsBack(compressible(size), CompressionParams.lz4());
+    }
+
+    // Compresses well below one block, so after sync() the whole chunk+CRC is still buffered and the
+    // on-disk file is empty — the worst case for the early-open read.
+    private static byte[] compressible(int size)
+    {
+        byte[] data = new byte[size];
+        for (int i = 0; i < size; i++)
+            data[i] = (byte) (i % 7);
+        return data;
+    }
+
+    private static void assertEarlyOpenReadsBack(byte[] payload, CompressionParams params) throws IOException
+    {
+        File dataFile = FileUtils.createTempFile("early_open_direct", ".db");
+        File metadataFile = new File(dataFile.absolutePath() + ".metadata");
+        try
+        {
+            MetadataCollector collector = newCollector();
+            try (DirectCompressedSequentialWriter writer = new DirectCompressedSequentialWriter(
+                dataFile, metadataFile, null, SequentialWriterOption.DEFAULT, params, collector, null))
+            {
+                writer.write(payload);
+
+                // Mirror openFinalEarly: sync(), then open(0) == openDataFile(NO_LENGTH_OVERRIDE).
+                writer.sync();
+                try (CompressionMetadata md = writer.open(0);
+                     FileHandle fh = new FileHandle.Builder(dataFile).withCompressionMetadata(md).complete();
+                     RandomAccessReader reader = fh.createReader())
+                {
+                    assertEquals("early-open length mismatch, size=" + payload.length, payload.length, reader.length());
+                    byte[] readBack = new byte[payload.length];
+                    reader.readFully(readBack);
+                    assertArrayEquals("early-open read-back mismatch, size=" + payload.length, payload, readBack);
+                }
+
+                writer.finish();
+            }
+        }
+        finally
+        {
+            dataFile.tryDelete();
+            metadataFile.tryDelete();
+        }
+    }
+
     private void testWriteAndRead(String testName, int dataSize, CompressionParams params) throws IOException
     {
         File dataFile = FileUtils.createTempFile(testName + "_direct", ".db");