Apache/activemq
1
0
Fork 0
mirror of https://github.com/apache/activemq.git synced 2025-02-10 03:56:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

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

Browse Source 
Adding a flag to enable or disable Ack Compaction
Currently defaults to enabled for 5.14.0 and above

(cherry picked from commit cbad8babe507f63cc80e42fe37070b898d4d8dbc)
This commit is contained in:
Christopher L. Shannon (cshannon) 2016-04-12 13:30:10 +00:00
parent 72eecbe260
commit b8ac1b4c56
2 changed files with 38 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -669,6 +669,24 @@ public class KahaDBPersistenceAdapter extends LockableServiceSupport implements
this.letter.setCompactAcksIgnoresStoreGrowth(compactAcksIgnoresStoreGrowth); this.letter.setCompactAcksIgnoresStoreGrowth(compactAcksIgnoresStoreGrowth);
} }
/**
* Returns whether Ack compaction is enabled
*
* @return enableAckCompaction
*/
public boolean isEnableAckCompaction() {
return letter.isEnableAckCompaction();
}
/**
* Configure if the Ack compaction task should be enabled to run
*
* @param enableAckCompaction
*/
public void setEnableAckCompaction(boolean enableAckCompaction) {
letter.setEnableAckCompaction(enableAckCompaction);
}
public KahaDBStore getStore() { public KahaDBStore getStore() {
return letter; return letter;
} }

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

@ -274,6 +274,7 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
private boolean enableIndexPageCaching = true; private boolean enableIndexPageCaching = true;
ReentrantReadWriteLock checkpointLock = new ReentrantReadWriteLock(); ReentrantReadWriteLock checkpointLock = new ReentrantReadWriteLock();
private boolean enableAckCompaction = true;
private int compactAcksAfterNoGC = 10; private int compactAcksAfterNoGC = 10;
private boolean compactAcksIgnoresStoreGrowth = false; private boolean compactAcksIgnoresStoreGrowth = false;
private int checkPointCyclesWithNoGC; private int checkPointCyclesWithNoGC;
@ -1817,7 +1818,7 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
if (ackMessageFileMapMod) { if (ackMessageFileMapMod) {
checkpointUpdate(tx, false); checkpointUpdate(tx, false);
} }
} else { } else if (isEnableAckCompaction()) {
if (++checkPointCyclesWithNoGC >= getCompactAcksAfterNoGC()) { if (++checkPointCyclesWithNoGC >= getCompactAcksAfterNoGC()) {
// First check length of journal to make sure it makes sense to even try. // First check length of journal to make sure it makes sense to even try.
// //
@ -3671,4 +3672,22 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
public void setCompactAcksIgnoresStoreGrowth(boolean compactAcksIgnoresStoreGrowth) { public void setCompactAcksIgnoresStoreGrowth(boolean compactAcksIgnoresStoreGrowth) {
this.compactAcksIgnoresStoreGrowth = compactAcksIgnoresStoreGrowth; this.compactAcksIgnoresStoreGrowth = compactAcksIgnoresStoreGrowth;
} }
/**
* Returns whether Ack compaction is enabled
*
* @return enableAckCompaction
*/
public boolean isEnableAckCompaction() {
return enableAckCompaction;
}
/**
* Configure if the Ack compaction task should be enabled to run
*
* @param enableAckCompaction
*/
public void setEnableAckCompaction(boolean enableAckCompaction) {
this.enableAckCompaction = enableAckCompaction;
}
} }