Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

https://issues.apache.org/jira/browse/AMQ-6084 - add broker.adjustUsageLimits to disable the adjustment of limits to what is available. In this way, a broker will fail to start if constrained, ensuring it won't accept connections and block pending resources if it has earlier exited for that reason 

(cherry picked from commit d7febddb67)
This commit is contained in:
gtully 2015-12-10 11:23:56 +00:00 committed by Timothy Bish
parent 99fce5bae9
commit 2be754583c
3 changed files with 127 additions and 53 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

61
activemq-broker/src/main/java/org/apache/activemq/broker/BrokerService.java
View File

@ -243,6 +243,7 @@ public class BrokerService implements Service {
private int maxPurgedDestinationsPerSweep = 0;
private int schedulePeriodForDiskUsageCheck = 0;
private int diskUsageCheckRegrowThreshold = -1;
private boolean adjustUsageLimits = true;
private BrokerContext brokerContext;
private boolean networkConnectorStartAsync = false;
private boolean allowTempAutoCreationOnSend;
@ -591,6 +592,7 @@ public class BrokerService implements Service {
MDC.put("activemq.broker", brokerName);
try {
checkMemorySystemUsageLimits();
if (systemExitOnShutdown && useShutdownHook) {
throw new ConfigurationException("'useShutdownHook' property cannot be be used with 'systemExitOnShutdown', please turn it off (useShutdownHook=false)");
}
@ -740,7 +742,7 @@ public class BrokerService implements Service {
LOG.info("For help or more information please see: http://activemq.apache.org");
getBroker().brokerServiceStarted();
checkSystemUsageLimits();
checkStoreSystemUsageLimits();
startedLatch.countDown();
getBroker().nowMasterBroker();
}
@ -1973,7 +1975,7 @@ public class BrokerService implements Service {
* Check that the store usage limit is not greater than max usable
* space and adjust if it is
*/
protected void checkStoreUsageLimits() throws IOException {
protected void checkStoreUsageLimits() throws Exception {
final SystemUsage usage = getSystemUsage();
if (getPersistenceAdapter() != null) {
@ -2001,7 +2003,7 @@ public class BrokerService implements Service {
* Check that temporary usage limit is not greater than max usable
* space and adjust if it is
*/
protected void checkTmpStoreUsageLimits() throws IOException {
protected void checkTmpStoreUsageLimits() throws Exception {
final SystemUsage usage = getSystemUsage();
File tmpDir = getTmpDataDirectory();
@ -2030,7 +2032,7 @@ public class BrokerService implements Service {
}
}
protected void checkUsageLimit(File dir, Usage<?> storeUsage, int percentLimit) {
protected void checkUsageLimit(File dir, Usage<?> storeUsage, int percentLimit) throws ConfigurationException {
if (dir != null) {
dir = StoreUtil.findParentDirectory(dir);
String storeName = storeUsage instanceof StoreUsage ? "Store" : "Temporary Store";
@ -2064,6 +2066,17 @@ public class BrokerService implements Service {
//check if the limit is too large for the amount of usable space
} else if (storeLimit > totalUsableSpace) {
final String message = storeName + " limit is " + storeLimit / oneMeg
+ " mb (current store usage is " + storeCurrent / oneMeg
+ " mb). The data directory: " + dir.getAbsolutePath()
+ " only has " + totalUsableSpace / oneMeg
+ " mb of usable space.";
if (!isAdjustUsageLimits()) {
LOG.error(message);
throw new ConfigurationException(message);
}
if (percentLimit > 0) {
LOG.warn(storeName + " limit has been set to "
+ percentLimit + "% (" + bytePercentLimit / oneMeg + " mb)"
@ -2072,14 +2085,10 @@ public class BrokerService implements Service {
+ " previous usage limit check) is set to (" + storeLimit / oneMeg + " mb)"
+ " but only " + totalUsableSpace * 100 / totalSpace + "% (" + totalUsableSpace / oneMeg + " mb)"
+ " is available - resetting limit");
}
LOG.warn(storeName + " limit is " + storeLimit / oneMeg +
" mb (current store usage is " + storeCurrent / oneMeg +
" mb). The data directory: " + dir.getAbsolutePath() +
" only has " + totalUsableSpace / oneMeg +
" mb of usable space - resetting to maximum available disk space: " +
} else {
LOG.warn(message + " - resetting to maximum available disk space: " +
totalUsableSpace / oneMeg + " mb");
}
storeUsage.setLimit(totalUsableSpace);
}
}
@ -2098,13 +2107,13 @@ public class BrokerService implements Service {
public void run() {
try {
checkStoreUsageLimits();
} catch (IOException e) {
} catch (Exception e) {
LOG.error("Failed to check persistent disk usage limits", e);
}
try {
checkTmpStoreUsageLimits();
} catch (IOException e) {
} catch (Exception e) {
LOG.error("Failed to check temporary store usage limits", e);
}
}
@ -2113,17 +2122,27 @@ public class BrokerService implements Service {
}
}
protected void checkSystemUsageLimits() throws IOException {
protected void checkMemorySystemUsageLimits() throws Exception {
final SystemUsage usage = getSystemUsage();
long memLimit = usage.getMemoryUsage().getLimit();
long jvmLimit = Runtime.getRuntime().maxMemory();
if (memLimit > jvmLimit) {
final String message = "Memory Usage for the Broker (" + memLimit / (1024 * 1024)
+ "mb) is more than the maximum available for the JVM: " + jvmLimit / (1024 * 1024);
if (adjustUsageLimits) {
usage.getMemoryUsage().setPercentOfJvmHeap(70);
LOG.warn("Memory Usage for the Broker (" + memLimit / (1024 * 1024) +
" mb) is more than the maximum available for the JVM: " +
jvmLimit / (1024 * 1024) + " mb - resetting to 70% of maximum available: " + (usage.getMemoryUsage().getLimit() / (1024 * 1024)) + " mb");
LOG.warn(message + " mb - resetting to 70% of maximum available: " + (usage.getMemoryUsage().getLimit() / (1024 * 1024)) + " mb");
} else {
LOG.error(message);
throw new ConfigurationException(message);
}
}
}
protected void checkStoreSystemUsageLimits() throws Exception {
final SystemUsage usage = getSystemUsage();
//Check the persistent store and temp store limits if they exist
//and schedule a periodic check to update disk limits if
@ -3168,4 +3187,12 @@ public class BrokerService implements Service {
boolean useVirtualDestSubsOnCreation) {
this.useVirtualDestSubsOnCreation = useVirtualDestSubsOnCreation;
}
public boolean isAdjustUsageLimits() {
return adjustUsageLimits;
}
public void setAdjustUsageLimits(boolean adjustUsageLimits) {
this.adjustUsageLimits = adjustUsageLimits;
}
}

18
activemq-unit-tests/src/test/java/org/apache/activemq/usage/PercentDiskUsageLimitTest.java
View File

@ -17,9 +17,12 @@
package org.apache.activemq.usage;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;
import java.io.File;
import org.apache.activemq.ConfigurationException;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.store.PersistenceAdapter;
import org.apache.activemq.util.StoreUtil;
@ -120,6 +123,21 @@ public class PercentDiskUsageLimitTest {
}
}
@Test(timeout=30000)
public void testStartFailDiskLimitOverMaxFree() throws Exception {
broker.setAdjustUsageLimits(false);
int freePercent = getFreePercentage();
if (freePercent > 1) {
storeUsage.setPercentLimit(freePercent + 1);
try {
startBroker();
fail("Expect ex");
} catch (ConfigurationException expected) {}
}
}
@Test(timeout=30000)
public void testDiskLimitOver100Percent() throws Exception {
int freePercent = getFreePercentage();

93
activemq-unit-tests/src/test/java/org/apache/activemq/usage/StoreUsageLimitsTest.java
View File

@ -17,59 +17,88 @@
package org.apache.activemq.usage;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.InputStreamReader;
import java.nio.charset.Charset;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.apache.activemq.ConfigurationException;
import org.junit.Test;
import org.apache.activemq.EmbeddedBrokerTestSupport;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.util.DefaultTestAppender;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.apache.log4j.spi.LoggingEvent;
public class StoreUsageLimitsTest extends EmbeddedBrokerTestSupport {
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
public class StoreUsageLimitsTest {
final int WAIT_TIME_MILLS = 20 * 1000;
private static final String limitsLogLevel = "warn";
final String toMatch = new String(Long.toString(Long.MAX_VALUE / (1024 * 1024)));
@Override
protected BrokerService createBroker() throws Exception {
BrokerService broker = super.createBroker();
BrokerService broker = new BrokerService();
broker.setPersistent(false);
broker.getSystemUsage().getMemoryUsage().setLimit(Long.MAX_VALUE);
broker.getSystemUsage().setCheckLimitsLogLevel(limitsLogLevel);
broker.deleteAllMessages();
return broker;
}
@Override
protected boolean isPersistent() {
return true;
}
@Test
public void testCheckLimitsLogLevel() throws Exception {
File file = new File("target/activemq-test.log");
if (!file.exists()) {
fail("target/activemq-test.log was not created.");
final CountDownLatch foundMessage = new CountDownLatch(1);
DefaultTestAppender appender = new DefaultTestAppender() {
@Override
public void doAppend(LoggingEvent event) {
String message = (String) event.getMessage();
if (message.contains(toMatch) && event.getLevel().equals(Level.WARN)) {
foundMessage.countDown();
}
}
};
Logger.getRootLogger().addAppender(appender);
BrokerService brokerService = createBroker();
brokerService.start();
brokerService.stop();
assertTrue("Fount log message", foundMessage.await(WAIT_TIME_MILLS, TimeUnit.MILLISECONDS));
Logger.getRootLogger().removeAppender(appender);
}
BufferedReader br = null;
boolean foundUsage = false;
@Test
public void testCheckLimitsFailStart() throws Exception {
final CountDownLatch foundMessage = new CountDownLatch(1);
DefaultTestAppender appender = new DefaultTestAppender() {
@Override
public void doAppend(LoggingEvent event) {
String message = (String) event.getMessage();
if (message.contains(toMatch) && event.getLevel().equals(Level.ERROR)) {
foundMessage.countDown();
}
}
};
Logger.getRootLogger().addAppender(appender);
BrokerService brokerService = createBroker();
brokerService.setAdjustUsageLimits(false);
try {
br = new BufferedReader(new InputStreamReader(new FileInputStream(file), Charset.forName("UTF-8")));
String line = null;
while ((line = br.readLine()) != null) {
if (line.contains(new String(Long.toString(Long.MAX_VALUE / (1024 * 1024)))) && line.contains(limitsLogLevel.toUpperCase())) {
foundUsage = true;
}
}
} catch (Exception e) {
fail(e.getMessage());
} finally {
br.close();
brokerService.start();
fail("expect ConfigurationException");
} catch (ConfigurationException expected) {
assertTrue("exception message match", expected.getLocalizedMessage().contains(toMatch));
}
brokerService.stop();
if (!foundUsage)
fail("checkLimitsLogLevel message did not write to log target/activemq-test.log");
assertTrue("Fount log message", foundMessage.await(WAIT_TIME_MILLS, TimeUnit.MILLISECONDS));
Logger.getRootLogger().removeAppender(appender);
}
}