https://issues.apache.org/jira/browse/AMQ-6203

Expose configuration on the persistence adapter level.
Double the test timeout and increase the compaction frequency to account
for very slow CI boxes.

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

@@ -636,6 +636,39 @@ public class KahaDBPersistenceAdapter extends LockableServiceSupport implements
         return letter.isEnableIndexPageCaching();
     }
 
+    public int getCompactAcksAfterNoGC() {
+        return letter.getCompactAcksAfterNoGC();
+    }
+
+    /**
+     * Sets the number of GC cycles where no journal logs were removed before an attempt to
+     * move forward all the acks in the last log that contains them and is otherwise unreferenced.
+     * <p>
+     * A value of -1 will disable this feature.
+     *
+     * @param compactAcksAfterNoGC
+     *      Number of empty GC cycles before we rewrite old ACKS.
+     */
+    public void setCompactAcksAfterNoGC(int compactAcksAfterNoGC) {
+        this.letter.setCompactAcksAfterNoGC(compactAcksAfterNoGC);
+    }
+
+    public boolean isCompactAcksIgnoresStoreGrowth() {
+        return this.letter.isCompactAcksIgnoresStoreGrowth();
+    }
+
+    /**
+     * Configure if Ack compaction will occur regardless of continued growth of the
+     * journal logs meaning that the store has not run out of space yet.  Because the
+     * compaction operation can be costly this value is defaulted to off and the Ack
+     * compaction is only done when it seems that the store cannot grow and larger.
+     *
+     * @param compactAcksIgnoresStoreGrowth the compactAcksIgnoresStoreGrowth to set
+     */
+    public void setCompactAcksIgnoresStoreGrowth(boolean compactAcksIgnoresStoreGrowth) {
+        this.letter.setCompactAcksIgnoresStoreGrowth(compactAcksIgnoresStoreGrowth);
+    }
+
     public KahaDBStore getStore() {
         return letter;
     }

activemq-unit-tests/src/test/java/org/apache/activemq/bugs/TransactedStoreUsageSuspendResumeTest.java

@@ -117,6 +117,7 @@ public class TransactedStoreUsageSuspendResumeTest {
         KahaDBPersistenceAdapter kahaDB = new KahaDBPersistenceAdapter();
         kahaDB.setJournalMaxFileLength(256 * 1024);
         kahaDB.setCleanupInterval(10*1000);
+        kahaDB.setCompactAcksAfterNoGC(5);
         broker.setPersistenceAdapter(kahaDB);
 
         broker.getSystemUsage().getStoreUsage().setLimit(7*1024*1024);
@@ -146,7 +147,7 @@ public class TransactedStoreUsageSuspendResumeTest {
             }
         });
         sendExecutor.shutdown();
-        sendExecutor.awaitTermination(5, TimeUnit.MINUTES);
+        sendExecutor.awaitTermination(10, TimeUnit.MINUTES);
 
         boolean allMessagesReceived = messagesReceivedCountDown.await(5, TimeUnit.MINUTES);
         if (!allMessagesReceived) {
@@ -160,7 +161,7 @@ public class TransactedStoreUsageSuspendResumeTest {
         assertTrue("Got all messages: " + messagesReceivedCountDown, allMessagesReceived);
 
         // give consumers a chance to exit gracefully
-        TimeUnit.SECONDS.sleep(2);
+        TimeUnit.SECONDS.sleep(5);
     }
 
     private void sendMessages() throws Exception {