[AMQ-9366] Add DataFileFactroy to KahaDBPersistenceAdapter and update test to remove SecurityManager

Co-authored-by: Gurpartap Singh <gurpartap0306>

Author: mattrpav

activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBPersistenceAdapter.java

@@ -44,6 +44,7 @@
 import org.apache.activemq.store.kahadb.data.KahaLocalTransactionId;
 import org.apache.activemq.store.kahadb.data.KahaTransactionInfo;
 import org.apache.activemq.store.kahadb.data.KahaXATransactionId;
+import org.apache.activemq.store.kahadb.disk.journal.DataFileFactory;
 import org.apache.activemq.store.kahadb.disk.journal.Journal.JournalDiskSyncStrategy;
 import org.apache.activemq.usage.SystemUsage;
 import org.apache.activemq.util.ServiceStopper;
@@ -840,4 +841,8 @@ public void setCleanupOnStop(boolean cleanupOnStop) {
     public boolean getCleanupOnStop() {
         return this.letter.getCleanupOnStop();
     }
+
+    public void setDataFileFactory(DataFileFactory dataFileFactory) {
+        this.letter.setDataFileFactory(dataFileFactory);
+    }
 }

activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java

@@ -92,11 +92,8 @@
 import org.apache.activemq.store.kahadb.disk.index.BTreeIndex;
 import org.apache.activemq.store.kahadb.disk.index.BTreeVisitor;
 import org.apache.activemq.store.kahadb.disk.index.ListIndex;
-import org.apache.activemq.store.kahadb.disk.journal.DataFile;
-import org.apache.activemq.store.kahadb.disk.journal.Journal;
+import org.apache.activemq.store.kahadb.disk.journal.*;
 import org.apache.activemq.store.kahadb.disk.journal.Journal.JournalDiskSyncStrategy;
-import org.apache.activemq.store.kahadb.disk.journal.Location;
-import org.apache.activemq.store.kahadb.disk.journal.TargetedDataFileAppender;
 import org.apache.activemq.store.kahadb.disk.page.Page;
 import org.apache.activemq.store.kahadb.disk.page.PageFile;
 import org.apache.activemq.store.kahadb.disk.page.Transaction;
@@ -266,6 +263,7 @@ public enum PurgeRecoveredXATransactionStrategy {
 
     protected JournalDiskSyncStrategy journalDiskSyncStrategy = JournalDiskSyncStrategy.ALWAYS;
     protected boolean archiveDataLogs;
+    protected DataFileFactory dataFileFactory;
     protected File directoryArchive;
     protected AtomicLong journalSize = new AtomicLong(0);
     long journalDiskSyncInterval = 1000;
@@ -3421,6 +3419,9 @@ protected Journal createJournal() throws IOException {
             IOHelper.mkdirs(getDirectoryArchive());
             manager.setDirectoryArchive(getDirectoryArchive());
         }
+        if (getDataFileFactory() != null) {
+            manager.setDataFileFactory(getDataFileFactory());
+        }
         return manager;
     }
 
@@ -4297,4 +4298,12 @@ private void handleIOException(String taskName, IOException ioe) {
             LOG.debug(e.getMessage(), e);
         }
     }
+
+    public DataFileFactory getDataFileFactory() {
+        return this.dataFileFactory;
+    }
+
+    public void  setDataFileFactory(DataFileFactory dataFileFactory) {
+        this.dataFileFactory = dataFileFactory;
+    }
 }

activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/disk/journal/DataFileFactory.java

@@ -0,0 +1,23 @@
+/**
+ * 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.activemq.store.kahadb.disk.journal;
+
+import java.io.File;
+
+public interface DataFileFactory {
+    DataFile create(File file, int number);
+}

activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/disk/journal/DefaultDataFileFactory.java

@@ -0,0 +1,27 @@
+/**
+ * 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.activemq.store.kahadb.disk.journal;
+
+import java.io.File;
+
+public class DefaultDataFileFactory implements DataFileFactory {
+
+    @Override
+    public DataFile create(File file, int number) {
+        return new DataFile(file, number);
+    }
+}

activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/disk/journal/Journal.java

@@ -241,6 +241,7 @@ private static byte[] createEofBatchAndLocationRecord() {
     private long cleanupInterval = DEFAULT_CLEANUP_INTERVAL;
 
     protected JournalDiskSyncStrategy journalDiskSyncStrategy = JournalDiskSyncStrategy.ALWAYS;
+    protected DataFileFactory dataFileFactory = new DefaultDataFileFactory();
 
     public interface DataFileRemovedListener {
         void fileRemoved(DataFile datafile);
@@ -272,7 +273,7 @@ public boolean accept(File dir, String n) {
                     String n = file.getName();
                     String numStr = n.substring(filePrefix.length(), n.length()-fileSuffix.length());
                     int num = Integer.parseInt(numStr);
-                    DataFile dataFile = new DataFile(file, num);
+                    DataFile dataFile = dataFileFactory.create(file, num);
                     fileMap.put(dataFile.getDataFileId(), dataFile);
                     totalLength.addAndGet(dataFile.getLength());
                 } catch (NumberFormatException e) {
@@ -687,7 +688,7 @@ public void run() {
     private DataFile newDataFile() throws IOException {
         int nextNum = nextDataFileId++;
         File file = getFile(nextNum);
-        DataFile nextWriteFile = new DataFile(file, nextNum);
+        DataFile nextWriteFile = dataFileFactory.create(file, nextNum);
         preallocateEntireJournalDataFile(nextWriteFile.appendRandomAccessFile());
         return nextWriteFile;
     }
@@ -697,7 +698,7 @@ public DataFile reserveDataFile() {
         synchronized (dataFileIdLock) {
             int nextNum = nextDataFileId++;
             File file = getFile(nextNum);
-            DataFile reservedDataFile = new DataFile(file, nextNum);
+            DataFile reservedDataFile = dataFileFactory.create(file, nextNum);
             synchronized (currentDataFile) {
                 fileMap.put(reservedDataFile.getDataFileId(), reservedDataFile);
                 fileByFileMap.put(file, reservedDataFile);
@@ -1164,6 +1165,14 @@ public void setDataFileRemovedListener(DataFileRemovedListener dataFileRemovedLi
         this.dataFileRemovedListener = dataFileRemovedListener;
     }
 
+    public void setDataFileFactory(DataFileFactory dataFileFactory) {
+        this.dataFileFactory = dataFileFactory;
+    }
+
+    public DataFileFactory getDataFileFactory() {
+        return  this.dataFileFactory;
+    }
+
     public static class WriteCommand extends LinkedNode<WriteCommand> {
         public final Location location;
         public final ByteSequence data;

activemq-kahadb-store/src/test/java/org/apache/activemq/store/kahadb/JournalArchiveTest.java

@@ -22,6 +22,9 @@
 import org.apache.activemq.broker.region.policy.PolicyMap;
 import org.apache.activemq.command.ActiveMQQueue;
 import org.apache.activemq.store.kahadb.disk.journal.DataFile;
+import org.apache.activemq.store.kahadb.disk.journal.DataFileFactory;
+import org.apache.activemq.store.kahadb.disk.journal.DefaultDataFileFactory;
+import org.apache.activemq.store.kahadb.disk.journal.ErrorDataFileFactory;
 import org.junit.After;
 import org.junit.Test;
 import org.slf4j.Logger;
@@ -54,6 +57,7 @@ public class JournalArchiveTest {
     private BrokerService broker = null;
     private final Destination destination = new ActiveMQQueue("Test");
     private KahaDBPersistenceAdapter adapter;
+    private DataFileFactory dataFileFactory;
 
     protected void startBroker() throws Exception {
         doStartBroker(true);
@@ -104,6 +108,7 @@ protected void configurePersistence(BrokerService brokerService) throws Exceptio
         adapter.setCheckForCorruptJournalFiles(true);
 
         adapter.setArchiveDataLogs(true);
+        adapter.setDataFileFactory(dataFileFactory);
     }
 
     @After
@@ -119,16 +124,8 @@ public void tearDown() throws Exception {
     public void testRecoveryOnArchiveFailure() throws Exception {
         final AtomicInteger atomicInteger = new AtomicInteger();
 
-        System.setSecurityManager(new SecurityManager() {
-            public void checkPermission(Permission perm) {}
-            public void checkPermission(Permission perm, Object context) {}
+        this.dataFileFactory = new ErrorDataFileFactory();
 
-            public void checkWrite(String file) {
-                if (file.contains(DEFAULT_ARCHIVE_DIRECTORY) && atomicInteger.incrementAndGet() > 4) {
-                    throw new SecurityException("No Perms to write to archive times:" + atomicInteger.get());
-                }
-            }
-        });
         startBroker();
 
         int sent = produceMessagesToConsumeMultipleDataFiles(50);
@@ -151,7 +148,7 @@ public void run() {
         assertTrue("broker got shutdown on page in error", gotShutdown.await(10, TimeUnit.SECONDS));
 
         // no restrictions
-        System.setSecurityManager(null);
+        this.dataFileFactory = new DefaultDataFileFactory();
 
         int numFilesAfterRestart = 0;
         try {

activemq-kahadb-store/src/test/java/org/apache/activemq/store/kahadb/disk/journal/ErrorDataFileFactory.java

@@ -0,0 +1,47 @@
+/**
+ * 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.activemq.store.kahadb.disk.journal;
+
+import org.apache.activemq.util.IOHelper;
+
+import java.io.File;
+import java.io.IOException;
+
+import static org.apache.activemq.store.kahadb.disk.journal.Journal.DEFAULT_ARCHIVE_DIRECTORY;
+
+public class ErrorDataFileFactory implements DataFileFactory {
+
+    @Override
+    public DataFile create(File file, int number) {
+        return new ErrorDataFile(file, number);
+    }
+
+    public static class ErrorDataFile extends DataFile {
+
+        ErrorDataFile(File file, int number) {
+            super(file, number);
+        }
+
+        @Override
+        public synchronized void move(File targetDirectory) throws IOException {
+            if (targetDirectory.getName().contains(DEFAULT_ARCHIVE_DIRECTORY) && this.dataFileId > 4) {
+                throw new SecurityException("No Perms to write to archive times:" + this.dataFileId);
+            }
+            IOHelper.moveFile(file, targetDirectory);
+        }
+    }
+}