CNDB-14950 follow up for SequentialWriter

Author: driftx

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

@@ -92,7 +92,7 @@ public EncryptedSequentialWriter(File file,
                                      SequentialWriterOption option,
                                      ICompressor encryptor)
     {
-        super(file, SequentialWriterOption.newBuilder()
+        super(file, true, SequentialWriterOption.newBuilder()
                             .bufferSize(maxBytesInPage(encryptor))
                             .bufferType(BufferType.preferredForCompression())
                             .finishOnClose(option.finishOnClose())

src/java/org/apache/cassandra/io/util/SequentialWriter.java

@@ -105,16 +105,25 @@ protected Throwable doAbort(Throwable accumulate)
 
     // TODO: we should specify as a parameter if we permit an existing file or not
     private static FileChannel openChannel(File file)
+    {
+        return openChannel(file, true);
+    }
+
+    private static FileChannel openChannel(File file, boolean readable)
     {
         try
         {
             if (file.exists())
             {
-                return FileChannel.open(file.toPath(), StandardOpenOption.READ, StandardOpenOption.WRITE);
+                StandardOpenOption[] options = readable ? new StandardOpenOption[] {StandardOpenOption.WRITE, StandardOpenOption.READ}
+                                                        : new StandardOpenOption[] {StandardOpenOption.WRITE};
+                return FileChannel.open(file.toPath(), options);
             }
             else
             {
-                FileChannel channel = FileChannel.open(file.toPath(), StandardOpenOption.READ, StandardOpenOption.WRITE, StandardOpenOption.CREATE_NEW);
+                StandardOpenOption[] options = readable ? new StandardOpenOption[] {StandardOpenOption.WRITE, StandardOpenOption.READ, StandardOpenOption.CREATE_NEW}
+                                                        : new StandardOpenOption[] {StandardOpenOption.WRITE, StandardOpenOption.CREATE_NEW};
+                FileChannel channel = FileChannel.open(file.toPath(), options);
                 try
                 {
                     SyncUtil.trySyncDir(file.parent());
@@ -162,7 +171,12 @@ public SequentialWriter(File file, SequentialWriterOption option)
      */
     public SequentialWriter(File file, SequentialWriterOption option, boolean strictFlushing)
     {
-        super(openChannel(file), option.allocateBuffer());
+        this(file, true, option, strictFlushing);
+    }
+
+    public SequentialWriter(File file, boolean readable, SequentialWriterOption option, boolean strictFlushing)
+    {
+        super(openChannel(file, readable), option.allocateBuffer());
         this.strictFlushing = strictFlushing;
         this.fchannel = (FileChannel)channel;
 
@@ -295,6 +309,29 @@ public long paddedPosition()
         return PageAware.padded(position());
     }
 
+    public void updateFileHandle(FileHandle.Builder fhBuilder)
+    {
+        updateFileHandle(fhBuilder, -1);
+    }
+
+    public void updateFileHandle(FileHandle.Builder fhBuilder, long dataLength)
+    {
+        // Set actual length to avoid having to read it off the file system.
+        fhBuilder.withLengthOverride(dataLength > 0 ? dataLength : lastFlushOffset);
+    }
+
+    /**
+     * Some writers cannot feasibly calculate the exact length of a file. If any user needs to be able to store
+     * metadata at the end, they should use this function to ensure the content to be written can be addressed
+     * using `fileLength - bytesNeeded`.
+     *
+     * See PartitionIndexBuilder#complete and PartitionIndex#load for usage example.
+     */
+    public void establishEndAddressablePosition(int bytesNeeded) throws IOException
+    {
+        // Nothing to do when file length can be exactly determined.
+    }
+
     /**
      * Returns the current file pointer of the underlying on-disk file.
      * Note that since write works by buffering data, the value of this will increase by buffer

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

@@ -0,0 +1,123 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.cassandra.io.compress;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.file.StandardOpenOption;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ThreadLocalRandom;
+
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.io.util.File;
+import org.apache.cassandra.io.util.FileUtils;
+import org.apache.cassandra.io.util.SequentialWriterOption;
+import org.apache.cassandra.schema.CompressionParams;
+
+import static org.junit.Assert.*;
+
+public class EncryptedSequentialWriterReadableTest
+{
+    static final ICompressor AESEncryptor;
+    static final CompressionParams AESEncryptorParams;
+    
+    static
+    {
+        DatabaseDescriptor.daemonInitialization();
+        
+        Map<String, String> opts = new HashMap<>();
+        opts.put(CompressionParams.CLASS, Encryptor.class.getName());
+        opts.put(EncryptionConfig.CIPHER_ALGORITHM, "AES/CBC/PKCS5Padding");
+        opts.put(EncryptionConfig.SECRET_KEY_STRENGTH, Integer.toString(128));
+        opts.put(EncryptionConfig.KEY_PROVIDER, EncryptorTest.KeyProviderFactoryStub.class.getName());
+        AESEncryptor = Encryptor.create(opts);
+        AESEncryptorParams = CompressionParams.fromMap(opts);
+    }
+    
+    @BeforeClass
+    public static void setupDD()
+    {
+        // Already initialized in static block
+    }
+
+    @Test
+    public void testEncryptedWriterWithReadableChannel() throws IOException
+    {
+        File tempFile = FileUtils.createTempFile("encrypted", "test");
+        tempFile.tryDelete();
+        
+        byte[] testData = new byte[8192];
+        ThreadLocalRandom.current().nextBytes(testData);
+        
+        ICompressor encryptor = AESEncryptor;
+        SequentialWriterOption option = SequentialWriterOption.newBuilder()
+                                                              .bufferSize(1024)
+                                                              .build();
+        
+        try (EncryptedSequentialWriter writer = new EncryptedSequentialWriter(tempFile, option, encryptor))
+        {
+            writer.write(testData);
+            writer.finish();
+            
+            assertTrue(tempFile.exists());
+            assertTrue(tempFile.length() > 0);
+            
+            try (FileChannel channel = FileChannel.open(tempFile.toPath(), StandardOpenOption.READ))
+            {
+                ByteBuffer headerBuffer = ByteBuffer.allocate(128);
+                int bytesRead = channel.read(headerBuffer);
+                assertTrue(bytesRead > 0);
+            }
+        }
+        finally
+        {
+            tempFile.tryDelete();
+        }
+    }
+
+    @Test
+    public void testEncryptedWriterConstructorPassesReadableTrue() throws IOException
+    {
+        File tempFile = FileUtils.createTempFile("encrypted2", "test");
+        tempFile.tryDelete();
+        
+        ICompressor encryptor = AESEncryptor;
+        SequentialWriterOption option = SequentialWriterOption.newBuilder()
+                                                              .bufferSize(1024)
+                                                              .finishOnClose(true)
+                                                              .build();
+        
+        byte[] testData = "test data for encryption".getBytes();
+        
+        try (EncryptedSequentialWriter writer = new EncryptedSequentialWriter(tempFile, option, encryptor))
+        {
+            writer.write(testData);
+        }
+        
+        assertTrue(tempFile.exists());
+        assertTrue(tempFile.length() > testData.length);
+        
+        tempFile.tryDelete();
+    }
+
+}