AMQ-7084 - ensure allocated and unused free pages are visible to subsequent transactions, fix and test with test updates to reflect proper usage

activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/disk/page/Transaction.java

@@ -670,6 +670,8 @@ public class Transaction implements Iterable<Page> {
             allocateList.clear();
             writes.clear();
             writeTransactionId = -1;
+        } else {
+            freePages(allocateList);
         }
         size = 0;
     }
@@ -692,6 +694,8 @@ public class Transaction implements Iterable<Page> {
             allocateList.clear();
             writes.clear();
             writeTransactionId = -1;
+        } else {
+            freePages(allocateList);
         }
         size = 0;
     }

activemq-kahadb-store/src/test/java/org/apache/activemq/store/kahadb/disk/index/BTreeIndexTest.java

@@ -57,11 +57,12 @@ public class BTreeIndexTest extends IndexTestSupport {
     protected Index<String, Long> createIndex() throws Exception {
 
         long id = tx.allocate().getPageId();
-        tx.commit();
 
         BTreeIndex<String, Long> index = new BTreeIndex<String,Long>(pf, id);
         index.setKeyMarshaller(StringMarshaller.INSTANCE);
         index.setValueMarshaller(LongMarshaller.INSTANCE);
+        index.load(tx);
+        tx.commit();
 
         return index;
     }
@@ -231,8 +232,6 @@ public class BTreeIndexTest extends IndexTestSupport {
         this.index.load(tx);
 
         long id = tx.allocate().getPageId();
-        tx.commit();
-
         BTreeIndex<String, String> sindex = new BTreeIndex<String,String>(pf, id);
         sindex.setKeyMarshaller(StringMarshaller.INSTANCE);
         sindex.setValueMarshaller(StringMarshaller.INSTANCE);
@@ -273,7 +272,6 @@ public class BTreeIndexTest extends IndexTestSupport {
         this.index.load(tx);
 
         long id = tx.allocate().getPageId();
-        tx.commit();
 
         BTreeIndex<String, String> sindex = new BTreeIndex<String,String>(pf, id);
         sindex.setKeyMarshaller(StringMarshaller.INSTANCE);
@@ -364,7 +362,6 @@ public class BTreeIndexTest extends IndexTestSupport {
         pf.load();
         tx = pf.tx();
         long id = tx.allocate().getPageId();
-        tx.commit();
 
         BTreeIndex<Long, HashSet<String>> test = new BTreeIndex<Long, HashSet<String>>(pf, id);
         test.setKeyMarshaller(LongMarshaller.INSTANCE);

activemq-kahadb-store/src/test/java/org/apache/activemq/store/kahadb/disk/index/HashIndexBenchMark.java

@@ -27,11 +27,12 @@ public class HashIndexBenchMark extends IndexBenchmark {
 
         Transaction tx = pf.tx();
         long id = tx.allocate().getPageId();
-        tx.commit();
 
         HashIndex<String, Long> index = new HashIndex<String, Long>(pf, id);
         index.setKeyMarshaller(StringMarshaller.INSTANCE);
         index.setValueMarshaller(LongMarshaller.INSTANCE);
+        index.load(tx);
+        tx.commit();
 
         return index;
     }

activemq-kahadb-store/src/test/java/org/apache/activemq/store/kahadb/disk/index/HashIndexTest.java

@@ -25,13 +25,12 @@ public class HashIndexTest extends IndexTestSupport {
     protected Index<String, Long> createIndex() throws Exception {
         
         long id = tx.allocate().getPageId();
-        tx.commit();
-
         HashIndex<String, Long> index = new HashIndex<String,Long>(pf, id);
         index.setBinCapacity(12);
         index.setKeyMarshaller(StringMarshaller.INSTANCE);
         index.setValueMarshaller(LongMarshaller.INSTANCE);
-        
+        index.load(tx);
+        tx.commit();
         return index;
     }
 

activemq-kahadb-store/src/test/java/org/apache/activemq/store/kahadb/disk/index/ListIndexTest.java

@@ -57,12 +57,11 @@ public class ListIndexTest extends IndexTestSupport {
     protected Index<String, Long> createIndex() throws Exception {
 
         long id = tx.allocate().getPageId();
-        tx.commit();
-
         ListIndex<String, Long> index = new ListIndex<String, Long>(pf, id);
         index.setKeyMarshaller(StringMarshaller.INSTANCE);
         index.setValueMarshaller(LongMarshaller.INSTANCE);
-
+        index.load(tx);
+        tx.commit();
         return index;
     }
 

activemq-kahadb-store/src/test/java/org/apache/activemq/store/kahadb/disk/page/PageFileTest.java

@@ -136,7 +136,6 @@ public class PageFileTest extends TestCase {
 
         Transaction tx = pf.tx();
         Page page = tx.allocate();
-        tx.commit();
 
         OutputStream pos = tx.openOutputStream(page, true);
         DataOutputStream os = new DataOutputStream(pos);
@@ -240,4 +239,26 @@ public class PageFileTest extends TestCase {
             pf2.unload();
         }
     }
+
+    public void testAllocatedAndUnusedAreFree() throws Exception {
+
+        PageFile pf = new PageFile(new File("target/test-data"), getName());
+        pf.delete();
+        pf.load();
+
+        Transaction tx = pf.tx();
+        tx.allocate(10);
+        tx.commit();
+
+        assertEquals(10, pf.getPageCount());
+        assertEquals(pf.getFreePageCount(), 10);
+
+        // free pages should get reused
+
+        tx.allocate(10);
+        tx.rollback();
+        assertEquals(10, pf.getPageCount());
+        assertEquals(pf.getFreePageCount(), 10);
+
+    }
 }