You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by jj...@apache.org on 2020/05/07 18:15:48 UTC
[geode] branch support/1.13 updated: GEODE-8071: Use daemon threads
in RebalanceCommand (#5054)
This is an automated email from the ASF dual-hosted git repository.
jjramos pushed a commit to branch support/1.13
in repository https://gitbox.apache.org/repos/asf/geode.git
The following commit(s) were added to refs/heads/support/1.13 by this push:
new 193a98a GEODE-8071: Use daemon threads in RebalanceCommand (#5054)
193a98a is described below
commit 193a98a1975a2df417e50755d7badd320c7cb8af
Author: Juan José Ramos <ju...@users.noreply.github.com>
AuthorDate: Wed May 6 12:37:10 2020 +0100
GEODE-8071: Use daemon threads in RebalanceCommand (#5054)
Changed the ExecutorService within RebalanceCommand to use daemon
threads, otherwise the locator refuses to gracefully shutdown.
- Fixed minor warnings.
- Added distributed tests.
- Refactored RebalanceCommandDistributedTest.
(cherry picked from commit d8e86cb720c054b154a16cc88fee88e73db709f3)
---
...e.java => RebalanceCommandDistributedTest.java} | 159 +++++++++++++--------
.../commands/RebalanceCommandDistributedTest.java | 18 ---
.../internal/cli/commands/RebalanceCommand.java | 9 +-
.../RebalanceCommandOverHttpDistributedTest.java | 24 ----
4 files changed, 105 insertions(+), 105 deletions(-)
diff --git a/geode-dunit/src/main/java/org/apache/geode/management/internal/cli/commands/RebalanceCommandDistributedTestBase.java b/geode-dunit/src/main/java/org/apache/geode/management/internal/cli/commands/RebalanceCommandDistributedTest.java
similarity index 56%
rename from geode-dunit/src/main/java/org/apache/geode/management/internal/cli/commands/RebalanceCommandDistributedTestBase.java
rename to geode-dunit/src/main/java/org/apache/geode/management/internal/cli/commands/RebalanceCommandDistributedTest.java
index d69619f..a1a2bdd 100644
--- a/geode-dunit/src/main/java/org/apache/geode/management/internal/cli/commands/RebalanceCommandDistributedTestBase.java
+++ b/geode-dunit/src/main/java/org/apache/geode/management/internal/cli/commands/RebalanceCommandDistributedTest.java
@@ -14,17 +14,20 @@
*/
package org.apache.geode.management.internal.cli.commands;
-
+import static org.apache.geode.cache.Region.SEPARATOR;
+import static org.apache.geode.test.junit.rules.GfshCommandRule.PortType.http;
import static org.apache.geode.test.junit.rules.GfshCommandRule.PortType.jmxManager;
import static org.assertj.core.api.Assertions.assertThat;
+import java.io.Serializable;
import java.util.List;
import java.util.Map;
import org.junit.Before;
-import org.junit.BeforeClass;
-import org.junit.ClassRule;
+import org.junit.Rule;
import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.Region;
@@ -35,40 +38,89 @@ import org.apache.geode.test.dunit.rules.ClusterStartupRule;
import org.apache.geode.test.dunit.rules.MemberVM;
import org.apache.geode.test.junit.assertions.TabularResultModelAssert;
import org.apache.geode.test.junit.rules.GfshCommandRule;
+import org.apache.geode.test.junit.rules.MemberStarterRule;
+
+@RunWith(Parameterized.class)
+public class RebalanceCommandDistributedTest implements Serializable {
+ private static final int ENTRIES_PER_REGION = 200;
+ private static final String REGION_ONE_NAME = "region-1";
+ private static final String REGION_TWO_NAME = "region-2";
+ private static final String REGION_THREE_NAME = "region-3";
-@SuppressWarnings("serial")
-public class RebalanceCommandDistributedTestBase {
+ @Rule
+ public transient GfshCommandRule gfsh = new GfshCommandRule();
- @ClassRule
- public static ClusterStartupRule cluster = new ClusterStartupRule();
+ @Rule
+ public ClusterStartupRule cluster = new ClusterStartupRule();
- @ClassRule
- public static GfshCommandRule gfsh = new GfshCommandRule();
+ protected MemberVM locator, server1, server2;
- protected static MemberVM locator, server1, server2, server3;
+ @Parameterized.Parameters(name = "ConnectionType:{0}")
+ public static GfshCommandRule.PortType[] connectionTypes() {
+ return new GfshCommandRule.PortType[] {http, jmxManager};
+ }
- @BeforeClass
- public static void beforeClass() {
- locator = cluster.startLocatorVM(0, l -> l.withHttpService());
+ @Parameterized.Parameter
+ public static GfshCommandRule.PortType portType;
+ private void setUpRegions() {
+ server1.invoke(() -> {
+ Cache cache = ClusterStartupRule.getCache();
+ assertThat(cache).isNotNull();
+ RegionFactory<String, String> dataRegionFactory =
+ cache.createRegionFactory(RegionShortcut.PARTITION);
+ Region<String, String> region1 = dataRegionFactory.create(REGION_ONE_NAME);
+ Region<String, String> region2 = dataRegionFactory.create(REGION_TWO_NAME);
+
+ for (int i = 0; i < ENTRIES_PER_REGION; i++) {
+ region1.put("key" + i, "Value" + i);
+ region2.put("key" + i, "Value" + i);
+ }
+ });
+
+ server2.invoke(() -> {
+ // no need to close cache as it will be closed as part of teardown2
+ Cache cache = ClusterStartupRule.getCache();
+ assertThat(cache).isNotNull();
+ RegionFactory<String, String> dataRegionFactory =
+ cache.createRegionFactory(RegionShortcut.PARTITION);
+ dataRegionFactory.create(REGION_ONE_NAME);
+ Region<String, String> region3 = dataRegionFactory.create(REGION_THREE_NAME);
+
+ for (int i = 0; i < ENTRIES_PER_REGION; i++) {
+ region3.put("key" + i, "Value" + i);
+ }
+ });
+ }
+
+ @Before
+ public void setUp() throws Exception {
+ locator = cluster.startLocatorVM(0, MemberStarterRule::withHttpService);
int locatorPort = locator.getPort();
+
server1 = cluster.startServerVM(1, "localhost", locatorPort);
server2 = cluster.startServerVM(2, "localhost", locatorPort);
-
setUpRegions();
- }
- @Before
- public void before() throws Exception {
- gfsh.connect(locator.getJmxPort(), jmxManager);
+ switch (portType) {
+ case http:
+ gfsh.connectAndVerify(locator.getHttpPort(), http);
+ break;
+ case jmxManager:
+ gfsh.connectAndVerify(locator.getJmxPort(), jmxManager);
+ break;
+
+ default:
+ throw new IllegalArgumentException("Invalid PortType Configured");
+ }
}
@Test
public void testSimulateForEntireDSWithTimeout() {
// check if DistributedRegionMXBean is available so that command will not fail
- locator.waitUntilRegionIsReadyOnExactlyThisManyServers("/region-1", 2);
- locator.waitUntilRegionIsReadyOnExactlyThisManyServers("/region-2", 1);
- locator.waitUntilRegionIsReadyOnExactlyThisManyServers("/region-3", 1);
+ locator.waitUntilRegionIsReadyOnExactlyThisManyServers(SEPARATOR + REGION_ONE_NAME, 2);
+ locator.waitUntilRegionIsReadyOnExactlyThisManyServers(SEPARATOR + REGION_TWO_NAME, 1);
+ locator.waitUntilRegionIsReadyOnExactlyThisManyServers(SEPARATOR + REGION_THREE_NAME, 1);
String command = "rebalance --simulate=true --time-out=-1";
@@ -78,7 +130,7 @@ public class RebalanceCommandDistributedTestBase {
@Test
public void testRebalanceResultOutput() {
// check if DistributedRegionMXBean is available so that command will not fail
- locator.waitUntilRegionIsReadyOnExactlyThisManyServers("/region-1", 2);
+ locator.waitUntilRegionIsReadyOnExactlyThisManyServers(SEPARATOR + REGION_ONE_NAME, 2);
String command = "rebalance";
@@ -100,18 +152,15 @@ public class RebalanceCommandDistributedTestBase {
@Test
public void testRebalanceResultOutputMemberCount() {
- server3 = cluster.startServerVM(3, "localhost", locator.getPort());
+ MemberVM server3 = cluster.startServerVM(3, "localhost", locator.getPort());
server3.invoke(() -> {
Cache cache = ClusterStartupRule.getCache();
- RegionFactory<Integer, Integer> dataRegionFactory =
- cache.createRegionFactory(RegionShortcut.PARTITION);
- Region region = dataRegionFactory.create("region-1");
- for (int i = 0; i < 100; i++) {
- region.put("key" + (i + 400), "value" + (i + 400));
- }
+ assertThat(cache).isNotNull();
+ cache.createRegionFactory(RegionShortcut.PARTITION).create(REGION_ONE_NAME);
});
+
// check if DistributedRegionMXBean is available so that command will not fail
- locator.waitUntilRegionIsReadyOnExactlyThisManyServers("/region-1", 3);
+ locator.waitUntilRegionIsReadyOnExactlyThisManyServers(SEPARATOR + REGION_ONE_NAME, 3);
Map<String, List<String>> listMembersResult = gfsh.executeAndAssertThat("list members")
.hasTableSection().getActual().getContent();
@@ -132,33 +181,27 @@ public class RebalanceCommandDistributedTestBase {
assertThat(rebalanceResult.get("Value").get(9)).isEqualTo("2");
}
- private static void setUpRegions() {
- server1.invoke(() -> {
- Cache cache = ClusterStartupRule.getCache();
- RegionFactory<Integer, Integer> dataRegionFactory =
- cache.createRegionFactory(RegionShortcut.PARTITION);
- Region region = dataRegionFactory.create("region-1");
- for (int i = 0; i < 10; i++) {
- region.put("key" + (i + 200), "value" + (i + 200));
- }
- region = dataRegionFactory.create("region-2");
- for (int i = 0; i < 100; i++) {
- region.put("key" + (i + 200), "value" + (i + 200));
- }
- });
- server2.invoke(() -> {
- // no need to close cache as it will be closed as part of teardown2
- Cache cache = ClusterStartupRule.getCache();
- RegionFactory<Integer, Integer> dataRegionFactory =
- cache.createRegionFactory(RegionShortcut.PARTITION);
- Region region = dataRegionFactory.create("region-1");
- for (int i = 0; i < 100; i++) {
- region.put("key" + (i + 400), "value" + (i + 400));
- }
- region = dataRegionFactory.create("region-3");
- for (int i = 0; i < 10; i++) {
- region.put("key" + (i + 200), "value" + (i + 200));
- }
- });
+ /**
+ * See GEODE-8071.
+ * The test asserts that the amount of non-daemon threads on the locator VM remains constant
+ * before and after executing the re-balance command.
+ */
+ @Test
+ public void rebalanceCommandShouldNotLaunchNonDaemonThreads() {
+ long nonDaemonThreadsBeforeCommand = locator.invoke(() -> Thread.getAllStackTraces()
+ .keySet()
+ .stream()
+ .filter(thread -> !thread.isDaemon())
+ .count());
+
+ gfsh.executeAndAssertThat("rebalance").statusIsSuccess();
+
+ long daemonThreadsAfterCommand = locator.invoke(() -> Thread.getAllStackTraces()
+ .keySet()
+ .stream()
+ .filter(thread -> !thread.isDaemon())
+ .count());
+
+ assertThat(daemonThreadsAfterCommand).isEqualTo(nonDaemonThreadsBeforeCommand);
}
}
diff --git a/geode-gfsh/src/distributedTest/java/org/apache/geode/management/internal/cli/commands/RebalanceCommandDistributedTest.java b/geode-gfsh/src/distributedTest/java/org/apache/geode/management/internal/cli/commands/RebalanceCommandDistributedTest.java
deleted file mode 100644
index c224dd5..0000000
--- a/geode-gfsh/src/distributedTest/java/org/apache/geode/management/internal/cli/commands/RebalanceCommandDistributedTest.java
+++ /dev/null
@@ -1,18 +0,0 @@
-/*
- * 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;
-
-public class RebalanceCommandDistributedTest extends RebalanceCommandDistributedTestBase {
-}
diff --git a/geode-gfsh/src/main/java/org/apache/geode/management/internal/cli/commands/RebalanceCommand.java b/geode-gfsh/src/main/java/org/apache/geode/management/internal/cli/commands/RebalanceCommand.java
index 5fcce07..d9b54ba 100644
--- a/geode-gfsh/src/main/java/org/apache/geode/management/internal/cli/commands/RebalanceCommand.java
+++ b/geode-gfsh/src/main/java/org/apache/geode/management/internal/cli/commands/RebalanceCommand.java
@@ -60,7 +60,7 @@ public class RebalanceCommand extends GfshCommand {
help = CliStrings.REBALANCE__SIMULATE__HELP) boolean simulate) {
ExecutorService commandExecutors =
- LoggingExecutors.newSingleThreadExecutor("RebalanceCommand", false);
+ LoggingExecutors.newSingleThreadExecutor("RebalanceCommand", true);
List<Future<ResultModel>> commandResult = new ArrayList<>();
ResultModel result;
try {
@@ -193,11 +193,10 @@ public class RebalanceCommand extends GfshCommand {
// TODO EY Move this to its own class
private class ExecuteRebalanceWithTimeout implements Callable<ResultModel> {
-
- String[] includeRegions = null;
- String[] excludeRegions = null;
boolean simulate;
- InternalCache cache = null;
+ InternalCache cache;
+ String[] includeRegions;
+ String[] excludeRegions;
@Override
public ResultModel call() throws Exception {
diff --git a/geode-web/src/distributedTest/java/org/apache/geode/management/internal/cli/commands/RebalanceCommandOverHttpDistributedTest.java b/geode-web/src/distributedTest/java/org/apache/geode/management/internal/cli/commands/RebalanceCommandOverHttpDistributedTest.java
deleted file mode 100644
index a4acb20..0000000
--- a/geode-web/src/distributedTest/java/org/apache/geode/management/internal/cli/commands/RebalanceCommandOverHttpDistributedTest.java
+++ /dev/null
@@ -1,24 +0,0 @@
-/*
- * 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.test.junit.rules.GfshCommandRule.PortType.http;
-
-public class RebalanceCommandOverHttpDistributedTest extends RebalanceCommandDistributedTestBase {
- @Override
- public void before() throws Exception {
- gfsh.connectAndVerify(locator.getHttpPort(), http);
- }
-}