[
https://issues.apache.org/jira/browse/GEODE-3788?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16270147#comment-16270147
]
ASF GitHub Bot commented on GEODE-3788:
---------------------------------------
jinmeiliao closed pull request #1084: GEODE-3788: add alter async-event-queue
command and tests
URL: https://github.com/apache/geode/pull/1084
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/geode-core/src/main/java/org/apache/geode/distributed/internal/ClusterConfigurationService.java
b/geode-core/src/main/java/org/apache/geode/distributed/internal/ClusterConfigurationService.java
index 6640bac4c1..16b86aadc1 100644
---
a/geode-core/src/main/java/org/apache/geode/distributed/internal/ClusterConfigurationService.java
+++
b/geode-core/src/main/java/org/apache/geode/distributed/internal/ClusterConfigurationService.java
@@ -559,11 +559,6 @@ public Configuration getConfiguration(String groupName) {
return getConfigurationRegion().get(groupName);
}
- public Map<String, Configuration> getEntireConfiguration() {
- Set<String> keys = getConfigurationRegion().keySet();
- return getConfigurationRegion().getAll(keys);
- }
-
/**
* Returns the path of Shared configuration directory
*
@@ -653,12 +648,11 @@ public void writeConfigToFile(final Configuration
configuration) throws IOExcept
FileUtils.writeStringToFile(xmlFile, configuration.getCacheXmlContent(),
"UTF-8");
}
- // TODO: return value is never used
- private boolean lockSharedConfiguration() {
+ public boolean lockSharedConfiguration() {
return this.sharedConfigLockingService.lock(SHARED_CONFIG_LOCK_NAME, -1,
-1);
}
- private void unlockSharedConfiguration() {
+ public void unlockSharedConfiguration() {
this.sharedConfigLockingService.unlock(SHARED_CONFIG_LOCK_NAME);
}
@@ -681,7 +675,7 @@ private void unlockSharedConfiguration() {
*
* @return {@link Region} ConfigurationRegion, this should never be null
*/
- private Region<String, Configuration> getConfigurationRegion() {
+ public Region<String, Configuration> getConfigurationRegion() {
Region<String, Configuration> configRegion =
this.cache.getRegion(CONFIG_REGION_NAME);
try {
diff --git
a/geode-core/src/main/java/org/apache/geode/management/internal/cli/commands/AlterAsyncEventQueueCommand.java
b/geode-core/src/main/java/org/apache/geode/management/internal/cli/commands/AlterAsyncEventQueueCommand.java
new file mode 100644
index 0000000000..42f0079b9f
--- /dev/null
+++
b/geode-core/src/main/java/org/apache/geode/management/internal/cli/commands/AlterAsyncEventQueueCommand.java
@@ -0,0 +1,168 @@
+/*
+ * 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.geode.management.internal.cli.commands;
+
+import static
org.apache.geode.management.internal.cli.i18n.CliStrings.CREATE_ASYNC_EVENT_QUEUE__BATCHTIMEINTERVAL__HELP;
+import static
org.apache.geode.management.internal.cli.i18n.CliStrings.CREATE_ASYNC_EVENT_QUEUE__BATCH_SIZE__HELP;
+import static
org.apache.geode.management.internal.cli.i18n.CliStrings.CREATE_ASYNC_EVENT_QUEUE__MAXIMUM_QUEUE_MEMORY__HELP;
+import static
org.apache.geode.management.internal.cli.i18n.CliStrings.IFEXISTS;
+import static
org.apache.geode.management.internal.cli.i18n.CliStrings.IFEXISTS_HELP;
+
+import java.io.IOException;
+
+import javax.xml.parsers.ParserConfigurationException;
+import javax.xml.transform.TransformerException;
+
+import org.springframework.shell.core.annotation.CliCommand;
+import org.springframework.shell.core.annotation.CliOption;
+import org.w3c.dom.Document;
+import org.w3c.dom.Element;
+import org.w3c.dom.NodeList;
+import org.xml.sax.SAXException;
+
+import org.apache.geode.cache.Region;
+import org.apache.geode.distributed.internal.ClusterConfigurationService;
+import org.apache.geode.management.cli.CliMetaData;
+import org.apache.geode.management.cli.Result;
+import org.apache.geode.management.internal.cli.AbstractCliAroundInterceptor;
+import org.apache.geode.management.internal.cli.GfshParseResult;
+import
org.apache.geode.management.internal.cli.exceptions.EntityNotFoundException;
+import org.apache.geode.management.internal.cli.result.ResultBuilder;
+import org.apache.geode.management.internal.cli.result.TabularResultData;
+import org.apache.geode.management.internal.configuration.domain.Configuration;
+import org.apache.geode.management.internal.configuration.utils.XmlUtils;
+import org.apache.geode.management.internal.security.ResourceOperation;
+import org.apache.geode.security.ResourcePermission;
+
+/**
+ * this command currently only updates the cluster configuration. Requires
server restart to pick up
+ * the changes.
+ */
+public class AlterAsyncEventQueueCommand implements GfshCommand {
+
+ static final String COMMAND_NAME = "alter async-event-queue";
+ static final String ID = "id";
+ static final String BATCH_SIZE = "batch-size";
+ static final String BATCH_TIME_INTERVAL = "batch-time-interval";
+ static final String MAX_QUEUE_MEMORY = "max-queue-memory";
+ static final String MAXIMUM_QUEUE_MEMORY = "maximum-queue-memory";
+
+ static final String COMMAND_HELP =
+ "alter attributes of async-event-queue, needs rolling restart for new
attributes to take effect. ";
+ static final String ID_HELP = "Id of the async event queue to be changed.";
+ static final String BATCH_SIZE_HELP =
CREATE_ASYNC_EVENT_QUEUE__BATCH_SIZE__HELP;
+ static final String BATCH_TIME_INTERVAL_HELP =
CREATE_ASYNC_EVENT_QUEUE__BATCHTIMEINTERVAL__HELP;
+ static final String MAXIMUM_QUEUE_MEMORY_HELP =
+ CREATE_ASYNC_EVENT_QUEUE__MAXIMUM_QUEUE_MEMORY__HELP;
+
+
+ @CliCommand(value = COMMAND_NAME, help = COMMAND_HELP)
+ @CliMetaData(
+ interceptor =
"org.apache.geode.management.internal.cli.commands.AlterAsyncEventQueueCommand$Interceptor")
+ @ResourceOperation(resource = ResourcePermission.Resource.CLUSTER,
+ operation = ResourcePermission.Operation.MANAGE, target =
ResourcePermission.Target.DEPLOY)
+ public Result execute(@CliOption(key = ID, mandatory = true, help = ID_HELP)
String id,
+ @CliOption(key = BATCH_SIZE, help = BATCH_SIZE_HELP) Integer batchSize,
+ @CliOption(key = BATCH_TIME_INTERVAL,
+ help = BATCH_TIME_INTERVAL_HELP) Integer batchTimeInterval,
+ @CliOption(key = MAX_QUEUE_MEMORY, help = MAXIMUM_QUEUE_MEMORY_HELP)
Integer maxQueueMemory,
+ @CliOption(key = IFEXISTS, help = IFEXISTS_HELP, specifiedDefaultValue =
"true",
+ unspecifiedDefaultValue = "false") boolean ifExists)
+ throws IOException, SAXException, ParserConfigurationException,
TransformerException {
+
+ // need not check if any running servers has this async-event-queue. A
server with this queue id
+ // may be shutdown, but we still need to update Cluster Configuration.
+ ClusterConfigurationService service = getSharedConfiguration();
+
+ boolean locked = service.lockSharedConfiguration();
+ if (!locked) {
+ return ResultBuilder.createGemFireErrorResult("Unable to lock the
cluster configuration.");
+ }
+
+ TabularResultData tableData = ResultBuilder.createTabularResultData();
+ try {
+ Region<String, Configuration> configRegion =
service.getConfigurationRegion();
+ for (String group : configRegion.keySet()) {
+ Configuration config = configRegion.get(group);
+ if (config.getCacheXmlContent() == null) {
+ // skip to the next group
+ continue;
+ }
+
+ boolean xmlUpdated = false;
+ Document document =
XmlUtils.createDocumentFromXml(config.getCacheXmlContent());
+ NodeList nodeList = document.getElementsByTagName("async-event-queue");
+ for (int i = 0; i < nodeList.getLength(); i++) {
+ Element item = (Element) nodeList.item(i);
+ String queueId = item.getAttribute("id");
+ if (!id.equals(queueId)) {
+ // skip to the next async-event-queue found in this xml
+ continue;
+ }
+ // this node is the async-event-queue with the correct id
+ if (batchSize != null) {
+ item.setAttribute(BATCH_SIZE, batchSize + "");
+ }
+ if (batchTimeInterval != null) {
+ item.setAttribute(BATCH_TIME_INTERVAL, batchTimeInterval + "");
+ }
+ if (maxQueueMemory != null) {
+ item.setAttribute(MAXIMUM_QUEUE_MEMORY, maxQueueMemory + "");
+ }
+ // each group should have only one queue with this id defined
+ tableData.accumulate("Group", group);
+ tableData.accumulate("Status", "Cluster Configuration Updated");
+ xmlUpdated = true;
+ break;
+ }
+
+ if (xmlUpdated) {
+ String newXml = XmlUtils.prettyXml(document.getFirstChild());
+ config.setCacheXmlContent(newXml);
+ configRegion.put(group, config);
+ }
+ }
+ } finally {
+ service.unlockSharedConfiguration();
+ }
+
+ if (tableData.rowSize("Group") == 0) {
+ String message = String.format("Can not find an async event queue with
id '%s'.", id);
+ throw new EntityNotFoundException(message, ifExists);
+ }
+
+ // some configurations are changed, print out the warning message as well.
+ tableData.setFooter(System.lineSeparator()
+ + "These changes won't take effect on the running servers. " +
System.lineSeparator()
+ + "Please restart the servers in these groups for the changes to take
effect.");
+ return ResultBuilder.buildResult(tableData);
+ }
+
+ public static class Interceptor extends AbstractCliAroundInterceptor {
+ @Override
+ public Result preExecution(GfshParseResult parseResult) {
+ Object batchSize = parseResult.getParamValue(BATCH_SIZE);
+ Object batchTimeInterval =
parseResult.getParamValue(BATCH_TIME_INTERVAL);
+ Object maxQueueMemory = parseResult.getParamValue(MAX_QUEUE_MEMORY);
+
+ if (batchSize == null && batchTimeInterval == null && maxQueueMemory ==
null) {
+ return ResultBuilder
+ .createUserErrorResult("need to specify at least one option to
modify.");
+ }
+ return ResultBuilder.createInfoResult("");
+ }
+ }
+}
diff --git
a/geode-core/src/main/java/org/apache/geode/management/internal/cli/commands/ExportImportClusterConfigurationCommands.java
b/geode-core/src/main/java/org/apache/geode/management/internal/cli/commands/ExportImportClusterConfigurationCommands.java
index 89e452dd90..e85c1aed36 100644
---
a/geode-core/src/main/java/org/apache/geode/management/internal/cli/commands/ExportImportClusterConfigurationCommands.java
+++
b/geode-core/src/main/java/org/apache/geode/management/internal/cli/commands/ExportImportClusterConfigurationCommands.java
@@ -96,7 +96,7 @@ public Result exportSharedConfig(@CliOption(key =
{CliStrings.EXPORT_SHARED_CONF
Result result;
try {
- for (Configuration config : sc.getEntireConfiguration().values()) {
+ for (Configuration config : sc.getConfigurationRegion().values()) {
sc.writeConfigToFile(config);
}
ZipUtils.zipDirectory(sc.getSharedConfigurationDirPath(),
zipFile.getCanonicalPath());
@@ -162,7 +162,7 @@ public Result importSharedConfig(@CliOption(key =
{CliStrings.IMPORT_SHARED_CONF
ClusterConfigurationService sc = locator.getSharedConfiguration();
// backup the old config
- for (Configuration config : sc.getEntireConfiguration().values()) {
+ for (Configuration config : sc.getConfigurationRegion().values()) {
sc.writeConfigToFile(config);
}
sc.renameExistingSharedConfigDirectory();
diff --git
a/geode-core/src/main/java/org/apache/geode/management/internal/cli/json/GfJsonObject.java
b/geode-core/src/main/java/org/apache/geode/management/internal/cli/json/GfJsonObject.java
index 0000ec5b78..33261184df 100644
---
a/geode-core/src/main/java/org/apache/geode/management/internal/cli/json/GfJsonObject.java
+++
b/geode-core/src/main/java/org/apache/geode/management/internal/cli/json/GfJsonObject.java
@@ -305,6 +305,9 @@ public boolean has(String key) {
return jsonObject.keys();
}
+ /**
+ * @return the column size of this GfJsonObject
+ */
public int size() {
return jsonObject.length();
}
diff --git
a/geode-core/src/main/java/org/apache/geode/management/internal/cli/result/TabularResultData.java
b/geode-core/src/main/java/org/apache/geode/management/internal/cli/result/TabularResultData.java
index c694a8b671..92e326123f 100644
---
a/geode-core/src/main/java/org/apache/geode/management/internal/cli/result/TabularResultData.java
+++
b/geode-core/src/main/java/org/apache/geode/management/internal/cli/result/TabularResultData.java
@@ -51,6 +51,24 @@ public GfJsonArray getHeaders() {
return null;
}
+ public int columnSize() {
+ return contentObject.size();
+ }
+
+ public int rowSize(String key) {
+ GfJsonArray jsonArray = null;
+ try {
+ jsonArray = contentObject.getJSONArray(key);
+ } catch (GfJsonException e) {
+ throw new RuntimeException("unable to get the row size of " + key);
+ }
+ if (jsonArray == null) {
+ return 0;
+ }
+
+ return jsonArray.getInternalJsonArray().length();
+ }
+
/**
* @return the gfJsonObject
*/
diff --git
a/geode-core/src/test/java/org/apache/geode/management/internal/cli/commands/AlterAsyncEventQueueCommandDUnitTest.java
b/geode-core/src/test/java/org/apache/geode/management/internal/cli/commands/AlterAsyncEventQueueCommandDUnitTest.java
new file mode 100644
index 0000000000..daf36be02f
--- /dev/null
+++
b/geode-core/src/test/java/org/apache/geode/management/internal/cli/commands/AlterAsyncEventQueueCommandDUnitTest.java
@@ -0,0 +1,100 @@
+/*
+ * 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.geode.management.internal.cli.commands;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.cache.asyncqueue.AsyncEventQueue;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.cache.wan.MyAsyncEventListener;
+import org.apache.geode.test.dunit.rules.LocatorServerStartupRule;
+import org.apache.geode.test.dunit.rules.MemberVM;
+import org.apache.geode.test.junit.categories.AcceptanceTest;
+import org.apache.geode.test.junit.rules.GfshCommandRule;
+
+
+@Category(AcceptanceTest.class)
+public class AlterAsyncEventQueueCommandDUnitTest {
+
+ @Rule
+ public LocatorServerStartupRule lsRule = new LocatorServerStartupRule();
+
+ @Rule
+ public GfshCommandRule gfsh = new GfshCommandRule();
+
+ private static MemberVM locator, server1, server2;
+
+ @Before
+ public void beforeClass() throws Exception {
+ locator = lsRule.startLocatorVM(0);
+ server1 = lsRule.startServerVM(1, "group1", locator.getPort());
+ gfsh.connectAndVerify(locator);
+ }
+
+
+
+ @Test
+ public void testAlterAsyncEventQueue() throws Exception {
+ gfsh.executeAndAssertThat("create async-event-queue --id=queue1
--group=group1 --listener="
+ + MyAsyncEventListener.class.getName()).statusIsSuccess();
+
+ locator.waitTillAsyncEventQueuesAreReadyOnServers("queue1", 1);
+
+ // verify that server1's event queue has the default value
+ server1.invoke(() -> {
+ InternalCache cache = LocatorServerStartupRule.getCache();
+ AsyncEventQueue queue = cache.getAsyncEventQueue("queue1");
+ assertThat(queue.getBatchSize()).isEqualTo(100);
+ assertThat(queue.getBatchTimeInterval()).isEqualTo(1000);
+ assertThat(queue.getMaximumQueueMemory()).isEqualTo(100);
+ });
+
+ gfsh.executeAndAssertThat("alter async-event-queue --id=queue1 " +
"--batch-size=200 "
+ + "--batch-time-interval=300 " +
"--max-queue-memory=400").statusIsSuccess();
+
+ // verify that server1's event queue still has the default value
+ // without restart
+ server1.invoke(() -> {
+ InternalCache cache = LocatorServerStartupRule.getCache();
+ AsyncEventQueue queue = cache.getAsyncEventQueue("queue1");
+ assertThat(queue.getBatchSize()).isEqualTo(100);
+ assertThat(queue.getBatchTimeInterval()).isEqualTo(1000);
+ assertThat(queue.getMaximumQueueMemory()).isEqualTo(100);
+ assertThat(cache.getAsyncEventQueue("queue2")).isNull();
+ });
+
+ // restart locator and server without clearing the file system
+ lsRule.stopVM(1, false);
+ lsRule.stopVM(0, false);
+
+ locator = lsRule.startLocatorVM(0);
+ server1 = lsRule.startServerVM(1, "group1", locator.getPort());
+ // verify that server1's queue is updated
+ server1.invoke(() -> {
+ InternalCache cache = LocatorServerStartupRule.getCache();
+ AsyncEventQueue queue = cache.getAsyncEventQueue("queue1");
+ assertThat(queue.getBatchSize()).isEqualTo(200);
+ assertThat(queue.getBatchTimeInterval()).isEqualTo(300);
+ assertThat(queue.getMaximumQueueMemory()).isEqualTo(400);
+ assertThat(cache.getAsyncEventQueue("queue2")).isNull();
+ });
+ }
+}
diff --git
a/geode-core/src/test/java/org/apache/geode/management/internal/cli/commands/AlterAsyncEventQueueCommandTest.java
b/geode-core/src/test/java/org/apache/geode/management/internal/cli/commands/AlterAsyncEventQueueCommandTest.java
new file mode 100644
index 0000000000..38dc072cff
--- /dev/null
+++
b/geode-core/src/test/java/org/apache/geode/management/internal/cli/commands/AlterAsyncEventQueueCommandTest.java
@@ -0,0 +1,239 @@
+/*
+ * 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.geode.management.internal.cli.commands;
+
+import static
org.apache.geode.management.internal.cli.commands.AlterAsyncEventQueueCommand.BATCH_SIZE;
+import static
org.apache.geode.management.internal.cli.commands.AlterAsyncEventQueueCommand.BATCH_TIME_INTERVAL;
+import static
org.apache.geode.management.internal.cli.commands.AlterAsyncEventQueueCommand.ID;
+import static
org.apache.geode.management.internal.cli.commands.AlterAsyncEventQueueCommand.MAXIMUM_QUEUE_MEMORY;
+import static
org.apache.geode.management.internal.cli.commands.AlterAsyncEventQueueCommand.MAX_QUEUE_MEMORY;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.stream.Collectors;
+
+import org.junit.Before;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.w3c.dom.Document;
+import org.w3c.dom.Element;
+import org.w3c.dom.NodeList;
+
+import org.apache.geode.cache.Region;
+import org.apache.geode.distributed.internal.ClusterConfigurationService;
+import org.apache.geode.management.internal.configuration.domain.Configuration;
+import org.apache.geode.management.internal.configuration.utils.XmlUtils;
+import org.apache.geode.test.junit.categories.UnitTest;
+import org.apache.geode.test.junit.rules.GfshParserRule;
+
+
+@Category(UnitTest.class)
+public class AlterAsyncEventQueueCommandTest {
+
+ @ClassRule
+ public static GfshParserRule gfsh = new GfshParserRule();
+
+ private AlterAsyncEventQueueCommand command;
+ private ClusterConfigurationService service;
+ private Region<String, Configuration> configRegion;
+
+ @Before
+ public void before() throws Exception {
+ command = spy(AlterAsyncEventQueueCommand.class);
+ service = mock(ClusterConfigurationService.class);
+ doReturn(service).when(command).getSharedConfiguration();
+ configRegion = mock(Region.class);
+ when(service.getConfigurationRegion()).thenReturn(configRegion);
+ when(service.lockSharedConfiguration()).thenReturn(true);
+
+ when(configRegion.keySet())
+
.thenReturn(Arrays.stream("group1,group2".split(",")).collect(Collectors.toSet()));
+ Configuration configuration1 = new Configuration("group1");
+ configuration1.setCacheXmlContent(getCacheXml("queue1"));
+ when(configRegion.get("group1")).thenReturn(configuration1);
+
+ Configuration configuration2 = new Configuration("group2");
+ configuration2.setCacheXmlContent(getCacheXml("queue2"));
+ when(configRegion.get("group2")).thenReturn(configuration2);
+
+ }
+
+ @Test
+ public void mandatoryOption() throws Exception {
+ gfsh.executeAndAssertThat(command, "alter
async-event-queue").statusIsError()
+ .containsOutput("Invalid command");
+ }
+
+ @Test
+ public void noOptionToModify() throws Exception {
+ gfsh.executeAndAssertThat(command, "alter async-event-queue
--id=test").statusIsError()
+ .containsOutput("need to specify at least one option to modify.");
+ }
+
+ @Test
+ public void emptyConfiguration() throws Exception {
+ gfsh.executeAndAssertThat(command, "alter async-event-queue --id=test
--batch-size=100")
+ .statusIsError().containsOutput("Can not find an async event queue");
+
+ verify(service).lockSharedConfiguration();
+ verify(service).unlockSharedConfiguration();
+ }
+
+ @Test
+ public void emptyConfiguration_ifExists() throws Exception {
+ gfsh.executeAndAssertThat(command,
+ "alter async-event-queue --id=test --batch-size=100
--if-exists").statusIsSuccess()
+ .containsOutput("Skipping: Can not find an async event queue with id");
+
+ verify(service).lockSharedConfiguration();
+ verify(service).unlockSharedConfiguration();
+ }
+
+ @Test
+ public void queueIdNotFoundInTheMap() throws Exception {
+ Configuration configuration = new Configuration("group");
+ configuration.setCacheXmlContent(getCacheXml("queue1", "queue2"));
+ configRegion.put("group", configuration);
+
+ gfsh.executeAndAssertThat(command, "alter async-event-queue
--batch-size=100 --id=queue")
+ .statusIsError().containsOutput("Can not find an async event queue");
+
+ verify(service).lockSharedConfiguration();
+ verify(service).unlockSharedConfiguration();
+ }
+
+ @Test
+ public void cannotLockClusterConfiguration() throws Exception {
+ when(service.lockSharedConfiguration()).thenReturn(false);
+ gfsh.executeAndAssertThat(command, "alter async-event-queue
--batch-size=100 --id=queue")
+ .statusIsError().containsOutput("Unable to lock the cluster
configuration");
+ }
+
+ @Test
+ public void queueIdFoundInTheMap_updateBatchSize() throws Exception {
+ gfsh.executeAndAssertThat(command, "alter async-event-queue
--batch-size=100 --id=queue1")
+ .statusIsSuccess().tableHasRowCount("Group", 1)
+ .tableHasRowWithValues("Group", "Status", "group1", "Cluster
Configuration Updated")
+ .containsOutput("Please restart the servers");
+
+ // verify that the xml is updated
+ Element element =
+
findAsyncEventQueueElement(configRegion.get("group1").getCacheXmlContent(), 0);
+ assertThat(element.getAttribute(ID)).isEqualTo("queue1");
+ assertThat(element.getAttribute(BATCH_SIZE)).isEqualTo("100");
+ assertThat(element.getAttribute(BATCH_TIME_INTERVAL)).isEqualTo("");
+ assertThat(element.getAttribute(MAX_QUEUE_MEMORY)).isEqualTo("");
+
+ verify(service).lockSharedConfiguration();
+ verify(service).unlockSharedConfiguration();
+ }
+
+ @Test
+ public void queueIdFoundInTheMap_updateBatchTimeInterval() throws Exception {
+ gfsh.executeAndAssertThat(command,
+ "alter async-event-queue --batch-time-interval=100
--id=queue1").statusIsSuccess()
+ .tableHasRowCount("Group", 1)
+ .tableHasRowWithValues("Group", "Status", "group1", "Cluster
Configuration Updated")
+ .containsOutput("Please restart the servers");
+
+ // verify that the xml is updated
+ Element element =
+
findAsyncEventQueueElement(configRegion.get("group1").getCacheXmlContent(), 0);
+ assertThat(element.getAttribute(ID)).isEqualTo("queue1");
+ assertThat(element.getAttribute(BATCH_SIZE)).isEqualTo("");
+ assertThat(element.getAttribute(BATCH_TIME_INTERVAL)).isEqualTo("100");
+ assertThat(element.getAttribute(MAX_QUEUE_MEMORY)).isEqualTo("");
+
+ verify(service).lockSharedConfiguration();
+ verify(service).unlockSharedConfiguration();
+ }
+
+ @Test
+ public void queueIdFoundInTheMap_updateMaxMemory() throws Exception {
+ gfsh.executeAndAssertThat(command, "alter async-event-queue
--max-queue-memory=100 --id=queue1")
+ .statusIsSuccess().tableHasRowCount("Group", 1)
+ .tableHasRowWithValues("Group", "Status", "group1", "Cluster
Configuration Updated")
+ .containsOutput("Please restart the servers");
+
+ // verify that the xml is updated
+ Element element =
+
findAsyncEventQueueElement(configRegion.get("group1").getCacheXmlContent(), 0);
+ assertThat(element.getAttribute(ID)).isEqualTo("queue1");
+ assertThat(element.getAttribute(BATCH_SIZE)).isEqualTo("");
+ assertThat(element.getAttribute(BATCH_TIME_INTERVAL)).isEqualTo("");
+ assertThat(element.getAttribute(MAXIMUM_QUEUE_MEMORY)).isEqualTo("100");
+
+ verify(service).lockSharedConfiguration();
+ verify(service).unlockSharedConfiguration();
+ }
+
+ @Test
+ public void multipleQueuesInClusterConfig() throws Exception {
+ when(configRegion.keySet()).thenReturn(Collections.singleton("group"));
+ Configuration configuration = new Configuration("group");
+ configuration.setCacheXmlContent(getCacheXml("queue1", "queue2"));
+ when(configRegion.get("group")).thenReturn(configuration);
+
+ gfsh.executeAndAssertThat(command, "alter async-event-queue
--batch-size=100 --id=queue1")
+ .statusIsSuccess().tableHasRowCount("Group", 1)
+ .tableHasRowWithValues("Group", "Status", "group", "Cluster
Configuration Updated")
+ .containsOutput("Please restart the servers");
+
+ // verify that queue1's xml is updated
+ Element element =
findAsyncEventQueueElement(configRegion.get("group").getCacheXmlContent(), 0);
+ assertThat(element.getAttribute(ID)).isEqualTo("queue1");
+ assertThat(element.getAttribute(BATCH_SIZE)).isEqualTo("100");
+ assertThat(element.getAttribute(BATCH_TIME_INTERVAL)).isEqualTo("");
+ assertThat(element.getAttribute(MAX_QUEUE_MEMORY)).isEqualTo("");
+
+ // verify that queue2's xml is untouched
+ element =
findAsyncEventQueueElement(configRegion.get("group").getCacheXmlContent(), 1);
+ assertThat(element.getAttribute(ID)).isEqualTo("queue2");
+ assertThat(element.getAttribute(BATCH_SIZE)).isEqualTo("");
+ assertThat(element.getAttribute(BATCH_TIME_INTERVAL)).isEqualTo("");
+ assertThat(element.getAttribute(MAX_QUEUE_MEMORY)).isEqualTo("");
+
+ verify(service).lockSharedConfiguration();
+ verify(service).unlockSharedConfiguration();
+ }
+
+ private Element findAsyncEventQueueElement(String xml, int index) throws
Exception {
+ Document document = XmlUtils.createDocumentFromXml(xml);
+ NodeList nodeList = document.getElementsByTagName("async-event-queue");
+ return (Element) nodeList.item(index);
+ }
+
+ private String getAsyncEventQueueXml(String queueId) {
+ String xml = "<async-event-queue dispatcher-threads=\"1\" id=\"" + queueId
+ "\">\n"
+ + " <async-event-listener>\n"
+ + "
<class-name>org.apache.geode.internal.cache.wan.MyAsyncEventListener</class-name>\n"
+ + " </async-event-listener>\n" + " </async-event-queue>\n";
+ return xml;
+ }
+
+ private String getCacheXml(String... queueIds) {
+ String xml = "<cache>\n" + Arrays.stream(queueIds).map(x ->
getAsyncEventQueueXml(x))
+ .collect(Collectors.joining("\n")) + "</cache>";
+ return xml;
+ }
+}
diff --git
a/geode-core/src/test/java/org/apache/geode/management/internal/cli/result/TabularResultDataTest.java
b/geode-core/src/test/java/org/apache/geode/management/internal/cli/result/TabularResultDataTest.java
new file mode 100644
index 0000000000..f812952a9a
--- /dev/null
+++
b/geode-core/src/test/java/org/apache/geode/management/internal/cli/result/TabularResultDataTest.java
@@ -0,0 +1,51 @@
+/*
+ * 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.geode.management.internal.cli.result;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.test.junit.categories.UnitTest;
+
+
+@Category(UnitTest.class)
+public class TabularResultDataTest {
+
+ TabularResultData data;
+
+ @Before
+ public void before() throws Exception {
+ data = new TabularResultData();
+ }
+
+ @Test
+ public void rowColumnSize() throws Exception {
+ data.accumulate("key", "value1");
+ assertThat(data.rowSize("key")).isEqualTo(1);
+ assertThat(data.columnSize()).isEqualTo(1);
+
+ data.accumulate("key", "value2");
+ assertThat(data.rowSize("key")).isEqualTo(2);
+ assertThat(data.columnSize()).isEqualTo(1);
+
+ data.accumulate("key1", "value1");
+ assertThat(data.rowSize("key1")).isEqualTo(1);
+ assertThat(data.columnSize()).isEqualTo(2);
+ }
+}
diff --git
a/geode-core/src/test/java/org/apache/geode/management/internal/configuration/ClusterConfig.java
b/geode-core/src/test/java/org/apache/geode/management/internal/configuration/ClusterConfig.java
index 57c05879e2..2a6f96286f 100644
---
a/geode-core/src/test/java/org/apache/geode/management/internal/configuration/ClusterConfig.java
+++
b/geode-core/src/test/java/org/apache/geode/management/internal/configuration/ClusterConfig.java
@@ -97,7 +97,7 @@ public void verifyLocator(MemberVM locatorVM) {
ClusterConfigurationService sc =
internalLocator.getSharedConfiguration();
// verify no extra configs exist in memory
- Set<String> actualGroupConfigs = sc.getEntireConfiguration().keySet();
+ Set<String> actualGroupConfigs = sc.getConfigurationRegion().keySet();
assertThat(actualGroupConfigs).isEqualTo(expectedGroupConfigs);
for (ConfigGroup configGroup : this.getGroups()) {
diff --git
a/geode-core/src/test/java/org/apache/geode/management/internal/configuration/ImportClusterConfigDistributedTest.java
b/geode-core/src/test/java/org/apache/geode/management/internal/configuration/ImportClusterConfigDistributedTest.java
index 46d1f7a83e..11795f7a50 100644
---
a/geode-core/src/test/java/org/apache/geode/management/internal/configuration/ImportClusterConfigDistributedTest.java
+++
b/geode-core/src/test/java/org/apache/geode/management/internal/configuration/ImportClusterConfigDistributedTest.java
@@ -75,8 +75,8 @@ public void exportClusterConfig() throws Exception {
.statusIsSuccess();
gfsh.disconnect();
- locator.stopMember();
- server.stopMember();
+ locator.stopMember(true);
+ server.stopMember(true);
assertThat(this.exportedClusterConfig).exists();
assertThat(this.exportedClusterConfig.length()).isGreaterThan(100);
diff --git
a/geode-core/src/test/java/org/apache/geode/test/dunit/rules/LocatorServerStartupRule.java
b/geode-core/src/test/java/org/apache/geode/test/dunit/rules/LocatorServerStartupRule.java
index a3c2944898..c9e2c8337c 100644
---
a/geode-core/src/test/java/org/apache/geode/test/dunit/rules/LocatorServerStartupRule.java
+++
b/geode-core/src/test/java/org/apache/geode/test/dunit/rules/LocatorServerStartupRule.java
@@ -301,10 +301,14 @@ public MemberVM startServerAsEmbededLocator(int index,
Properties properties) th
}
public void stopVM(int index) {
+ stopVM(index, true);
+ }
+
+ public void stopVM(int index, boolean cleanWorkingDir) {
MemberVM member = members.get(index);
// user has started a server/locator in this VM
if (member != null) {
- member.stopMember();
+ member.stopMember(cleanWorkingDir);
}
// user may have used this VM as a client VM
else {
diff --git
a/geode-core/src/test/java/org/apache/geode/test/dunit/rules/MemberVM.java
b/geode-core/src/test/java/org/apache/geode/test/dunit/rules/MemberVM.java
index 4e04b8eb8b..a61fabb5e3 100644
--- a/geode-core/src/test/java/org/apache/geode/test/dunit/rules/MemberVM.java
+++ b/geode-core/src/test/java/org/apache/geode/test/dunit/rules/MemberVM.java
@@ -101,8 +101,12 @@ public int getEmbeddedLocatorPort() {
return ((Server) member).getEmbeddedLocatorPort();
}
- public void stopMember() {
+ public void stopMember(boolean cleanWorkingDir) {
this.invoke(LocatorServerStartupRule::stopMemberInThisVM);
+ if (!cleanWorkingDir) {
+ return;
+ }
+
if (tempWorkingDir) {
/*
* this temporary workingDir will dynamically change the "user.dir".
system property to point
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
> alter async event queue attributes
> ----------------------------------
>
> Key: GEODE-3788
> URL: https://issues.apache.org/jira/browse/GEODE-3788
> Project: Geode
> Issue Type: Sub-task
> Components: docs, gfsh
> Reporter: Swapnil Bawaskar
> Fix For: 1.4.0
>
>
> We should add a new {{alter async-event-queue}} gfsh command that will allow
> users to change the following attributes on the AsyncEventQueue:
> - batch size
> - batch time interval
> - maximum queue memory
> Attributes changed with this command should only be reflected in cluster
> configuration. We will require users to do a rolling re-start of the servers
> for the new settings to take effect.
--
This message was sent by Atlassian JIRA
(v6.4.14#64029)
