You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by nn...@apache.org on 2017/12/22 19:41:23 UTC
[geode] branch develop updated: GEODE-4154: Removed
GemFireCacheImpl.getInstance
This is an automated email from the ASF dual-hosted git repository.
nnag pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/geode.git
The following commit(s) were added to refs/heads/develop by this push:
new 507c419 GEODE-4154: Removed GemFireCacheImpl.getInstance
507c419 is described below
commit 507c4199c49c41c1b272a03df04c5e760e738661
Author: nabarun <nn...@pivotal.io>
AuthorDate: Thu Dec 21 16:05:43 2017 -0800
GEODE-4154: Removed GemFireCacheImpl.getInstance
* Refactored the test code so that GemFireCacheImpl.getInstance is nolonger called.
* Upgraded the WANTestBase to use LogService than LoggerServiceUtils.
* Optimized the imports.
---
.../geode/internal/cache/wan/WANTestBase.java | 373 ++++++++++++++++++--
.../geode/management/WANManagementDUnitTest.java | 374 ++++-----------------
.../internal/pulse/TestRemoteClusterDUnitTest.java | 250 --------------
3 files changed, 410 insertions(+), 587 deletions(-)
diff --git a/geode-wan/src/test/java/org/apache/geode/internal/cache/wan/WANTestBase.java b/geode-wan/src/test/java/org/apache/geode/internal/cache/wan/WANTestBase.java
index 0e98dea..5c361b1 100644
--- a/geode-wan/src/test/java/org/apache/geode/internal/cache/wan/WANTestBase.java
+++ b/geode-wan/src/test/java/org/apache/geode/internal/cache/wan/WANTestBase.java
@@ -70,7 +70,10 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
+import javax.management.ObjectName;
+
import org.apache.commons.io.FileUtils;
+import org.apache.logging.log4j.Logger;
import org.awaitility.Awaitility;
import org.junit.experimental.categories.Category;
@@ -112,6 +115,7 @@ import org.apache.geode.cache.wan.GatewaySender.OrderPolicy;
import org.apache.geode.cache.wan.GatewaySenderFactory;
import org.apache.geode.cache.wan.GatewayTransportFilter;
import org.apache.geode.cache30.CacheTestCase;
+import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.Locator;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.InternalLocator;
@@ -143,6 +147,15 @@ import org.apache.geode.internal.cache.wan.parallel.ParallelGatewaySenderEventPr
import org.apache.geode.internal.cache.wan.parallel.ParallelGatewaySenderQueue;
import org.apache.geode.internal.cache.wan.serial.ConcurrentSerialGatewaySenderEventProcessor;
import org.apache.geode.internal.cache.wan.serial.SerialGatewaySenderQueue;
+import org.apache.geode.internal.logging.LogService;
+import org.apache.geode.management.AsyncEventQueueMXBean;
+import org.apache.geode.management.DistributedSystemMXBean;
+import org.apache.geode.management.GatewayReceiverMXBean;
+import org.apache.geode.management.GatewaySenderMXBean;
+import org.apache.geode.management.MBeanUtil;
+import org.apache.geode.management.ManagementService;
+import org.apache.geode.management.RegionMXBean;
+import org.apache.geode.management.internal.SystemManagementService;
import org.apache.geode.pdx.SimpleClass;
import org.apache.geode.pdx.SimpleClass1;
import org.apache.geode.test.dunit.Assert;
@@ -151,6 +164,7 @@ import org.apache.geode.test.dunit.Host;
import org.apache.geode.test.dunit.IgnoredException;
import org.apache.geode.test.dunit.Invoke;
import org.apache.geode.test.dunit.LogWriterUtils;
+import org.apache.geode.test.dunit.SerializableRunnable;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.Wait;
import org.apache.geode.test.dunit.WaitCriterion;
@@ -195,6 +209,8 @@ public class WANTestBase extends JUnit4DistributedTestCase {
// this will be set for each test method run with one of the values from above list
protected static int numDispatcherThreadsForTheRun = 1;
+ private static final Logger logger = LogService.getLogger();
+
public WANTestBase() {
super();
}
@@ -784,8 +800,7 @@ public class WANTestBase extends JUnit4DistributedTestCase {
customerRegion =
(PartitionedRegion) cache.createRegionFactory(fact.create()).create(customerRegionName);
assertNotNull(customerRegion);
- LogWriterUtils.getLogWriter()
- .info("Partitioned Region CUSTOMER created Successfully :" + customerRegion.toString());
+ logger.info("Partitioned Region CUSTOMER created Successfully :" + customerRegion.toString());
paf = new PartitionAttributesFactory();
paf.setRedundantCopies(redundantCopies).setTotalNumBuckets(totalNumBuckets)
@@ -804,8 +819,7 @@ public class WANTestBase extends JUnit4DistributedTestCase {
orderRegion =
(PartitionedRegion) cache.createRegionFactory(fact.create()).create(orderRegionName);
assertNotNull(orderRegion);
- LogWriterUtils.getLogWriter()
- .info("Partitioned Region ORDER created Successfully :" + orderRegion.toString());
+ logger.info("Partitioned Region ORDER created Successfully :" + orderRegion.toString());
paf = new PartitionAttributesFactory();
paf.setRedundantCopies(redundantCopies).setTotalNumBuckets(totalNumBuckets)
@@ -824,8 +838,7 @@ public class WANTestBase extends JUnit4DistributedTestCase {
shipmentRegion =
(PartitionedRegion) cache.createRegionFactory(fact.create()).create(shipmentRegionName);
assertNotNull(shipmentRegion);
- LogWriterUtils.getLogWriter()
- .info("Partitioned Region SHIPMENT created Successfully :" + shipmentRegion.toString());
+ logger.info("Partitioned Region SHIPMENT created Successfully :" + shipmentRegion.toString());
} finally {
exp.remove();
}
@@ -994,8 +1007,7 @@ public class WANTestBase extends JUnit4DistributedTestCase {
gemFireProps.setProperty(MCAST_PORT, "0");
gemFireProps.setProperty(LOCATORS, "localhost[" + locPort + "]");
- LogWriterUtils.getLogWriter()
- .info("Starting cache ds with following properties \n" + gemFireProps);
+ logger.info("Starting cache ds with following properties \n" + gemFireProps);
InternalDistributedSystem ds = test.getSystem(gemFireProps);
cache = CacheFactory.create(ds);
@@ -1795,7 +1807,7 @@ public class WANTestBase extends JUnit4DistributedTestCase {
} else {
persistentDirectory = new File(dsStore);
}
- LogWriterUtils.getLogWriter().info("The ds is : " + persistentDirectory.getName());
+ logger.info("The ds is : " + persistentDirectory.getName());
persistentDirectory.mkdir();
DiskStoreFactory dsf = cache.createDiskStoreFactory();
@@ -1817,11 +1829,11 @@ public class WANTestBase extends JUnit4DistributedTestCase {
gateway.setPersistenceEnabled(true);
String dsname = dsf.setDiskDirs(dirs1).create(dsName).getName();
gateway.setDiskStoreName(dsname);
- LogWriterUtils.getLogWriter().info("The DiskStoreName is : " + dsname);
+ logger.info("The DiskStoreName is : " + dsname);
} else {
DiskStore store = dsf.setDiskDirs(dirs1).create(dsName);
gateway.setDiskStoreName(store.getName());
- LogWriterUtils.getLogWriter().info("The ds is : " + store.getName());
+ logger.info("The ds is : " + store.getName());
}
gateway.setBatchConflationEnabled(isConflation);
gateway.create(dsName, remoteDsId);
@@ -1956,7 +1968,7 @@ public class WANTestBase extends JUnit4DistributedTestCase {
receiver.start();
fail("Expected GatewayReceiver Exception");
} catch (GatewayReceiverException gRE) {
- LogWriterUtils.getLogWriter().fine("Got the GatewayReceiverException", gRE);
+ logger.debug("Got the GatewayReceiverException", gRE);
assertTrue(gRE.getMessage().contains("Failed to create server socket on"));
} catch (IOException e) {
e.printStackTrace();
@@ -1986,8 +1998,7 @@ public class WANTestBase extends JUnit4DistributedTestCase {
gemFireProps.setProperty(MCAST_PORT, "0");
gemFireProps.setProperty(LOCATORS, "localhost[" + locPort + "]");
- LogWriterUtils.getLogWriter()
- .info("Starting cache ds with following properties \n" + gemFireProps);
+ logger.info("Starting cache ds with following properties \n" + gemFireProps);
InternalDistributedSystem ds = test.getSystem(gemFireProps);
cache = CacheFactory.create(ds);
@@ -2067,8 +2078,7 @@ public class WANTestBase extends JUnit4DistributedTestCase {
region = cache.createRegion(regionName, attrs);
region.registerInterest("ALL_KEYS");
assertNotNull(region);
- LogWriterUtils.getLogWriter()
- .info("Distributed Region " + regionName + " created Successfully :" + region.toString());
+ logger.info("Distributed Region " + regionName + " created Successfully :" + region.toString());
}
public static void createClientWithLocator(final int port0, final String host) {
@@ -2283,7 +2293,7 @@ public class WANTestBase extends JUnit4DistributedTestCase {
"putCustomerPartitionedRegion : failed while doing put operation in CustomerPartitionedRegion ",
e);
}
- LogWriterUtils.getLogWriter().info("Customer :- { " + custid + " : " + customer + " }");
+ logger.info("Customer :- { " + custid + " : " + customer + " }");
}
return custKeyValues;
}
@@ -2308,7 +2318,7 @@ public class WANTestBase extends JUnit4DistributedTestCase {
"putOrderPartitionedRegion : failed while doing put operation in OrderPartitionedRegion ",
e);
}
- LogWriterUtils.getLogWriter().info("Order :- { " + orderId + " : " + order + " }");
+ logger.info("Order :- { " + orderId + " : " + order + " }");
}
return orderKeyValues;
}
@@ -2331,7 +2341,7 @@ public class WANTestBase extends JUnit4DistributedTestCase {
"putOrderPartitionedRegionUsingCustId : failed while doing put operation in OrderPartitionedRegion ",
e);
}
- LogWriterUtils.getLogWriter().info("Order :- { " + custid + " : " + order + " }");
+ logger.info("Order :- { " + custid + " : " + order + " }");
}
return orderKeyValues;
}
@@ -2356,7 +2366,7 @@ public class WANTestBase extends JUnit4DistributedTestCase {
"updateOrderPartitionedRegion : failed while doing put operation in OrderPartitionedRegion ",
e);
}
- LogWriterUtils.getLogWriter().info("Order :- { " + orderId + " : " + order + " }");
+ logger.info("Order :- { " + orderId + " : " + order + " }");
}
return orderKeyValues;
}
@@ -2378,7 +2388,7 @@ public class WANTestBase extends JUnit4DistributedTestCase {
"updateOrderPartitionedRegionUsingCustId : failed while doing put operation in OrderPartitionedRegion ",
e);
}
- LogWriterUtils.getLogWriter().info("Order :- { " + custid + " : " + order + " }");
+ logger.info("Order :- { " + custid + " : " + order + " }");
}
return orderKeyValues;
}
@@ -2404,7 +2414,7 @@ public class WANTestBase extends JUnit4DistributedTestCase {
"putShipmentPartitionedRegion : failed while doing put operation in ShipmentPartitionedRegion ",
e);
}
- LogWriterUtils.getLogWriter().info("Shipment :- { " + shipmentId + " : " + shipment + " }");
+ logger.info("Shipment :- { " + shipmentId + " : " + shipment + " }");
}
return shipmentKeyValue;
}
@@ -2446,7 +2456,7 @@ public class WANTestBase extends JUnit4DistributedTestCase {
"putShipmentPartitionedRegionUsingCustId : failed while doing put operation in ShipmentPartitionedRegion ",
e);
}
- LogWriterUtils.getLogWriter().info("Shipment :- { " + custid + " : " + shipment + " }");
+ logger.info("Shipment :- { " + custid + " : " + shipment + " }");
}
return shipmentKeyValue;
}
@@ -2472,7 +2482,7 @@ public class WANTestBase extends JUnit4DistributedTestCase {
"updateShipmentPartitionedRegion : failed while doing put operation in ShipmentPartitionedRegion ",
e);
}
- LogWriterUtils.getLogWriter().info("Shipment :- { " + shipmentId + " : " + shipment + " }");
+ logger.info("Shipment :- { " + shipmentId + " : " + shipment + " }");
}
return shipmentKeyValue;
}
@@ -2494,7 +2504,7 @@ public class WANTestBase extends JUnit4DistributedTestCase {
"updateShipmentPartitionedRegionUsingCustId : failed while doing put operation in ShipmentPartitionedRegion ",
e);
}
- LogWriterUtils.getLogWriter().info("Shipment :- { " + custid + " : " + shipment + " }");
+ logger.info("Shipment :- { " + custid + " : " + shipment + " }");
}
return shipmentKeyValue;
}
@@ -2743,7 +2753,7 @@ public class WANTestBase extends JUnit4DistributedTestCase {
final Map eventsMap = ((MyAsyncEventListener) theListener).getEventsMap();
assertNotNull(eventsMap);
- LogWriterUtils.getLogWriter().info("The events map size is " + eventsMap.size());
+ logger.info("The events map size is " + eventsMap.size());
return eventsMap.size();
}
@@ -2757,7 +2767,7 @@ public class WANTestBase extends JUnit4DistributedTestCase {
+ r.keySet().size() + " present region keyset " + r.keySet(),
true, (regionSize <= r.keySet().size())));
for (int i = 0; i < regionSize; i++) {
- LogWriterUtils.getLogWriter().info("For Key : Key_" + i + " : Values : " + r.get("Key_" + i));
+ logger.info("For Key : Key_" + i + " : Values : " + r.get("Key_" + i));
assertEquals("keySet = " + r.keySet() + " values() = " + r.values(),
new SimpleClass(i, (byte) i), r.get("Key_" + i));
}
@@ -2845,8 +2855,8 @@ public class WANTestBase extends JUnit4DistributedTestCase {
boolean matchFlag = true;
for (Object key : keyValues.keySet()) {
if (!r.get(key).equals(keyValues.get(key))) {
- LogWriterUtils.getLogWriter().info("The values are for key " + " " + key + " "
- + r.get(key) + " in the map " + keyValues.get(key));
+ logger.info("The values are for key " + " " + key + " " + r.get(key) + " in the map "
+ + keyValues.get(key));
matchFlag = false;
}
}
@@ -2896,7 +2906,7 @@ public class WANTestBase extends JUnit4DistributedTestCase {
}
}
if (sender.isPrimary()) {
- LogWriterUtils.getLogWriter().info("Gateway sender is killed by a test");
+ logger.info("Gateway sender is killed by a test");
cache.getDistributedSystem().disconnect();
return Boolean.TRUE;
}
@@ -2909,10 +2919,10 @@ public class WANTestBase extends JUnit4DistributedTestCase {
}
public static void killSender() {
- LogWriterUtils.getLogWriter().info("Gateway sender is going to be killed by a test");
+ logger.info("Gateway sender is going to be killed by a test");
cache.close();
cache.getDistributedSystem().disconnect();
- LogWriterUtils.getLogWriter().info("Gateway sender is killed by a test");
+ logger.info("Gateway sender is killed by a test");
}
public static void checkAllSiteMetaData(
@@ -3723,4 +3733,303 @@ public class WANTestBase extends JUnit4DistributedTestCase {
public boolean isOffHeap() {
return false;
}
+
+ /**
+ * Checks whether a Async Queue MBean is created or not
+ *
+ * @param vm reference to VM
+ */
+ @SuppressWarnings("serial")
+ public static void checkAsyncQueueMBean(final VM vm, final boolean shouldExist) {
+ SerializableRunnable checkAsyncQueueMBean =
+ new SerializableRunnable("Check Async Queue MBean") {
+ public void run() {
+ ManagementService service = ManagementService.getManagementService(cache);
+ AsyncEventQueueMXBean bean = service.getLocalAsyncEventQueueMXBean("pn");
+ if (shouldExist) {
+ assertNotNull(bean);
+ } else {
+ assertNull(bean);
+ }
+ }
+ };
+ vm.invoke(checkAsyncQueueMBean);
+ }
+
+ /**
+ * Checks Proxy GatewayReceiver
+ *
+ * @param vm reference to VM
+ */
+ @SuppressWarnings("serial")
+ public static void checkProxyReceiver(final VM vm, final DistributedMember senderMember) {
+ SerializableRunnable checkProxySender = new SerializableRunnable("Check Proxy Receiver") {
+ public void run() {
+ ManagementService service = ManagementService.getManagementService(cache);
+ GatewayReceiverMXBean bean = null;
+ try {
+ bean = MBeanUtil.getGatewayReceiverMbeanProxy(senderMember);
+ } catch (Exception e) {
+ fail("Could not obtain Sender Proxy in desired time " + e);
+ }
+ assertNotNull(bean);
+ final ObjectName receiverMBeanName = service.getGatewayReceiverMBeanName(senderMember);
+ try {
+ MBeanUtil.printBeanDetails(receiverMBeanName);
+ } catch (Exception e) {
+ fail("Error while Printing Bean Details " + e);
+ }
+
+ }
+ };
+ vm.invoke(checkProxySender);
+ }
+
+ /**
+ * Checks Proxy GatewaySender
+ *
+ * @param vm reference to VM
+ */
+ @SuppressWarnings("serial")
+ public static void checkProxySender(final VM vm, final DistributedMember senderMember) {
+ SerializableRunnable checkProxySender = new SerializableRunnable("Check Proxy Sender") {
+ public void run() {
+ ManagementService service = ManagementService.getManagementService(cache);
+ GatewaySenderMXBean bean = null;
+ try {
+ bean = MBeanUtil.getGatewaySenderMbeanProxy(senderMember, "pn");
+ } catch (Exception e) {
+ fail("Could not obtain Sender Proxy in desired time " + e);
+ }
+ assertNotNull(bean);
+ final ObjectName senderMBeanName = service.getGatewaySenderMBeanName(senderMember, "pn");
+ try {
+ MBeanUtil.printBeanDetails(senderMBeanName);
+ } catch (Exception e) {
+ fail("Error while Printing Bean Details " + e);
+ }
+
+ if (service.isManager()) {
+ DistributedSystemMXBean dsBean = service.getDistributedSystemMXBean();
+ Awaitility.await().atMost(1, TimeUnit.MINUTES).until(() -> {
+ Map<String, Boolean> dsMap = dsBean.viewRemoteClusterStatus();
+ dsMap.entrySet().stream()
+ .forEach(entry -> assertTrue("Should be true " + entry.getKey(), entry.getValue()));
+ });
+ }
+
+ }
+ };
+ vm.invoke(checkProxySender);
+ }
+
+ /**
+ * Checks whether a GatewayReceiverMBean is created or not
+ *
+ * @param vm reference to VM
+ */
+ @SuppressWarnings("serial")
+ public static void checkReceiverMBean(final VM vm) {
+ SerializableRunnable checkMBean = new SerializableRunnable("Check Receiver MBean") {
+ public void run() {
+ ManagementService service = ManagementService.getManagementService(cache);
+ GatewayReceiverMXBean bean = service.getLocalGatewayReceiverMXBean();
+ assertNotNull(bean);
+ }
+ };
+ vm.invoke(checkMBean);
+ }
+
+ @SuppressWarnings("serial")
+ public static void checkReceiverNavigationAPIS(final VM vm,
+ final DistributedMember receiverMember) {
+ SerializableRunnable checkNavigationAPIS =
+ new SerializableRunnable("Check Receiver Navigation APIs") {
+ public void run() {
+ ManagementService service = ManagementService.getManagementService(cache);
+ DistributedSystemMXBean bean = service.getDistributedSystemMXBean();
+ ObjectName expectedName = service.getGatewayReceiverMBeanName(receiverMember);
+ try {
+ ObjectName actualName = bean.fetchGatewayReceiverObjectName(receiverMember.getId());
+ assertEquals(expectedName, actualName);
+ } catch (Exception e) {
+ fail("Receiver Navigation Failed " + e);
+ }
+
+ assertEquals(1, bean.listGatewayReceiverObjectNames().length);
+
+ }
+ };
+ vm.invoke(checkNavigationAPIS);
+ }
+
+ /**
+ * Checks whether a GatewayReceiverMBean is created or not
+ *
+ * @param vm reference to VM
+ */
+ @SuppressWarnings("serial")
+ public static void checkSenderMBean(final VM vm, final String regionPath, boolean connected) {
+ SerializableRunnable checkMBean = new SerializableRunnable("Check Sender MBean") {
+ public void run() {
+ ManagementService service = ManagementService.getManagementService(cache);
+
+ GatewaySenderMXBean bean = service.getLocalGatewaySenderMXBean("pn");
+ assertNotNull(bean);
+ Awaitility.await().atMost(1, TimeUnit.MINUTES)
+ .until(() -> assertEquals(connected, bean.isConnected()));
+
+ ObjectName regionBeanName = service.getRegionMBeanName(
+ cache.getDistributedSystem().getDistributedMember(), "/" + regionPath);
+ RegionMXBean rBean = service.getMBeanInstance(regionBeanName, RegionMXBean.class);
+ assertTrue(rBean.isGatewayEnabled());
+
+
+ }
+ };
+ vm.invoke(checkMBean);
+ }
+
+ @SuppressWarnings("serial")
+ public static void checkSenderNavigationAPIS(final VM vm, final DistributedMember senderMember) {
+ SerializableRunnable checkNavigationAPIS =
+ new SerializableRunnable("Check Sender Navigation APIs") {
+ public void run() {
+ ManagementService service = ManagementService.getManagementService(cache);
+ DistributedSystemMXBean bean = service.getDistributedSystemMXBean();
+ ObjectName expectedName = service.getGatewaySenderMBeanName(senderMember, "pn");
+ try {
+ ObjectName actualName = bean.fetchGatewaySenderObjectName(senderMember.getId(), "pn");
+ assertEquals(expectedName, actualName);
+ } catch (Exception e) {
+ fail("Sender Navigation Failed " + e);
+ }
+
+ assertEquals(2, bean.listGatewaySenderObjectNames().length);
+ try {
+ assertEquals(1, bean.listGatewaySenderObjectNames(senderMember.getId()).length);
+ } catch (Exception e) {
+ fail("Sender Navigation Failed " + e);
+ }
+
+ }
+ };
+ vm.invoke(checkNavigationAPIS);
+ }
+
+ /**
+ * start a gateway sender
+ *
+ * @param vm reference to VM
+ */
+ @SuppressWarnings("serial")
+ public static void startGatewaySender(final VM vm) {
+ SerializableRunnable stopGatewaySender = new SerializableRunnable("Start Gateway Sender") {
+ public void run() {
+ ManagementService service = ManagementService.getManagementService(cache);
+ GatewaySenderMXBean bean = service.getLocalGatewaySenderMXBean("pn");
+ assertNotNull(bean);
+ bean.start();
+ assertTrue(bean.isRunning());
+ }
+ };
+ vm.invoke(stopGatewaySender);
+ }
+
+ /**
+ * stops a gateway sender
+ *
+ * @param vm reference to VM
+ */
+ @SuppressWarnings("serial")
+ public static void stopGatewaySender(final VM vm) {
+ SerializableRunnable stopGatewaySender = new SerializableRunnable("Stop Gateway Sender") {
+ public void run() {
+ ManagementService service = ManagementService.getManagementService(cache);
+ GatewaySenderMXBean bean = service.getLocalGatewaySenderMXBean("pn");
+ assertNotNull(bean);
+ bean.stop();
+ assertFalse(bean.isRunning());
+ }
+ };
+ vm.invoke(stopGatewaySender);
+ }
+
+ /**
+ * Checks Proxy Async Queue
+ *
+ * @param vm reference to VM
+ */
+ @SuppressWarnings("serial")
+ public static void checkProxyAsyncQueue(final VM vm, final DistributedMember senderMember,
+ final boolean shouldExist) {
+ SerializableRunnable checkProxyAsyncQueue =
+ new SerializableRunnable("Check Proxy Async Queue") {
+ public void run() {
+ SystemManagementService service =
+ (SystemManagementService) ManagementService.getManagementService(cache);
+ final ObjectName queueMBeanName =
+ service.getAsyncEventQueueMBeanName(senderMember, "pn");
+ AsyncEventQueueMXBean bean = null;
+ if (shouldExist) {
+ // Verify the MBean proxy exists
+ try {
+ bean = MBeanUtil.getAsyncEventQueueMBeanProxy(senderMember, "pn");
+ } catch (Exception e) {
+ fail("Could not obtain Sender Proxy in desired time " + e);
+ }
+ assertNotNull(bean);
+
+ try {
+ MBeanUtil.printBeanDetails(queueMBeanName);
+ } catch (Exception e) {
+ fail("Error while Printing Bean Details " + e);
+ }
+ } else {
+ // Verify the MBean proxy doesn't exist
+ bean = service.getMBeanProxy(queueMBeanName, AsyncEventQueueMXBean.class);
+ assertNull(bean);
+ }
+ }
+ };
+ vm.invoke(checkProxyAsyncQueue);
+ }
+
+ public static DistributedMember getMember() {
+ return ((GemFireCacheImpl) cache).getMyId();
+ }
+
+ public static ManagementService getManagementService() {
+ return ManagementService.getManagementService(cache);
+ }
+
+ /**
+ * Checks Proxy GatewaySender
+ *
+ * @param vm reference to VM
+ */
+ @SuppressWarnings("serial")
+ public static void checkRemoteClusterStatus(final VM vm, final DistributedMember senderMember) {
+ SerializableRunnable checkProxySender = new SerializableRunnable("DS Map Size") {
+ public void run() {
+ Awaitility.await().atMost(120, TimeUnit.SECONDS).until(() -> {
+ final ManagementService service = ManagementService.getManagementService(cache);
+ final DistributedSystemMXBean dsBean = service.getDistributedSystemMXBean();
+ assertEquals(
+ "Failed while waiting for getDistributedSystemMXBean to complete and get results",
+ true, dsBean != null);
+ });
+ ManagementService service = ManagementService.getManagementService(cache);
+ final DistributedSystemMXBean dsBean = service.getDistributedSystemMXBean();
+ assertNotNull(dsBean);
+ Map<String, Boolean> dsMap = dsBean.viewRemoteClusterStatus();
+ logger.info("Ds Map is: " + dsMap.size());
+ assertNotNull(dsMap);
+ assertEquals(true, dsMap.size() > 0);
+ }
+ };
+ vm.invoke(checkProxySender);
+ }
+
+
}
diff --git a/geode-wan/src/test/java/org/apache/geode/management/WANManagementDUnitTest.java b/geode-wan/src/test/java/org/apache/geode/management/WANManagementDUnitTest.java
index 24c782d..52121bc 100644
--- a/geode-wan/src/test/java/org/apache/geode/management/WANManagementDUnitTest.java
+++ b/geode-wan/src/test/java/org/apache/geode/management/WANManagementDUnitTest.java
@@ -15,30 +15,18 @@
package org.apache.geode.management;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-import java.util.Map;
import java.util.concurrent.TimeUnit;
-import javax.management.ObjectName;
-
import org.awaitility.Awaitility;
import org.junit.Test;
import org.junit.experimental.categories.Category;
-import org.apache.geode.cache.Cache;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.Locator;
-import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.cache.wan.WANTestBase;
-import org.apache.geode.management.internal.SystemManagementService;
import org.apache.geode.test.dunit.Host;
-import org.apache.geode.test.dunit.SerializableRunnable;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.junit.categories.DistributedTest;
import org.apache.geode.test.junit.categories.FlakyTest;
@@ -101,23 +89,23 @@ public class WANManagementDUnitTest extends ManagementTestBase {
puneSender.invoke(() -> WANTestBase.waitForSenderRunningState("pn"));
managing.invoke(() -> WANTestBase.waitForSenderRunningState("pn"));
- checkSenderMBean(puneSender, getTestMethodName() + "_PR", true);
- checkSenderMBean(managing, getTestMethodName() + "_PR", true);
+ WANTestBase.checkSenderMBean(puneSender, getTestMethodName() + "_PR", true);
+ WANTestBase.checkSenderMBean(managing, getTestMethodName() + "_PR", true);
- checkReceiverMBean(nyReceiver);
+ WANTestBase.checkReceiverMBean(nyReceiver);
- stopGatewaySender(puneSender);
- startGatewaySender(puneSender);
+ WANTestBase.stopGatewaySender(puneSender);
+ WANTestBase.startGatewaySender(puneSender);
- DistributedMember puneMember = puneSender.invoke(() -> WANManagementDUnitTest.getMember());
+ DistributedMember puneMember = puneSender.invoke(() -> WANTestBase.getMember());
- checkProxySender(managing, puneMember);
- checkSenderNavigationAPIS(managing, puneMember);
+ WANTestBase.checkProxySender(managing, puneMember);
+ WANTestBase.checkSenderNavigationAPIS(managing, puneMember);
nyReceiver.invoke(() -> WANTestBase.stopReceivers());
- checkSenderMBean(puneSender, getTestMethodName() + "_PR", false);
- checkSenderMBean(managing, getTestMethodName() + "_PR", false);
+ WANTestBase.checkSenderMBean(puneSender, getTestMethodName() + "_PR", false);
+ WANTestBase.checkSenderMBean(managing, getTestMethodName() + "_PR", false);
}
@Category(FlakyTest.class) // GEODE-1603
@@ -157,13 +145,13 @@ public class WANManagementDUnitTest extends ManagementTestBase {
managing.invoke(() -> WANTestBase.createManagementCache(nyPort));
startManagingNode(managing);
- checkSenderMBean(puneSender, getTestMethodName() + "_PR", true);
- checkReceiverMBean(nyReceiver);
+ WANTestBase.checkSenderMBean(puneSender, getTestMethodName() + "_PR", true);
+ WANTestBase.checkReceiverMBean(nyReceiver);
- DistributedMember nyMember = nyReceiver.invoke(() -> WANManagementDUnitTest.getMember());
+ DistributedMember nyMember = nyReceiver.invoke(() -> WANTestBase.getMember());
- checkProxyReceiver(managing, nyMember);
- checkReceiverNavigationAPIS(managing, nyMember);
+ WANTestBase.checkProxyReceiver(managing, nyMember);
+ WANTestBase.checkReceiverNavigationAPIS(managing, nyMember);
}
@@ -199,12 +187,12 @@ public class WANManagementDUnitTest extends ManagementTestBase {
1, 100, false));
nyReceiver.invoke(() -> WANTestBase.createReceiver());
- checkAsyncQueueMBean(puneSender, true);
- checkAsyncQueueMBean(managing, true);
+ WANTestBase.checkAsyncQueueMBean(puneSender, true);
+ WANTestBase.checkAsyncQueueMBean(managing, true);
- DistributedMember puneMember = puneSender.invoke(() -> WANManagementDUnitTest.getMember());
+ DistributedMember puneMember = puneSender.invoke(() -> WANTestBase.getMember());
- checkProxyAsyncQueue(managing, puneMember, true);
+ WANTestBase.checkProxyAsyncQueue(managing, puneMember, true);
}
@@ -228,19 +216,19 @@ public class WANManagementDUnitTest extends ManagementTestBase {
() -> WANTestBase.createAsyncEventQueue(aeqId, false, 100, 100, false, false, null, false));
// Verify AsyncEventQueueMXBean exists
- checkAsyncQueueMBean(memberVM, true);
- checkAsyncQueueMBean(managerVm, true);
- DistributedMember member = memberVM.invoke(() -> WANManagementDUnitTest.getMember());
- checkProxyAsyncQueue(managerVm, member, true);
+ WANTestBase.checkAsyncQueueMBean(memberVM, true);
+ WANTestBase.checkAsyncQueueMBean(managerVm, true);
+ DistributedMember member = memberVM.invoke(() -> WANTestBase.getMember());
+ WANTestBase.checkProxyAsyncQueue(managerVm, member, true);
// Destroy AsyncEventQueue
memberVM.invoke(() -> WANTestBase.destroyAsyncEventQueue(aeqId));
managerVm.invoke(() -> WANTestBase.destroyAsyncEventQueue(aeqId));
// Verify AsyncEventQueueMXBean no longer exists
- checkAsyncQueueMBean(memberVM, false);
- checkAsyncQueueMBean(managerVm, false);
- checkProxyAsyncQueue(managerVm, member, false);
+ WANTestBase.checkAsyncQueueMBean(memberVM, false);
+ WANTestBase.checkAsyncQueueMBean(managerVm, false);
+ WANTestBase.checkProxyAsyncQueue(managerVm, member, false);
}
@Test
@@ -263,9 +251,7 @@ public class WANManagementDUnitTest extends ManagementTestBase {
String regionPath = "/" + regionName;
managing.invoke(() -> {
- Cache cache = GemFireCacheImpl.getInstance();
- ManagementService service = ManagementService.getManagementService(cache);
-
+ ManagementService service = WANTestBase.getManagementService();
Awaitility.await().atMost(5, TimeUnit.SECONDS)
.until(() -> assertNotNull(service.getDistributedRegionMXBean(regionPath)));
@@ -274,283 +260,61 @@ public class WANManagementDUnitTest extends ManagementTestBase {
});
}
- @SuppressWarnings("serial")
- protected void checkSenderNavigationAPIS(final VM vm, final DistributedMember senderMember) {
- SerializableRunnable checkNavigationAPIS =
- new SerializableRunnable("Check Sender Navigation APIs") {
- public void run() {
- Cache cache = GemFireCacheImpl.getInstance();
- ManagementService service = ManagementService.getManagementService(cache);
- DistributedSystemMXBean bean = service.getDistributedSystemMXBean();
- ObjectName expectedName = service.getGatewaySenderMBeanName(senderMember, "pn");
- try {
- ObjectName actualName = bean.fetchGatewaySenderObjectName(senderMember.getId(), "pn");
- assertEquals(expectedName, actualName);
- } catch (Exception e) {
- fail("Sender Navigation Failed " + e);
- }
-
- assertEquals(2, bean.listGatewaySenderObjectNames().length);
- try {
- assertEquals(1, bean.listGatewaySenderObjectNames(senderMember.getId()).length);
- } catch (Exception e) {
- fail("Sender Navigation Failed " + e);
- }
-
- }
- };
- vm.invoke(checkNavigationAPIS);
- }
+ @Category(FlakyTest.class) // GEODE-1629
+ @Test
+ public void testMBeanCallbackInRemoteCluster() throws Exception {
- @SuppressWarnings("serial")
- protected void checkReceiverNavigationAPIS(final VM vm, final DistributedMember receiverMember) {
- SerializableRunnable checkNavigationAPIS =
- new SerializableRunnable("Check Receiver Navigation APIs") {
- public void run() {
- Cache cache = GemFireCacheImpl.getInstance();
- ManagementService service = ManagementService.getManagementService(cache);
- DistributedSystemMXBean bean = service.getDistributedSystemMXBean();
- ObjectName expectedName = service.getGatewayReceiverMBeanName(receiverMember);
- try {
- ObjectName actualName = bean.fetchGatewayReceiverObjectName(receiverMember.getId());
- assertEquals(expectedName, actualName);
- } catch (Exception e) {
- fail("Receiver Navigation Failed " + e);
- }
-
- assertEquals(1, bean.listGatewayReceiverObjectNames().length);
-
- }
- };
- vm.invoke(checkNavigationAPIS);
- }
+ VM nyLocator = getManagedNodeList().get(0);
+ VM nyReceiver = getManagedNodeList().get(1);
+ VM puneSender = getManagedNodeList().get(2);
+ VM managing = getManagingNode();
+ VM puneLocator = Host.getLocator();
- private static int getLocatorPort() {
- return Locator.getLocators().get(0).getPort();
- }
+ int dsIdPort = puneLocator.invoke(() -> WANManagementDUnitTest.getLocatorPort());
- private static DistributedMember getMember() {
- return GemFireCacheImpl.getInstance().getMyId();
- }
+ Integer nyPort = nyLocator.invoke(() -> WANTestBase.createFirstRemoteLocator(12, dsIdPort));
- /**
- * Checks Proxy GatewaySender
- *
- * @param vm reference to VM
- */
- @SuppressWarnings("serial")
- protected void checkProxySender(final VM vm, final DistributedMember senderMember) {
- SerializableRunnable checkProxySender = new SerializableRunnable("Check Proxy Sender") {
- public void run() {
- Cache cache = GemFireCacheImpl.getInstance();
- ManagementService service = ManagementService.getManagementService(cache);
- GatewaySenderMXBean bean = null;
- try {
- bean = MBeanUtil.getGatewaySenderMbeanProxy(senderMember, "pn");
- } catch (Exception e) {
- fail("Could not obtain Sender Proxy in desired time " + e);
- }
- assertNotNull(bean);
- final ObjectName senderMBeanName = service.getGatewaySenderMBeanName(senderMember, "pn");
- try {
- MBeanUtil.printBeanDetails(senderMBeanName);
- } catch (Exception e) {
- fail("Error while Printing Bean Details " + e);
- }
-
- if (service.isManager()) {
- DistributedSystemMXBean dsBean = service.getDistributedSystemMXBean();
- Awaitility.await().atMost(1, TimeUnit.MINUTES).until(() -> {
- Map<String, Boolean> dsMap = dsBean.viewRemoteClusterStatus();
- dsMap.entrySet().stream()
- .forEach(entry -> assertTrue("Should be true " + entry.getKey(), entry.getValue()));
- });
- }
-
- }
- };
- vm.invoke(checkProxySender);
- }
+ puneSender.invoke(() -> WANTestBase.createCache(dsIdPort));
+ managing.invoke(() -> WANTestBase.createManagementCache(dsIdPort));
+ startManagingNode(managing);
- /**
- * Checks Proxy GatewayReceiver
- *
- * @param vm reference to VM
- */
- @SuppressWarnings("serial")
- protected void checkProxyReceiver(final VM vm, final DistributedMember senderMember) {
- SerializableRunnable checkProxySender = new SerializableRunnable("Check Proxy Receiver") {
- public void run() {
- Cache cache = GemFireCacheImpl.getInstance();
- ManagementService service = ManagementService.getManagementService(cache);
- GatewayReceiverMXBean bean = null;
- try {
- bean = MBeanUtil.getGatewayReceiverMbeanProxy(senderMember);
- } catch (Exception e) {
- fail("Could not obtain Sender Proxy in desired time " + e);
- }
- assertNotNull(bean);
- final ObjectName receiverMBeanName = service.getGatewayReceiverMBeanName(senderMember);
- try {
- MBeanUtil.printBeanDetails(receiverMBeanName);
- } catch (Exception e) {
- fail("Error while Printing Bean Details " + e);
- }
-
- }
- };
- vm.invoke(checkProxySender);
- }
+ puneSender
+ .invoke(() -> WANTestBase.createSender("pn", 12, true, 100, 300, false, false, null, true));
+ managing
+ .invoke(() -> WANTestBase.createSender("pn", 12, true, 100, 300, false, false, null, true));
+ puneSender.invoke(() -> WANTestBase.createPartitionedRegion(getTestMethodName() + "_PR", "pn",
+ 1, 100, false));
+ managing.invoke(() -> WANTestBase.createPartitionedRegion(getTestMethodName() + "_PR", "pn", 1,
+ 100, false));
- /**
- * stops a gateway sender
- *
- * @param vm reference to VM
- */
- @SuppressWarnings("serial")
- protected void stopGatewaySender(final VM vm) {
- SerializableRunnable stopGatewaySender = new SerializableRunnable("Stop Gateway Sender") {
- public void run() {
- Cache cache = GemFireCacheImpl.getInstance();
- ManagementService service = ManagementService.getManagementService(cache);
- GatewaySenderMXBean bean = service.getLocalGatewaySenderMXBean("pn");
- assertNotNull(bean);
- bean.stop();
- assertFalse(bean.isRunning());
- }
- };
- vm.invoke(stopGatewaySender);
- }
+ WANTestBase.createCacheInVMs(nyPort, nyReceiver);
+ nyReceiver.invoke(() -> WANTestBase.createReceiver());
+ nyReceiver.invoke(() -> WANTestBase.createPartitionedRegion(getTestMethodName() + "_PR", null,
+ 1, 100, false));
- /**
- * start a gateway sender
- *
- * @param vm reference to VM
- */
- @SuppressWarnings("serial")
- protected void startGatewaySender(final VM vm) {
- SerializableRunnable stopGatewaySender = new SerializableRunnable("Start Gateway Sender") {
- public void run() {
- Cache cache = GemFireCacheImpl.getInstance();
- ManagementService service = ManagementService.getManagementService(cache);
- GatewaySenderMXBean bean = service.getLocalGatewaySenderMXBean("pn");
- assertNotNull(bean);
- bean.start();
- assertTrue(bean.isRunning());
- }
- };
- vm.invoke(stopGatewaySender);
- }
+ WANTestBase.startSenderInVMs("pn", puneSender, managing);
+ // make sure all the senders are running before doing any puts
+ puneSender.invoke(() -> WANTestBase.waitForSenderRunningState("pn"));
+ managing.invoke(() -> WANTestBase.waitForSenderRunningState("pn"));
- /**
- * Checks whether a GatewayReceiverMBean is created or not
- *
- * @param vm reference to VM
- */
- @SuppressWarnings("serial")
- protected void checkReceiverMBean(final VM vm) {
- SerializableRunnable checkMBean = new SerializableRunnable("Check Receiver MBean") {
- public void run() {
- Cache cache = GemFireCacheImpl.getInstance();
- ManagementService service = ManagementService.getManagementService(cache);
- GatewayReceiverMXBean bean = service.getLocalGatewayReceiverMXBean();
- assertNotNull(bean);
- }
- };
- vm.invoke(checkMBean);
- }
+ WANTestBase.checkSenderMBean(puneSender, getTestMethodName() + "_PR", true);
+ WANTestBase.checkSenderMBean(managing, getTestMethodName() + "_PR", true);
- /**
- * Checks whether a GatewayReceiverMBean is created or not
- *
- * @param vm reference to VM
- */
- @SuppressWarnings("serial")
- protected void checkSenderMBean(final VM vm, final String regionPath, boolean connected) {
- SerializableRunnable checkMBean = new SerializableRunnable("Check Sender MBean") {
- public void run() {
- Cache cache = GemFireCacheImpl.getInstance();
- ManagementService service = ManagementService.getManagementService(cache);
-
- GatewaySenderMXBean bean = service.getLocalGatewaySenderMXBean("pn");
- assertNotNull(bean);
- Awaitility.await().atMost(1, TimeUnit.MINUTES)
- .until(() -> assertEquals(connected, bean.isConnected()));
-
- ObjectName regionBeanName = service.getRegionMBeanName(
- cache.getDistributedSystem().getDistributedMember(), "/" + regionPath);
- RegionMXBean rBean = service.getMBeanInstance(regionBeanName, RegionMXBean.class);
- assertTrue(rBean.isGatewayEnabled());
-
-
- }
- };
- vm.invoke(checkMBean);
- }
+ WANTestBase.checkReceiverMBean(nyReceiver);
+
+ WANTestBase.stopGatewaySender(puneSender);
+ WANTestBase.startGatewaySender(puneSender);
+
+ DistributedMember puneMember = puneSender.invoke(() -> WANTestBase.getMember());
+
+ WANTestBase.checkRemoteClusterStatus(managing, puneMember);
- /**
- * Checks whether a Async Queue MBean is created or not
- *
- * @param vm reference to VM
- */
- @SuppressWarnings("serial")
- protected void checkAsyncQueueMBean(final VM vm, final boolean shouldExist) {
- SerializableRunnable checkAsyncQueueMBean =
- new SerializableRunnable("Check Async Queue MBean") {
- public void run() {
- Cache cache = GemFireCacheImpl.getInstance();
- ManagementService service = ManagementService.getManagementService(cache);
- AsyncEventQueueMXBean bean = service.getLocalAsyncEventQueueMXBean("pn");
- if (shouldExist) {
- assertNotNull(bean);
- } else {
- assertNull(bean);
- }
- }
- };
- vm.invoke(checkAsyncQueueMBean);
}
- /**
- * Checks Proxy Async Queue
- *
- * @param vm reference to VM
- */
- @SuppressWarnings("serial")
- protected void checkProxyAsyncQueue(final VM vm, final DistributedMember senderMember,
- final boolean shouldExist) {
- SerializableRunnable checkProxyAsyncQueue =
- new SerializableRunnable("Check Proxy Async Queue") {
- public void run() {
- Cache cache = GemFireCacheImpl.getInstance();
- SystemManagementService service =
- (SystemManagementService) ManagementService.getManagementService(cache);
- final ObjectName queueMBeanName =
- service.getAsyncEventQueueMBeanName(senderMember, "pn");
- AsyncEventQueueMXBean bean = null;
- if (shouldExist) {
- // Verify the MBean proxy exists
- try {
- bean = MBeanUtil.getAsyncEventQueueMBeanProxy(senderMember, "pn");
- } catch (Exception e) {
- fail("Could not obtain Sender Proxy in desired time " + e);
- }
- assertNotNull(bean);
-
- try {
- MBeanUtil.printBeanDetails(queueMBeanName);
- } catch (Exception e) {
- fail("Error while Printing Bean Details " + e);
- }
- } else {
- // Verify the MBean proxy doesn't exist
- bean = service.getMBeanProxy(queueMBeanName, AsyncEventQueueMXBean.class);
- assertNull(bean);
- }
- }
- };
- vm.invoke(checkProxyAsyncQueue);
+ private static int getLocatorPort() {
+ return Locator.getLocators().get(0).getPort();
}
+
}
diff --git a/geode-wan/src/test/java/org/apache/geode/management/internal/pulse/TestRemoteClusterDUnitTest.java b/geode-wan/src/test/java/org/apache/geode/management/internal/pulse/TestRemoteClusterDUnitTest.java
deleted file mode 100644
index b55b792..0000000
--- a/geode-wan/src/test/java/org/apache/geode/management/internal/pulse/TestRemoteClusterDUnitTest.java
+++ /dev/null
@@ -1,250 +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.pulse;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
-
-import java.util.Map;
-
-import javax.management.MBeanServer;
-import javax.management.ObjectName;
-
-import org.junit.Test;
-import org.junit.experimental.categories.Category;
-
-import org.apache.geode.cache.Cache;
-import org.apache.geode.distributed.DistributedMember;
-import org.apache.geode.distributed.Locator;
-import org.apache.geode.internal.cache.GemFireCacheImpl;
-import org.apache.geode.internal.cache.wan.WANTestBase;
-import org.apache.geode.management.DistributedSystemMXBean;
-import org.apache.geode.management.GatewayReceiverMXBean;
-import org.apache.geode.management.GatewaySenderMXBean;
-import org.apache.geode.management.ManagementService;
-import org.apache.geode.management.ManagementTestBase;
-import org.apache.geode.management.RegionMXBean;
-import org.apache.geode.management.internal.MBeanJMXAdapter;
-import org.apache.geode.test.dunit.Host;
-import org.apache.geode.test.dunit.LogWriterUtils;
-import org.apache.geode.test.dunit.SerializableRunnable;
-import org.apache.geode.test.dunit.VM;
-import org.apache.geode.test.dunit.Wait;
-import org.apache.geode.test.dunit.WaitCriterion;
-import org.apache.geode.test.junit.categories.DistributedTest;
-import org.apache.geode.test.junit.categories.FlakyTest;
-
-/**
- * This is for testing remote Cluster
- *
- *
- */
-
-@Category(DistributedTest.class)
-public class TestRemoteClusterDUnitTest extends ManagementTestBase {
-
- private static final long serialVersionUID = 1L;
-
- public static MBeanServer mbeanServer = MBeanJMXAdapter.mbeanServer;
-
- public TestRemoteClusterDUnitTest() throws Exception {
- super();
- }
-
- @Category(FlakyTest.class) // GEODE-1629
- @Test
- public void testMBeanCallback() throws Exception {
-
- VM nyLocator = getManagedNodeList().get(0);
- VM nyReceiver = getManagedNodeList().get(1);
- VM puneSender = getManagedNodeList().get(2);
- VM managing = getManagingNode();
- VM puneLocator = Host.getLocator();
-
- int dsIdPort = puneLocator.invoke(() -> TestRemoteClusterDUnitTest.getLocatorPort());
-
- Integer nyPort = nyLocator.invoke(() -> WANTestBase.createFirstRemoteLocator(12, dsIdPort));
-
- puneSender.invoke(() -> WANTestBase.createCache(dsIdPort));
- managing.invoke(() -> WANTestBase.createManagementCache(dsIdPort));
- startManagingNode(managing);
-
- puneSender
- .invoke(() -> WANTestBase.createSender("pn", 12, true, 100, 300, false, false, null, true));
- managing
- .invoke(() -> WANTestBase.createSender("pn", 12, true, 100, 300, false, false, null, true));
-
- puneSender.invoke(() -> WANTestBase.createPartitionedRegion(getTestMethodName() + "_PR", "pn",
- 1, 100, false));
- managing.invoke(() -> WANTestBase.createPartitionedRegion(getTestMethodName() + "_PR", "pn", 1,
- 100, false));
-
- WANTestBase.createCacheInVMs(nyPort, nyReceiver);
- nyReceiver.invoke(() -> WANTestBase.createReceiver());
- nyReceiver.invoke(() -> WANTestBase.createPartitionedRegion(getTestMethodName() + "_PR", null,
- 1, 100, false));
-
- WANTestBase.startSenderInVMs("pn", puneSender, managing);
-
- // make sure all the senders are running before doing any puts
- puneSender.invoke(() -> WANTestBase.waitForSenderRunningState("pn"));
- managing.invoke(() -> WANTestBase.waitForSenderRunningState("pn"));
-
- checkSenderMBean(puneSender, getTestMethodName() + "_PR");
- checkSenderMBean(managing, getTestMethodName() + "_PR");
-
- checkReceiverMBean(nyReceiver);
-
- stopGatewaySender(puneSender);
- startGatewaySender(puneSender);
-
- DistributedMember puneMember = puneSender.invoke(() -> TestRemoteClusterDUnitTest.getMember());
-
- checkRemoteClusterStatus(managing, puneMember);
-
- }
-
- private static int getLocatorPort() {
- return Locator.getLocators().get(0).getPort();
- }
-
- private static DistributedMember getMember() {
- return GemFireCacheImpl.getInstance().getMyId();
- }
-
- /**
- * Checks Proxy GatewaySender
- *
- * @param vm reference to VM
- */
- @SuppressWarnings("serial")
- protected void checkRemoteClusterStatus(final VM vm, final DistributedMember senderMember) {
- SerializableRunnable checkProxySender = new SerializableRunnable("DS Map Size") {
- public void run() {
- Cache cache = GemFireCacheImpl.getInstance();
- final WaitCriterion waitCriteria2 = new WaitCriterion() {
- @Override
- public boolean done() {
- Cache cache = GemFireCacheImpl.getInstance();
- final ManagementService service = ManagementService.getManagementService(cache);
- final DistributedSystemMXBean dsBean = service.getDistributedSystemMXBean();
- return dsBean != null;
- }
-
- @Override
- public String description() {
- return "wait for getDistributedSystemMXBean to complete and get results";
- }
- };
- Wait.waitForCriterion(waitCriteria2, 2 * 60 * 1000, 5000, true);
- ManagementService service = ManagementService.getManagementService(cache);
- final DistributedSystemMXBean dsBean = service.getDistributedSystemMXBean();
- assertNotNull(dsBean);
- Map<String, Boolean> dsMap = dsBean.viewRemoteClusterStatus();
- LogWriterUtils.getLogWriter().info("Ds Map is: " + dsMap.size());
- assertNotNull(dsMap);
- assertEquals(true, dsMap.size() > 0);
- }
- };
- vm.invoke(checkProxySender);
- }
-
-
- /**
- * stops a gateway sender
- *
- * @param vm reference to VM
- */
- @SuppressWarnings("serial")
- protected void stopGatewaySender(final VM vm) {
- SerializableRunnable stopGatewaySender = new SerializableRunnable("Stop Gateway Sender") {
- public void run() {
- Cache cache = GemFireCacheImpl.getInstance();
- ManagementService service = ManagementService.getManagementService(cache);
- GatewaySenderMXBean bean = service.getLocalGatewaySenderMXBean("pn");
- assertNotNull(bean);
- bean.stop();
- assertFalse(bean.isRunning());
- }
- };
- vm.invoke(stopGatewaySender);
- }
-
- /**
- * start a gateway sender
- *
- * @param vm reference to VM
- */
- @SuppressWarnings("serial")
- protected void startGatewaySender(final VM vm) {
- SerializableRunnable stopGatewaySender = new SerializableRunnable("Start Gateway Sender") {
- public void run() {
- Cache cache = GemFireCacheImpl.getInstance();
- ManagementService service = ManagementService.getManagementService(cache);
- GatewaySenderMXBean bean = service.getLocalGatewaySenderMXBean("pn");
- assertNotNull(bean);
- bean.start();
- assertTrue(bean.isRunning());
- }
- };
- vm.invoke(stopGatewaySender);
- }
-
- /**
- * Checks whether a GatewayReceiverMBean is created or not
- *
- * @param vm reference to VM
- */
- @SuppressWarnings("serial")
- protected void checkReceiverMBean(final VM vm) {
- SerializableRunnable checkMBean = new SerializableRunnable("Check Receiver MBean") {
- public void run() {
- Cache cache = GemFireCacheImpl.getInstance();
- ManagementService service = ManagementService.getManagementService(cache);
- GatewayReceiverMXBean bean = service.getLocalGatewayReceiverMXBean();
- assertNotNull(bean);
- }
- };
- vm.invoke(checkMBean);
- }
-
- /**
- * Checks whether a GatewayReceiverMBean is created or not
- *
- * @param vm reference to VM
- */
- @SuppressWarnings("serial")
- protected void checkSenderMBean(final VM vm, final String regionPath) {
- SerializableRunnable checkMBean = new SerializableRunnable("Check Sender MBean") {
- public void run() {
- Cache cache = GemFireCacheImpl.getInstance();
- ManagementService service = ManagementService.getManagementService(cache);
-
- GatewaySenderMXBean bean = service.getLocalGatewaySenderMXBean("pn");
- assertNotNull(bean);
- assertTrue(bean.isConnected());
-
- ObjectName regionBeanName = service.getRegionMBeanName(
- cache.getDistributedSystem().getDistributedMember(), "/" + regionPath);
- RegionMXBean rBean = service.getMBeanInstance(regionBeanName, RegionMXBean.class);
- assertTrue(rBean.isGatewayEnabled());
-
- }
- };
- vm.invoke(checkMBean);
- }
-}
--
To stop receiving notification emails like this one, please contact
['"commits@geode.apache.org" <co...@geode.apache.org>'].