geode-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From kl...@apache.org
Subject [08/10] geode git commit: Convert from ManagementTestCase to ManagementTestRule
Date Thu, 26 Jan 2017 00:28:17 GMT
http://git-wip-us.apache.org/repos/asf/geode/blob/2f072c60/geode-core/src/test/java/org/apache/geode/management/DiskManagementDUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/management/DiskManagementDUnitTest.java b/geode-core/src/test/java/org/apache/geode/management/DiskManagementDUnitTest.java
index bcfadde..50533d3 100644
--- a/geode-core/src/test/java/org/apache/geode/management/DiskManagementDUnitTest.java
+++ b/geode-core/src/test/java/org/apache/geode/management/DiskManagementDUnitTest.java
@@ -14,24 +14,25 @@
  */
 package org.apache.geode.management;
 
-import org.junit.experimental.categories.Category;
-import org.junit.Test;
-
-import static org.junit.Assert.*;
-
-import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;
-import org.apache.geode.test.dunit.internal.JUnit4DistributedTestCase;
-import org.apache.geode.test.junit.categories.DistributedTest;
+import static java.util.concurrent.TimeUnit.*;
+import static org.assertj.core.api.Assertions.*;
 
 import java.io.File;
-import java.util.Arrays;
-import java.util.List;
+import java.io.Serializable;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeoutException;
 
 import javax.management.ObjectName;
 
-import org.apache.geode.LogWriter;
+import com.jayway.awaitility.Awaitility;
+import com.jayway.awaitility.core.ConditionFactory;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
 import org.apache.geode.cache.Cache;
 import org.apache.geode.cache.DataPolicy;
 import org.apache.geode.cache.DiskStore;
@@ -40,674 +41,366 @@ import org.apache.geode.cache.Region;
 import org.apache.geode.cache.RegionFactory;
 import org.apache.geode.cache.Scope;
 import org.apache.geode.distributed.DistributedMember;
-import org.apache.geode.internal.cache.DiskRegion;
 import org.apache.geode.internal.cache.DiskRegionStats;
 import org.apache.geode.internal.cache.DistributedRegion;
 import org.apache.geode.internal.cache.GemFireCacheImpl;
-import org.apache.geode.internal.cache.LocalRegion;
 import org.apache.geode.internal.cache.persistence.PersistentMemberID;
 import org.apache.geode.internal.cache.persistence.PersistentMemberManager;
-import org.apache.geode.management.internal.MBeanJMXAdapter;
+import org.apache.geode.internal.process.ProcessUtils;
+import org.apache.geode.management.internal.SystemManagementService;
 import org.apache.geode.test.dunit.AsyncInvocation;
-import org.apache.geode.test.dunit.SerializableCallable;
-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.rules.serializable.SerializableTemporaryFolder;
 
 /**
  * Test cases to cover all test cases which pertains to disk from Management layer
- * 
- * 
  */
 @Category(DistributedTest.class)
-public class DiskManagementDUnitTest extends ManagementTestBase {
-
-  /**
-   * 
-   */
-  private static final long serialVersionUID = 1L;
-
-  // This must be bigger than the dunit ack-wait-threshold for the revoke
-  // tests. The command line is setting the ack-wait-threshold to be
-  // 60 seconds.
-  private static final int MAX_WAIT = 70 * 1000;
-
-  boolean testFailed = false;
+@SuppressWarnings({"serial", "unused"})
+public class DiskManagementDUnitTest implements Serializable {
 
-  String failureCause = "";
-  static final String REGION_NAME = "region";
+  private static final String REGION_NAME =
+      DiskManagementDUnitTest.class.getSimpleName() + "_region";
 
   private File diskDir;
 
-  protected static LogWriter logWriter;
+  @Manager
+  private VM managerVM;
 
-  public DiskManagementDUnitTest() throws Exception {
-    super();
+  @Member
+  private VM[] memberVMs;
 
-    diskDir = new File("diskDir-" + getName()).getAbsoluteFile();
-    org.apache.geode.internal.FileUtil.delete(diskDir);
-    diskDir.mkdir();
-    diskDir.deleteOnExit();
-  }
+  @Rule
+  public ManagementTestRule managementTestRule = ManagementTestRule.builder().start(true).build();
 
-  @Override
-  protected final void postSetUpManagementTestBase() throws Exception {
-    failureCause = "";
-    testFailed = false;
-  }
+  @Rule
+  public SerializableTemporaryFolder temporaryFolder = new SerializableTemporaryFolder();
 
-  @Override
-  protected final void postTearDownManagementTestBase() throws Exception {
-    org.apache.geode.internal.FileUtil.delete(diskDir);
+  @Before
+  public void before() throws Exception {
+    this.diskDir = this.temporaryFolder.newFolder("diskDir");
   }
 
   /**
-   * Tests Disk Compaction from a MemberMbean which is at cache level. All the disks which belong to
-   * the cache should be compacted.
-   * 
-   * @throws Exception
+   * Tests Disk Compaction from a MemberMXBean which is at cache level. All the disks which belong
+   * to the cache should be compacted.
    */
-
   @Test
-  public void testDiskCompact() throws Throwable {
-    initManagement(false);
-    for (VM vm : getManagedNodeList()) {
-      createPersistentRegion(vm);
-      makeDiskCompactable(vm);
+  public void testDiskCompact() throws Exception {
+    for (VM memberVM : this.memberVMs) {
+      createPersistentRegion(memberVM);
+      makeDiskCompactable(memberVM);
     }
 
-    for (VM vm : getManagedNodeList()) {
-      compactAllDiskStores(vm);
+    for (VM memberVM : this.memberVMs) {
+      compactAllDiskStores(memberVM);
     }
-
   }
 
   /**
-   * Tests Disk Compaction from a MemberMbean which is at cache level. All the disks which belong to
-   * the cache should be compacted.
-   * 
-   * @throws Exception
+   * Tests Disk Compaction from a MemberMXBean which is at cache level. All the disks which belong
+   * to the cache should be compacted.
    */
-
   @Test
-  public void testDiskCompactRemote() throws Throwable {
-
-    initManagement(false);
-    for (VM vm : getManagedNodeList()) {
-      createPersistentRegion(vm);
-      makeDiskCompactable(vm);
+  public void testDiskCompactRemote() throws Exception {
+    for (VM memberVM : this.memberVMs) {
+      createPersistentRegion(memberVM);
+      makeDiskCompactable(memberVM);
     }
-    compactDiskStoresRemote(managingNode);
 
+    compactDiskStoresRemote(this.managerVM, this.memberVMs.length);
   }
 
   /**
    * Tests various operations defined on DiskStore Mbean
-   * 
-   * @throws Exception
    */
-
   @Test
-  public void testDiskOps() throws Throwable {
-
-    initManagement(false);
-    for (VM vm : getManagedNodeList()) {
-      createPersistentRegion(vm);
-      makeDiskCompactable(vm);
-      invokeFlush(vm);
-      invokeForceRoll(vm);
-      invokeForceCompaction(vm);
+  public void testDiskOps() throws Exception {
+    for (VM memberVM : this.memberVMs) {
+      createPersistentRegion(memberVM);
+      makeDiskCompactable(memberVM);
+      invokeFlush(memberVM);
+      invokeForceRoll(memberVM);
+      invokeForceCompaction(memberVM);
     }
-
   }
 
   @Test
-  public void testDiskBackupAllMembers() throws Throwable {
-    initManagement(false);
-    for (VM vm : getManagedNodeList()) {
-      createPersistentRegion(vm);
-      makeDiskCompactable(vm);
-
+  public void testDiskBackupAllMembers() throws Exception {
+    for (VM memberVM : this.memberVMs) {
+      createPersistentRegion(memberVM);
+      makeDiskCompactable(memberVM);
     }
-    backupAllMembers(managingNode);
+
+    backupAllMembers(this.managerVM, this.memberVMs.length);
   }
 
   /**
-   * Checks the test case of missing disks and revoking them through MemberMbean interfaces
-   * 
-   * @throws Throwable
+   * Checks the test case of missing disks and revoking them through MemberMXBean interfaces
    */
-  @SuppressWarnings("serial")
   @Test
-  public void testMissingMembers() throws Throwable {
+  public void testMissingMembers() throws Exception {
+    VM memberVM1 = this.memberVMs[0];
+    VM memberVM2 = this.memberVMs[1];
 
-    initManagement(false);
-    VM vm0 = getManagedNodeList().get(0);
-    VM vm1 = getManagedNodeList().get(1);
-    VM vm2 = getManagedNodeList().get(2);
+    createPersistentRegion(memberVM1);
+    createPersistentRegion(memberVM2);
 
-    org.apache.geode.test.dunit.LogWriterUtils.getLogWriter().info("Creating region in VM0");
-    createPersistentRegion(vm0);
-    org.apache.geode.test.dunit.LogWriterUtils.getLogWriter().info("Creating region in VM1");
-    createPersistentRegion(vm1);
+    putAnEntry(memberVM1);
 
-    putAnEntry(vm0);
+    this.managerVM.invoke("checkForMissingDiskStores", () -> {
+      ManagementService service = this.managementTestRule.getManagementService();
+      DistributedSystemMXBean distributedSystemMXBean = service.getDistributedSystemMXBean();
+      PersistentMemberDetails[] missingDiskStores = distributedSystemMXBean.listMissingDiskStores();
 
+      assertThat(missingDiskStores).isNull();
+    });
 
-    managingNode.invoke(new SerializableRunnable("Check for waiting regions") {
+    closeRegion(memberVM1);
 
-      public void run() {
-        Cache cache = getCache();
-        ManagementService service = getManagementService();
-        DistributedSystemMXBean bean = service.getDistributedSystemMXBean();
-        PersistentMemberDetails[] missingDiskStores = bean.listMissingDiskStores();
+    updateTheEntry(memberVM2, "C");
 
-        assertNull(missingDiskStores);
-      }
-    });
+    closeRegion(memberVM2);
 
-    org.apache.geode.test.dunit.LogWriterUtils.getLogWriter().info("closing region in vm0");
-    closeRegion(vm0);
-
-    updateTheEntry(vm1);
-
-    org.apache.geode.test.dunit.LogWriterUtils.getLogWriter().info("closing region in vm1");
-    closeRegion(vm1);
-    AsyncInvocation future = createPersistentRegionAsync(vm0);
-    waitForBlockedInitialization(vm0);
-    assertTrue(future.isAlive());
-
-    managingNode.invoke(new SerializableRunnable("Revoke the member") {
-
-      public void run() {
-        Cache cache = getCache();
-        GemFireCacheImpl cacheImpl = (GemFireCacheImpl) cache;
-        ManagementService service = getManagementService();
-        DistributedSystemMXBean bean = service.getDistributedSystemMXBean();
-        PersistentMemberDetails[] missingDiskStores = bean.listMissingDiskStores();
-        org.apache.geode.test.dunit.LogWriterUtils.getLogWriter()
-            .info("waiting members=" + missingDiskStores);
-        assertNotNull(missingDiskStores);
-        assertEquals(1, missingDiskStores.length);
-
-        for (PersistentMemberDetails id : missingDiskStores) {
-          org.apache.geode.test.dunit.LogWriterUtils.getLogWriter()
-              .info("Missing DiskStoreID is =" + id.getDiskStoreId());
-          org.apache.geode.test.dunit.LogWriterUtils.getLogWriter()
-              .info("Missing Host is =" + id.getHost());
-          org.apache.geode.test.dunit.LogWriterUtils.getLogWriter()
-              .info("Missing Directory is =" + id.getDirectory());
-
-          try {
-            bean.revokeMissingDiskStores(id.getDiskStoreId());
-          } catch (Exception e) {
-            fail("revokeMissingDiskStores failed with exception " + e);
-          }
-        }
-      }
-    });
+    AsyncInvocation creatingPersistentRegionAsync = createPersistentRegionAsync(memberVM1);
 
-    future.join(MAX_WAIT);
-    if (future.isAlive()) {
-      fail("Region not created within" + MAX_WAIT);
-    }
-    if (future.exceptionOccurred()) {
-      throw new Exception(future.getException());
-    }
-    checkForRecoveryStat(vm0, true);
-    // Check to make sure we recovered the old
-    // value of the entry.
-    SerializableRunnable checkForEntry = new SerializableRunnable("check for the entry") {
-
-      public void run() {
-        Cache cache = getCache();
-        Region region = cache.getRegion(REGION_NAME);
-        assertEquals("B", region.get("A"));
-      }
-    };
-    vm0.invoke(checkForEntry);
+    memberVM1.invoke(() -> await().until(() -> {
+      GemFireCacheImpl cache = (GemFireCacheImpl) this.managementTestRule.getCache();
+      PersistentMemberManager persistentMemberManager = cache.getPersistentMemberManager();
+      Map<String, Set<PersistentMemberID>> regions = persistentMemberManager.getWaitingRegions();
+      return !regions.isEmpty();
+    }));
 
-  }
+    assertThat(creatingPersistentRegionAsync.isAlive()).isTrue();
 
-  protected void checkNavigation(final VM vm, final DistributedMember diskMember,
-      final String diskStoreName) {
-    SerializableRunnable checkNavigation = new SerializableRunnable("Check Navigation") {
-      public void run() {
+    this.managerVM.invoke("revokeMissingDiskStore", () -> {
+      ManagementService service = this.managementTestRule.getManagementService();
+      DistributedSystemMXBean bean = service.getDistributedSystemMXBean();
+      PersistentMemberDetails[] missingDiskStores = bean.listMissingDiskStores();
 
-        final ManagementService service = getManagementService();
+      assertThat(missingDiskStores).isNotNull().hasSize(1);
 
-        DistributedSystemMXBean disMBean = service.getDistributedSystemMXBean();
-        try {
-          ObjectName expected =
-              MBeanJMXAdapter.getDiskStoreMBeanName(diskMember.getId(), diskStoreName);
-          ObjectName actual = disMBean.fetchDiskStoreObjectName(diskMember.getId(), diskStoreName);
-          assertEquals(expected, actual);
-        } catch (Exception e) {
-          fail("Disk Store Navigation Failed " + e);
-        }
+      assertThat(bean.revokeMissingDiskStores(missingDiskStores[0].getDiskStoreId())).isTrue();
+    });
 
+    await(creatingPersistentRegionAsync);
 
-      }
-    };
-    vm.invoke(checkNavigation);
-  }
+    verifyRecoveryStats(memberVM1, true);
 
-  /**
-   * get Distributed member for a given vm
-   */
-  @SuppressWarnings("serial")
-  protected static DistributedMember getMember() throws Exception {
-    GemFireCacheImpl cache = GemFireCacheImpl.getInstance();
-    return cache.getDistributedSystem().getDistributedMember();
+    // Check to make sure we recovered the old value of the entry.
+    memberVM1.invoke("check for the entry", () -> {
+      Cache cache = this.managementTestRule.getCache();
+      Region region = cache.getRegion(REGION_NAME);
+      assertThat(region.get("A")).isEqualTo("B");
+    });
   }
 
   /**
    * Invokes flush on the given disk store by MBean interface
-   * 
-   * @param vm reference to VM
    */
-  @SuppressWarnings("serial")
-  public void invokeFlush(final VM vm) {
-    SerializableRunnable invokeFlush = new SerializableRunnable("Invoke Flush On Disk") {
-      public void run() {
-        Cache cache = getCache();
-        DiskStoreFactory dsf = cache.createDiskStoreFactory();
-        String name = "testFlush_" + vm.getPid();
-        DiskStore ds = dsf.create(name);
-
-        ManagementService service = getManagementService();
-        DiskStoreMXBean bean = service.getLocalDiskStoreMBean(name);
-        assertNotNull(bean);
-        bean.flush();
-      }
-    };
-    vm.invoke(invokeFlush);
+  private void invokeFlush(final VM memberVM) {
+    memberVM.invoke("invokeFlush", () -> {
+      Cache cache = this.managementTestRule.getCache();
+      DiskStoreFactory diskStoreFactory = cache.createDiskStoreFactory();
+      String name = "testFlush_" + ProcessUtils.identifyPid();
+      DiskStore diskStore = diskStoreFactory.create(name);
+
+      ManagementService service = this.managementTestRule.getManagementService();
+      DiskStoreMXBean diskStoreMXBean = service.getLocalDiskStoreMBean(name);
+      assertThat(diskStoreMXBean).isNotNull();
+      assertThat(diskStoreMXBean.getName()).isEqualTo(diskStore.getName());
+
+      diskStoreMXBean.flush();
+    });
   }
 
   /**
    * Invokes force roll on disk store by MBean interface
-   * 
-   * @param vm reference to VM
    */
-  @SuppressWarnings("serial")
-  public void invokeForceRoll(final VM vm) {
-    SerializableRunnable invokeForceRoll = new SerializableRunnable("Invoke Force Roll") {
-      public void run() {
-        Cache cache = getCache();
-        DiskStoreFactory dsf = cache.createDiskStoreFactory();
-        String name = "testForceRoll_" + vm.getPid();
-        DiskStore ds = dsf.create(name);
-        ManagementService service = getManagementService();
-        DiskStoreMXBean bean = service.getLocalDiskStoreMBean(name);
-        assertNotNull(bean);
-        bean.forceRoll();
-      }
-    };
-    vm.invoke(invokeForceRoll);
+  private void invokeForceRoll(final VM memberVM) {
+    memberVM.invoke("invokeForceRoll", () -> {
+      Cache cache = this.managementTestRule.getCache();
+      DiskStoreFactory diskStoreFactory = cache.createDiskStoreFactory();
+      String name = "testForceRoll_" + ProcessUtils.identifyPid();
+      DiskStore diskStore = diskStoreFactory.create(name);
+
+      ManagementService service = this.managementTestRule.getManagementService();
+      DiskStoreMXBean diskStoreMXBean = service.getLocalDiskStoreMBean(name);
+      assertThat(diskStoreMXBean).isNotNull();
+      assertThat(diskStoreMXBean.getName()).isEqualTo(diskStore.getName());
+
+      diskStoreMXBean.forceRoll();
+    });
   }
 
   /**
    * Invokes force compaction on disk store by MBean interface
-   * 
-   * @param vm reference to VM
    */
-  @SuppressWarnings("serial")
-  public void invokeForceCompaction(final VM vm) {
-    SerializableRunnable invokeForceCompaction =
-        new SerializableRunnable("Invoke Force Compaction") {
-          public void run() {
-            Cache cache = getCache();
-            DiskStoreFactory dsf = cache.createDiskStoreFactory();
-            dsf.setAllowForceCompaction(true);
-            String name = "testForceCompaction_" + vm.getPid();
-            DiskStore ds = dsf.create(name);
-            ManagementService service = getManagementService();
-            DiskStoreMXBean bean = service.getLocalDiskStoreMBean(name);
-            assertNotNull(bean);
-            assertEquals(false, bean.forceCompaction());
-          }
-        };
-    vm.invoke(invokeForceCompaction);
+  private void invokeForceCompaction(final VM memberVM) {
+    memberVM.invoke("invokeForceCompaction", () -> {
+      Cache cache = this.managementTestRule.getCache();
+      DiskStoreFactory dsf = cache.createDiskStoreFactory();
+      dsf.setAllowForceCompaction(true);
+      String name = "testForceCompaction_" + ProcessUtils.identifyPid();
+      DiskStore diskStore = dsf.create(name);
+
+      ManagementService service = this.managementTestRule.getManagementService();
+      DiskStoreMXBean diskStoreMXBean = service.getLocalDiskStoreMBean(name);
+      assertThat(diskStoreMXBean).isNotNull();
+      assertThat(diskStoreMXBean.getName()).isEqualTo(diskStore.getName());
+
+      assertThat(diskStoreMXBean.forceCompaction()).isFalse();
+    });
   }
 
   /**
    * Makes the disk compactable by adding and deleting some entries
-   * 
-   * @throws Exception
    */
-  @SuppressWarnings("serial")
-  public void makeDiskCompactable(VM vm1) throws Exception {
-    vm1.invoke(new SerializableRunnable("Make The Disk Compactable") {
-
-      public void run() {
-        Cache cache = getCache();
-        Region region = cache.getRegion(REGION_NAME);
-        DiskRegion dr = ((LocalRegion) region).getDiskRegion();
-        org.apache.geode.test.dunit.LogWriterUtils.getLogWriter().info("putting key1");
-        region.put("key1", "value1");
-        org.apache.geode.test.dunit.LogWriterUtils.getLogWriter().info("putting key2");
-        region.put("key2", "value2");
-        org.apache.geode.test.dunit.LogWriterUtils.getLogWriter().info("removing key2");
-        region.remove("key2");
-        // now that it is compactable the following forceCompaction should
-        // go ahead and do a roll and compact it.
-      }
+  private void makeDiskCompactable(final VM memberVM) throws Exception {
+    memberVM.invoke("makeDiskCompactable", () -> {
+      Cache cache = this.managementTestRule.getCache();
+      Region region = cache.getRegion(REGION_NAME);
+      region.put("key1", "value1");
+      region.put("key2", "value2");
+      region.remove("key2");
+      // now that it is compactable the following forceCompaction should
+      // go ahead and do a roll and compact it.
     });
-
   }
 
-
-
   /**
    * Compacts all DiskStores belonging to a member
-   * 
-   * @param vm1 reference to VM
-   * @throws Exception
    */
-  @SuppressWarnings("serial")
-  public void compactAllDiskStores(VM vm1) throws Exception {
-
-    vm1.invoke(new SerializableCallable("Compact All Disk Stores") {
-
-      public Object call() throws Exception {
-        ManagementService service = getManagementService();
-        MemberMXBean memberBean = service.getMemberMXBean();
-        String[] compactedDiskStores = memberBean.compactAllDiskStores();
-
-        assertTrue(compactedDiskStores.length > 0);
-        for (int i = 0; i < compactedDiskStores.length; i++) {
-          org.apache.geode.test.dunit.LogWriterUtils.getLogWriter()
-              .info("<ExpectedString> Compacted Store " + i + " " + compactedDiskStores[i]
-                  + "</ExpectedString> ");
-        }
-
-        return null;
-      }
+  private void compactAllDiskStores(final VM memberVM) throws Exception {
+    memberVM.invoke("compactAllDiskStores", () -> {
+      ManagementService service = this.managementTestRule.getManagementService();
+      MemberMXBean memberMXBean = service.getMemberMXBean();
+      String[] compactedDiskStores = memberMXBean.compactAllDiskStores();
+      assertThat(compactedDiskStores).hasSize(1);
     });
-
   }
 
   /**
    * Takes a back up of all the disk store in a given directory
    */
-  @SuppressWarnings("serial")
-  public void backupAllMembers(final VM managingVM) throws Exception {
+  private void backupAllMembers(final VM managerVM, final int memberCount) {
+    managerVM.invoke("backupAllMembers", () -> {
+      ManagementService service = this.managementTestRule.getManagementService();
+      DistributedSystemMXBean bean = service.getDistributedSystemMXBean();
+      File backupDir = this.temporaryFolder.newFolder("backupDir");
 
-    managingVM.invoke(new SerializableCallable("Backup All Disk Stores") {
+      DiskBackupStatus status = bean.backupAllMembers(backupDir.getAbsolutePath(), null);
 
-      public Object call() throws Exception {
-        ManagementService service = getManagementService();
-        DistributedSystemMXBean bean = service.getDistributedSystemMXBean();
-        DiskBackupStatus status =
-            bean.backupAllMembers(getBackupDir("test_backupAllMembers").getAbsolutePath(), null);
-
-        return null;
-      }
+      assertThat(status.getBackedUpDiskStores().keySet().size()).isEqualTo(memberCount);
+      assertThat(status.getOfflineDiskStores()).isEqualTo(null); // TODO: fix GEODE-1946
     });
-
   }
 
   /**
-   * Compact a disk store from Managing node
+   * Compact a disk store from managerVM VM
    */
-  @SuppressWarnings("serial")
-  public void compactDiskStoresRemote(VM managingVM) throws Exception {
-    {
-
-      managingVM.invoke(new SerializableCallable("Compact All Disk Stores Remote") {
-
-        public Object call() throws Exception {
-          GemFireCacheImpl cache = GemFireCacheImpl.getInstance();
-          Set<DistributedMember> otherMemberSet =
-              cache.getDistributionManager().getOtherNormalDistributionManagerIds();
-
-          for (DistributedMember member : otherMemberSet) {
-            MemberMXBean bean = MBeanUtil.getMemberMbeanProxy(member);
-            String[] allDisks = bean.listDiskStores(true);
-            assertNotNull(allDisks);
-            List<String> listString = Arrays.asList(allDisks);
-            org.apache.geode.test.dunit.LogWriterUtils.getLogWriter()
-                .info("<ExpectedString> Remote All Disk Stores Are  " + listString.toString()
-                    + "</ExpectedString> ");
-            String[] compactedDiskStores = bean.compactAllDiskStores();
-            assertTrue(compactedDiskStores.length > 0);
-            for (int i = 0; i < compactedDiskStores.length; i++) {
-              org.apache.geode.test.dunit.LogWriterUtils.getLogWriter()
-                  .info("<ExpectedString> Remote Compacted Store " + i + " "
-                      + compactedDiskStores[i] + "</ExpectedString> ");
-            }
-
-          }
-          return null;
-        }
-      });
+  private void compactDiskStoresRemote(final VM managerVM, final int memberCount) {
+    managerVM.invoke("compactDiskStoresRemote", () -> {
+      Set<DistributedMember> otherMemberSet = this.managementTestRule.getOtherNormalMembers();// ((GemFireCacheImpl)cache).getDistributionManager().getOtherNormalDistributionManagerIds();
+      assertThat(otherMemberSet.size()).isEqualTo(memberCount);
 
-    }
+      SystemManagementService service = this.managementTestRule.getSystemManagementService();
 
-  }
-
-  /**
-   * Checks if a file with the given extension is present
-   * 
-   * @param fileExtension file extension
-   * @throws Exception
-   */
-  protected void checkIfContainsFileWithExt(String fileExtension) throws Exception {
-    File[] files = diskDir.listFiles();
-    for (int j = 0; j < files.length; j++) {
-      if (files[j].getAbsolutePath().endsWith(fileExtension)) {
-        fail("file \"" + files[j].getAbsolutePath() + "\" still exists");
-      }
-    }
+      for (DistributedMember member : otherMemberSet) {
+        MemberMXBean memberMXBean = awaitMemberMXBeanProxy(member);
 
-  }
+        String[] allDisks = memberMXBean.listDiskStores(true);
+        assertThat(allDisks).isNotNull().hasSize(1);
 
-  /**
-   * Update Entry
-   * 
-   * @param vm1 reference to VM
-   */
-  protected void updateTheEntry(VM vm1) {
-    updateTheEntry(vm1, "C");
-  }
-
-  /**
-   * Update an Entry
-   * 
-   * @param vm1 reference to VM
-   * @param value Value which is updated
-   */
-  @SuppressWarnings("serial")
-  protected void updateTheEntry(VM vm1, final String value) {
-    vm1.invoke(new SerializableRunnable("change the entry") {
-
-      public void run() {
-        Cache cache = getCache();
-        Region region = cache.getRegion(REGION_NAME);
-        region.put("A", value);
+        String[] compactedDiskStores = memberMXBean.compactAllDiskStores();
+        assertThat(compactedDiskStores).hasSize(1);
       }
     });
   }
 
-  /**
-   * Put an entry to region
-   * 
-   * @param vm0 reference to VM
-   */
-  @SuppressWarnings("serial")
-  protected void putAnEntry(VM vm0) {
-    vm0.invoke(new SerializableRunnable("Put an entry") {
-
-      public void run() {
-        Cache cache = getCache();
-        Region region = cache.getRegion(REGION_NAME);
-        region.put("A", "B");
-      }
+  private void updateTheEntry(final VM memberVM, final String value) {
+    memberVM.invoke("updateTheEntry", () -> {
+      Cache cache = this.managementTestRule.getCache();
+      Region region = cache.getRegion(REGION_NAME);
+      region.put("A", value);
     });
   }
 
-  /**
-   * Close the given region REGION_NAME
-   * 
-   * @param vm reference to VM
-   */
-  @SuppressWarnings("serial")
-  protected void closeRegion(final VM vm) {
-    SerializableRunnable closeRegion = new SerializableRunnable("Close persistent region") {
-      public void run() {
-        Cache cache = getCache();
-        Region region = cache.getRegion(REGION_NAME);
-        region.close();
-      }
-    };
-    vm.invoke(closeRegion);
+  private void putAnEntry(final VM memberVM) {
+    memberVM.invoke("putAnEntry", () -> {
+      Cache cache = managementTestRule.getCache();
+      Region region = cache.getRegion(REGION_NAME);
+      region.put("A", "B");
+    });
   }
 
-  /**
-   * Waiting to blocked waiting for another persistent member to come online
-   * 
-   * @param vm reference to VM
-   */
-  @SuppressWarnings("serial")
-  private void waitForBlockedInitialization(VM vm) {
-    vm.invoke(new SerializableRunnable() {
-
-      public void run() {
-        Wait.waitForCriterion(new WaitCriterion() {
-
-          public String description() {
-            return "Waiting to blocked waiting for another persistent member to come online";
-          }
-
-          public boolean done() {
-            Cache cache = getCache();
-            GemFireCacheImpl cacheImpl = (GemFireCacheImpl) cache;
-            PersistentMemberManager mm = cacheImpl.getPersistentMemberManager();
-            Map<String, Set<PersistentMemberID>> regions = mm.getWaitingRegions();
-            boolean done = !regions.isEmpty();
-            return done;
-          }
-
-        }, MAX_WAIT, 100, true);
-
-      }
-
+  private void closeRegion(final VM memberVM) {
+    memberVM.invoke("closeRegion", () -> {
+      Cache cache = this.managementTestRule.getCache();
+      Region region = cache.getRegion(REGION_NAME);
+      region.close();
     });
   }
 
-  /**
-   * Creates a persistent region
-   * 
-   * @param vm reference to VM
-   * @throws Throwable
-   */
-  protected void createPersistentRegion(VM vm) throws Throwable {
-    AsyncInvocation future = createPersistentRegionAsync(vm);
-    future.join(MAX_WAIT);
-    if (future.isAlive()) {
-      fail("Region not created within" + MAX_WAIT);
-    }
-    if (future.exceptionOccurred()) {
-      throw new RuntimeException(future.getException());
-    }
+  private void createPersistentRegion(final VM memberVM)
+      throws InterruptedException, ExecutionException, TimeoutException {
+    await(createPersistentRegionAsync(memberVM));
   }
 
-  /**
-   * Creates a persistent region in async manner
-   * 
-   * @param vm reference to VM
-   * @return reference to AsyncInvocation
-   */
-  @SuppressWarnings("serial")
-  protected AsyncInvocation createPersistentRegionAsync(final VM vm) {
-    SerializableRunnable createRegion = new SerializableRunnable("Create persistent region") {
-      public void run() {
-        Cache cache = getCache();
-        DiskStoreFactory dsf = cache.createDiskStoreFactory();
-        File dir = getDiskDirForVM(vm);
-        dir.mkdirs();
-        dsf.setDiskDirs(new File[] {dir});
-        dsf.setMaxOplogSize(1);
-        dsf.setAllowForceCompaction(true);
-        dsf.setAutoCompact(false);
-        DiskStore ds = dsf.create(REGION_NAME);
-        RegionFactory rf = cache.createRegionFactory();
-        rf.setDiskStoreName(ds.getName());
-        rf.setDiskSynchronous(true);
-        rf.setDataPolicy(DataPolicy.PERSISTENT_REPLICATE);
-        rf.setScope(Scope.DISTRIBUTED_ACK);
-        rf.create(REGION_NAME);
-      }
-    };
-    return vm.invokeAsync(createRegion);
+  private AsyncInvocation createPersistentRegionAsync(final VM memberVM) {
+    return memberVM.invokeAsync("createPersistentRegionAsync", () -> {
+      File dir = new File(diskDir, String.valueOf(ProcessUtils.identifyPid()));
+
+      Cache cache = this.managementTestRule.getCache();
+
+      DiskStoreFactory diskStoreFactory = cache.createDiskStoreFactory();
+      diskStoreFactory.setDiskDirs(new File[] {dir});
+      diskStoreFactory.setMaxOplogSize(1);
+      diskStoreFactory.setAllowForceCompaction(true);
+      diskStoreFactory.setAutoCompact(false);
+      DiskStore diskStore = diskStoreFactory.create(REGION_NAME);
+
+      RegionFactory regionFactory = cache.createRegionFactory();
+      regionFactory.setDiskStoreName(diskStore.getName());
+      regionFactory.setDiskSynchronous(true);
+      regionFactory.setDataPolicy(DataPolicy.PERSISTENT_REPLICATE);
+      regionFactory.setScope(Scope.DISTRIBUTED_ACK);
+      regionFactory.create(REGION_NAME);
+    });
   }
 
-  /**
-   * Validates a persistent region
-   * 
-   * @param vm reference to VM
-   */
-  @SuppressWarnings("serial")
-  protected void validatePersistentRegion(final VM vm) {
-    SerializableRunnable validateDisk = new SerializableRunnable("Validate persistent region") {
-      public void run() {
-        Cache cache = getCache();
-        ManagementService service = getManagementService();
-        DiskStoreMXBean bean = service.getLocalDiskStoreMBean(REGION_NAME);
-        assertNotNull(bean);
+  private void verifyRecoveryStats(final VM memberVM, final boolean localRecovery) {
+    memberVM.invoke("verifyRecoveryStats", () -> {
+      Cache cache = this.managementTestRule.getCache();
+      Region region = cache.getRegion(REGION_NAME);
+      DistributedRegion distributedRegion = (DistributedRegion) region;
+      DiskRegionStats stats = distributedRegion.getDiskRegion().getStats();
+
+      if (localRecovery) {
+        assertThat(stats.getLocalInitializations()).isEqualTo(1);
+        assertThat(stats.getRemoteInitializations()).isEqualTo(0);
+      } else {
+        assertThat(stats.getLocalInitializations()).isEqualTo(0);
+        assertThat(stats.getRemoteInitializations()).isEqualTo(1);
       }
-    };
-    vm.invoke(validateDisk);
+    });
   }
 
-  /**
-   * Appends vm id to disk dir
-   * 
-   * @param vm reference to VM
-   * @return
-   */
-  protected File getDiskDirForVM(final VM vm) {
-    File dir = new File(diskDir, String.valueOf(vm.getPid()));
-    return dir;
+  private MemberMXBean awaitMemberMXBeanProxy(final DistributedMember member) {
+    SystemManagementService service = this.managementTestRule.getSystemManagementService();
+    ObjectName objectName = service.getMemberMBeanName(member);
+    await()
+        .until(() -> assertThat(service.getMBeanProxy(objectName, MemberMXBean.class)).isNotNull());
+    return service.getMBeanProxy(objectName, MemberMXBean.class);
   }
 
-  /**
-   * Checks recovery status
-   * 
-   * @param vm reference to VM
-   * @param localRecovery local recovery on or not
-   */
-  @SuppressWarnings("serial")
-  private void checkForRecoveryStat(VM vm, final boolean localRecovery) {
-    vm.invoke(new SerializableRunnable("check disk region stat") {
-
-      public void run() {
-        Cache cache = getCache();
-        Region region = cache.getRegion(REGION_NAME);
-        DistributedRegion distributedRegion = (DistributedRegion) region;
-        DiskRegionStats stats = distributedRegion.getDiskRegion().getStats();
-        if (localRecovery) {
-          assertEquals(1, stats.getLocalInitializations());
-          assertEquals(0, stats.getRemoteInitializations());
-        } else {
-          assertEquals(0, stats.getLocalInitializations());
-          assertEquals(1, stats.getRemoteInitializations());
-        }
-
-      }
-    });
+  private void await(final AsyncInvocation createPersistentRegionAsync)
+      throws InterruptedException, ExecutionException, TimeoutException {
+    createPersistentRegionAsync.await(2, MINUTES);
   }
 
-  /**
-   * 
-   * @return back up directory
-   */
-  protected static File getBackupDir(String name) throws Exception {
-    File backUpDir = new File("BackupDir-" + name).getAbsoluteFile();
-    org.apache.geode.internal.FileUtil.delete(backUpDir);
-    backUpDir.mkdir();
-    backUpDir.deleteOnExit();
-    return backUpDir;
+  private ConditionFactory await() {
+    return Awaitility.await().atMost(2, MINUTES);
   }
 }

http://git-wip-us.apache.org/repos/asf/geode/blob/2f072c60/geode-core/src/test/java/org/apache/geode/management/DistributedSystemDUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/management/DistributedSystemDUnitTest.java b/geode-core/src/test/java/org/apache/geode/management/DistributedSystemDUnitTest.java
index 4eaba67..79a4ce4 100644
--- a/geode-core/src/test/java/org/apache/geode/management/DistributedSystemDUnitTest.java
+++ b/geode-core/src/test/java/org/apache/geode/management/DistributedSystemDUnitTest.java
@@ -14,23 +14,19 @@
  */
 package org.apache.geode.management;
 
-import org.junit.experimental.categories.Category;
-import org.junit.Test;
-
-import static org.junit.Assert.*;
-
-import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;
-import org.apache.geode.test.dunit.internal.JUnit4DistributedTestCase;
-import org.apache.geode.test.junit.categories.DistributedTest;
+import static java.util.concurrent.TimeUnit.*;
+import static org.apache.geode.test.dunit.Host.*;
+import static org.apache.geode.test.dunit.Invoke.*;
+import static org.assertj.core.api.Assertions.*;
 
+import java.io.Serializable;
+import java.lang.management.ManagementFactory;
 import java.util.ArrayList;
 import java.util.HashMap;
-import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 
-import javax.management.InstanceNotFoundException;
 import javax.management.ListenerNotFoundException;
 import javax.management.MBeanServer;
 import javax.management.Notification;
@@ -39,11 +35,16 @@ import javax.management.NotificationFilter;
 import javax.management.NotificationListener;
 import javax.management.ObjectName;
 
+import com.jayway.awaitility.Awaitility;
+import com.jayway.awaitility.core.ConditionFactory;
 import org.apache.logging.log4j.Logger;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+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.internal.InternalDistributedSystem;
 import org.apache.geode.internal.admin.Alert;
 import org.apache.geode.internal.cache.GemFireCacheImpl;
 import org.apache.geode.internal.logging.LogService;
@@ -57,428 +58,277 @@ import org.apache.geode.management.internal.SystemManagementService;
 import org.apache.geode.management.internal.beans.MemberMBean;
 import org.apache.geode.management.internal.beans.SequenceNumber;
 import org.apache.geode.test.dunit.IgnoredException;
-import org.apache.geode.test.dunit.LogWriterUtils;
-import org.apache.geode.test.dunit.Host;
-import org.apache.geode.test.dunit.SerializableCallable;
-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;
 
 /**
- * Distributed System tests
- * 
- * a) For all the notifications
- * 
- * i) gemfire.distributedsystem.member.joined
- * 
- * ii) gemfire.distributedsystem.member.left
- * 
- * iii) gemfire.distributedsystem.member.suspect
- * 
- * iv ) All notifications emitted by member mbeans
- * 
- * vi) Alerts
- * 
+ * Distributed System management tests
+ * <p>
+ * a) For all the notifications i) gemfire.distributedsystem.member.joined ii)
+ * gemfire.distributedsystem.member.left iii) gemfire.distributedsystem.member.suspect iv ) All
+ * notifications emitted by member mbeans vi) Alerts
+ * <p>
  * b) Concurrently modify proxy list by removing member and accessing the distributed system MBean
- * 
+ * <p>
  * c) Aggregate Operations like shutDownAll
- * 
+ * <p>
  * d) Member level operations like fetchJVMMetrics()
- * 
+ * <p>
  * e ) Statistics
- * 
- * 
- * 
  */
 @Category(DistributedTest.class)
-public class DistributedSystemDUnitTest extends ManagementTestBase {
+@SuppressWarnings({"serial", "unused"})
+public class DistributedSystemDUnitTest implements Serializable {
 
   private static final Logger logger = LogService.getLogger();
 
-  private static final long serialVersionUID = 1L;
-
+  private static final String WARNING_LEVEL_MESSAGE = "Warning Level Alert Message";
+  private static final String SEVERE_LEVEL_MESSAGE = "Severe Level Alert Message";
 
-  private static final int MAX_WAIT = 10 * 1000;
+  private static List<Notification> notifications;
+  private static Map<ObjectName, NotificationListener> notificationListenerMap;
 
-  private static MBeanServer mbeanServer = MBeanJMXAdapter.mbeanServer;
+  @Manager
+  private VM managerVM;
 
-  static List<Notification> notifList = new ArrayList<>();
+  @Member
+  private VM[] memberVMs;
 
-  static Map<ObjectName, NotificationListener> notificationListenerMap =
-      new HashMap<ObjectName, NotificationListener>();
+  @Rule
+  public ManagementTestRule managementTestRule = ManagementTestRule.builder().build();
 
-  static final String WARNING_LEVEL_MESSAGE = "Warninglevel Alert Message";
-
-  static final String SEVERE_LEVEL_MESSAGE = "Severelevel Alert Message";
+  @Before
+  public void before() throws Exception {
+    notifications = new ArrayList<>();
+    notificationListenerMap = new HashMap<>();
 
+    invokeInEveryVM(() -> notifications = new ArrayList<>());
+    invokeInEveryVM(() -> notificationListenerMap = new HashMap<>());
+  }
 
-  public DistributedSystemDUnitTest() {
-    super();
+  @After
+  public void after() throws Exception {
+    resetAlertCounts(this.managerVM);
   }
 
   /**
    * Tests each and every operations that is defined on the MemberMXBean
-   * 
-   * @throws Exception
    */
   @Test
   public void testDistributedSystemAggregate() throws Exception {
-    VM managingNode = getManagingNode();
-    createManagementCache(managingNode);
-    startManagingNode(managingNode);
-    addNotificationListener(managingNode);
+    this.managementTestRule.createManager(this.managerVM);
+    addNotificationListener(this.managerVM);
 
-    for (VM vm : getManagedNodeList()) {
-      createCache(vm);
+    for (VM memberVM : this.memberVMs) {
+      this.managementTestRule.createMember(memberVM);
     }
 
-    checkAggregate(managingNode);
-    for (VM vm : getManagedNodeList()) {
-      closeCache(vm);
-    }
-
-    closeCache(managingNode);
-
+    verifyDistributedSystemMXBean(this.managerVM);
   }
 
   /**
    * Tests each and every operations that is defined on the MemberMXBean
-   * 
-   * @throws Exception
    */
   @Test
   public void testAlertManagedNodeFirst() throws Exception {
-
-    for (VM vm : getManagedNodeList()) {
-      createCache(vm);
-      warnLevelAlert(vm);
-      severeLevelAlert(vm);
+    for (VM memberVM : this.memberVMs) {
+      this.managementTestRule.createMember(memberVM);
+      generateWarningAlert(memberVM);
+      generateSevereAlert(memberVM);
     }
 
-    VM managingNode = getManagingNode();
-
-    createManagementCache(managingNode);
-    startManagingNode(managingNode);
-    addAlertListener(managingNode);
-    checkAlertCount(managingNode, 0, 0);
+    this.managementTestRule.createManager(this.managerVM);
+    addAlertListener(this.managerVM);
+    verifyAlertCount(this.managerVM, 0, 0);
 
-    final DistributedMember managingMember = getMember(managingNode);
+    DistributedMember managerDistributedMember =
+        this.managementTestRule.getDistributedMember(this.managerVM);
 
     // Before we start we need to ensure that the initial (implicit) SEVERE alert has propagated
     // everywhere.
-    for (VM vm : getManagedNodeList()) {
-      ensureLoggerState(vm, managingMember, Alert.SEVERE);
+    for (VM memberVM : this.memberVMs) {
+      verifyAlertAppender(memberVM, managerDistributedMember, Alert.SEVERE);
     }
 
-    setAlertLevel(managingNode, AlertDetails.getAlertLevelAsString(Alert.WARNING));
+    setAlertLevel(this.managerVM, AlertDetails.getAlertLevelAsString(Alert.WARNING));
 
-    for (VM vm : getManagedNodeList()) {
-      ensureLoggerState(vm, managingMember, Alert.WARNING);
-      warnLevelAlert(vm);
-      severeLevelAlert(vm);
+    for (VM memberVM : this.memberVMs) {
+      verifyAlertAppender(memberVM, managerDistributedMember, Alert.WARNING);
+      generateWarningAlert(memberVM);
+      generateSevereAlert(memberVM);
     }
 
-    checkAlertCount(managingNode, 3, 3);
-    resetAlertCounts(managingNode);
+    verifyAlertCount(this.managerVM, 3, 3);
+    resetAlertCounts(this.managerVM);
 
-    setAlertLevel(managingNode, AlertDetails.getAlertLevelAsString(Alert.SEVERE));
+    setAlertLevel(this.managerVM, AlertDetails.getAlertLevelAsString(Alert.SEVERE));
 
-    for (VM vm : getManagedNodeList()) {
-      ensureLoggerState(vm, managingMember, Alert.SEVERE);
-      warnLevelAlert(vm);
-      severeLevelAlert(vm);
+    for (VM memberVM : this.memberVMs) {
+      verifyAlertAppender(memberVM, managerDistributedMember, Alert.SEVERE);
+      generateWarningAlert(memberVM);
+      generateSevereAlert(memberVM);
     }
 
-    checkAlertCount(managingNode, 3, 0);
-    resetAlertCounts(managingNode);
-
-    for (VM vm : getManagedNodeList()) {
-      closeCache(vm);
-    }
-
-    closeCache(managingNode);
-  }
-
-  @SuppressWarnings("serial")
-  public void ensureLoggerState(VM vm1, final DistributedMember member, final int alertLevel)
-      throws Exception {
-    {
-      vm1.invoke(new SerializableCallable("Ensure Logger State") {
-
-        public Object call() throws Exception {
-
-          Wait.waitForCriterion(new WaitCriterion() {
-            public String description() {
-              return "Waiting for all alert Listener to register with managed node";
-            }
-
-            public boolean done() {
-
-              if (AlertAppender.getInstance().hasAlertListener(member, alertLevel)) {
-                return true;
-              }
-              return false;
-            }
-
-          }, MAX_WAIT, 500, true);
-
-          return null;
-        }
-      });
-
-    }
+    verifyAlertCount(this.managerVM, 3, 0);
   }
 
   /**
    * Tests each and every operations that is defined on the MemberMXBean
-   * 
-   * @throws Exception
    */
   @Test
   public void testShutdownAll() throws Exception {
-    final Host host = Host.getHost(0);
-    VM managedNode1 = host.getVM(0);
-    VM managedNode2 = host.getVM(1);
-    VM managedNode3 = host.getVM(2);
-
-    VM managingNode = host.getVM(3);
-
-    // Managing Node is created first
-    createManagementCache(managingNode);
-    startManagingNode(managingNode);
-
-    createCache(managedNode1);
-    createCache(managedNode2);
-    createCache(managedNode3);
-    shutDownAll(managingNode);
-    closeCache(managingNode);
+    VM memberVM1 = getHost(0).getVM(0);
+    VM memberVM2 = getHost(0).getVM(1);
+    VM memberVM3 = getHost(0).getVM(2);
+
+    VM managerVM = getHost(0).getVM(3);
+
+    // managerVM Node is created first
+    this.managementTestRule.createManager(managerVM);
+
+    this.managementTestRule.createMember(memberVM1);
+    this.managementTestRule.createMember(memberVM2);
+    this.managementTestRule.createMember(memberVM3);
+
+    shutDownAll(managerVM);
   }
 
   @Test
   public void testNavigationAPIS() throws Exception {
+    this.managementTestRule.createManager(this.managerVM);
 
-    final Host host = Host.getHost(0);
-
-    createManagementCache(managingNode);
-    startManagingNode(managingNode);
-
-    for (VM vm : managedNodeList) {
-      createCache(vm);
+    for (VM memberVM : this.memberVMs) {
+      this.managementTestRule.createMember(memberVM);
     }
 
-    checkNavigationAPIs(managingNode);
+    verifyFetchMemberObjectName(this.managerVM, this.memberVMs.length + 1);
   }
 
   @Test
   public void testNotificationHub() throws Exception {
-    this.initManagement(false);
+    this.managementTestRule.createMembers();
+    this.managementTestRule.createManagers();
 
     class NotificationHubTestListener implements NotificationListener {
+
       @Override
       public synchronized void handleNotification(Notification notification, Object handback) {
         logger.info("Notification received {}", notification);
-        notifList.add(notification);
+        notifications.add(notification);
       }
     }
 
-    managingNode.invoke(new SerializableRunnable("Add Listener to MemberMXBean") {
-
-      public void run() {
-        Cache cache = getCache();
-        ManagementService service = getManagementService();
-        final DistributedSystemMXBean bean = service.getDistributedSystemMXBean();
-
-        Wait.waitForCriterion(new WaitCriterion() {
-          public String description() {
-            return "Waiting for all members to send their initial Data";
-          }
-
-          public boolean done() {
-            if (bean.listMemberObjectNames().length == 5) {// including locator
-              return true;
-            } else {
-              return false;
-            }
-          }
-        }, MAX_WAIT, 500, true);
-        for (ObjectName objectName : bean.listMemberObjectNames()) {
-          NotificationHubTestListener listener = new NotificationHubTestListener();
-          try {
-            mbeanServer.addNotificationListener(objectName, listener, null, null);
-            notificationListenerMap.put(objectName, listener);
-          } catch (InstanceNotFoundException e) {
-            LogWriterUtils.getLogWriter().error(e);
-          }
-        }
+    this.managerVM.invoke("addListenerToMemberMXBean", () -> {
+      ManagementService service = this.managementTestRule.getManagementService();
+      final DistributedSystemMXBean distributedSystemMXBean = service.getDistributedSystemMXBean();
+
+      await().until(() -> assertThat(distributedSystemMXBean.listMemberObjectNames()).hasSize(5));
+
+      for (ObjectName objectName : distributedSystemMXBean.listMemberObjectNames()) {
+        NotificationHubTestListener listener = new NotificationHubTestListener();
+        ManagementFactory.getPlatformMBeanServer().addNotificationListener(objectName, listener,
+            null, null);
+        notificationListenerMap.put(objectName, listener);
       }
     });
 
     // Check in all VMS
 
-    for (VM vm : managedNodeList) {
-      vm.invoke(new SerializableRunnable("Check Hub Listener num count") {
-
-        public void run() {
-          Cache cache = getCache();
-          SystemManagementService service = (SystemManagementService) getManagementService();
-          NotificationHub hub = service.getNotificationHub();
-          Map<ObjectName, NotificationHubListener> listenerObjectMap = hub.getListenerObjectMap();
-          assertEquals(1, listenerObjectMap.keySet().size());
-          ObjectName memberMBeanName = MBeanJMXAdapter
-              .getMemberMBeanName(cache.getDistributedSystem().getDistributedMember());
-
-          NotificationHubListener listener = listenerObjectMap.get(memberMBeanName);
-
-          /*
-           * Counter of listener should be 2 . One for default Listener which is added for each
-           * member mbean by distributed system mbean One for the added listener in test
-           */
-          assertEquals(2, listener.getNumCounter());
-
-          // Raise some notifications
-
-          NotificationBroadcasterSupport memberLevelNotifEmitter =
-              (MemberMBean) service.getMemberMXBean();
-
-          String memberSource = MBeanJMXAdapter
-              .getMemberNameOrId(cache.getDistributedSystem().getDistributedMember());
-
-          // Only a dummy notification , no actual region is creates
-          Notification notification = new Notification(JMXNotificationType.REGION_CREATED,
-              memberSource, SequenceNumber.next(), System.currentTimeMillis(),
-              ManagementConstants.REGION_CREATED_PREFIX + "/test");
-          memberLevelNotifEmitter.sendNotification(notification);
-
-        }
+    for (VM memberVM : this.memberVMs) {
+      memberVM.invoke("checkNotificationHubListenerCount", () -> {
+        SystemManagementService service = this.managementTestRule.getSystemManagementService();
+        NotificationHub notificationHub = service.getNotificationHub();
+        Map<ObjectName, NotificationHubListener> listenerMap =
+            notificationHub.getListenerObjectMap();
+        assertThat(listenerMap.keySet()).hasSize(1);
+
+        ObjectName memberMBeanName =
+            MBeanJMXAdapter.getMemberMBeanName(this.managementTestRule.getDistributedMember());
+        NotificationHubListener listener = listenerMap.get(memberMBeanName);
+
+        /*
+         * Counter of listener should be 2 . One for default Listener which is added for each member
+         * mbean by distributed system mbean One for the added listener in test
+         */
+        assertThat(listener.getNumCounter()).isEqualTo(2);
+
+        // Raise some notifications
+
+        NotificationBroadcasterSupport notifier = (MemberMBean) service.getMemberMXBean();
+        String memberSource =
+            MBeanJMXAdapter.getMemberNameOrId(this.managementTestRule.getDistributedMember());
+
+        // Only a dummy notification , no actual region is created
+        Notification notification = new Notification(JMXNotificationType.REGION_CREATED,
+            memberSource, SequenceNumber.next(), System.currentTimeMillis(),
+            ManagementConstants.REGION_CREATED_PREFIX + "/test");
+        notifier.sendNotification(notification);
       });
     }
 
-    managingNode.invoke(new SerializableRunnable("Check notifications && Remove Listeners") {
-
-      public void run() {
-
-        Wait.waitForCriterion(new WaitCriterion() {
-          public String description() {
-            return "Waiting for all Notifications to reach the Managing Node";
-          }
-
-          public boolean done() {
-            if (notifList.size() == 3) {
-              return true;
-            } else {
-              return false;
-            }
-          }
-        }, MAX_WAIT, 500, true);
-
-        notifList.clear();
+    this.managerVM.invoke("checkNotificationsAndRemoveListeners", () -> {
+      await().until(() -> assertThat(notifications).hasSize(3));
 
-        Iterator<ObjectName> it = notificationListenerMap.keySet().iterator();
-        while (it.hasNext()) {
-          ObjectName objectName = it.next();
-          NotificationListener listener = notificationListenerMap.get(objectName);
-          try {
-            mbeanServer.removeNotificationListener(objectName, listener);
-          } catch (ListenerNotFoundException e) {
-            LogWriterUtils.getLogWriter().error(e);
-          } catch (InstanceNotFoundException e) {
-            LogWriterUtils.getLogWriter().error(e);
-          }
-        }
+      notifications.clear();
 
+      for (ObjectName objectName : notificationListenerMap.keySet()) {
+        NotificationListener listener = notificationListenerMap.get(objectName);
+        ManagementFactory.getPlatformMBeanServer().removeNotificationListener(objectName, listener);
       }
     });
 
     // Check in all VMS again
 
-    for (VM vm : managedNodeList) {
-      vm.invoke(new SerializableRunnable("Check Hub Listener num count Again") {
-
-        public void run() {
-          Cache cache = getCache();
-          SystemManagementService service = (SystemManagementService) getManagementService();
-          NotificationHub hub = service.getNotificationHub();
-          Map<ObjectName, NotificationHubListener> listenerObjectMap = hub.getListenerObjectMap();
-
-          assertEquals(1, listenerObjectMap.keySet().size());
-
-          ObjectName memberMBeanName = MBeanJMXAdapter
-              .getMemberMBeanName(cache.getDistributedSystem().getDistributedMember());
-
-          NotificationHubListener listener = listenerObjectMap.get(memberMBeanName);
-
-          /*
-           * Counter of listener should be 1 for the default Listener which is added for each member
-           * mbean by distributed system mbean.
-           */
-          assertEquals(1, listener.getNumCounter());
-
-        }
+    for (VM memberVM : this.memberVMs) {
+      memberVM.invoke("checkNotificationHubListenerCountAgain", () -> {
+        SystemManagementService service = this.managementTestRule.getSystemManagementService();
+        NotificationHub hub = service.getNotificationHub();
+        Map<ObjectName, NotificationHubListener> listenerObjectMap = hub.getListenerObjectMap();
+        assertThat(listenerObjectMap.keySet().size()).isEqualTo(1);
+
+        ObjectName memberMBeanName =
+            MBeanJMXAdapter.getMemberMBeanName(this.managementTestRule.getDistributedMember());
+        NotificationHubListener listener = listenerObjectMap.get(memberMBeanName);
+
+        /*
+         * Counter of listener should be 1 for the default Listener which is added for each member
+         * mbean by distributed system mbean.
+         */
+        assertThat(listener.getNumCounter()).isEqualTo(1);
       });
     }
 
-    managingNode.invoke(new SerializableRunnable("Remove Listener from MemberMXBean") {
-
-      public void run() {
-        Cache cache = getCache();
-        ManagementService service = getManagementService();
-        final DistributedSystemMXBean bean = service.getDistributedSystemMXBean();
-
-        Wait.waitForCriterion(new WaitCriterion() {
-          public String description() {
-            return "Waiting for all members to send their initial Data";
-          }
-
-          public boolean done() {
-            if (bean.listMemberObjectNames().length == 5) {// including locator
-              return true;
-            } else {
-              return false;
-            }
-
-          }
-
-        }, MAX_WAIT, 500, true);
-        for (ObjectName objectName : bean.listMemberObjectNames()) {
-          NotificationHubTestListener listener = new NotificationHubTestListener();
-          try {
-            mbeanServer.removeNotificationListener(objectName, listener);
-          } catch (InstanceNotFoundException e) {
-            LogWriterUtils.getLogWriter().error(e);
-          } catch (ListenerNotFoundException e) {
-            // TODO: apparently there is never a notification listener on any these mbeans at this
-            // point
-            // fix this test so it doesn't hit these unexpected exceptions --
-            // getLogWriter().error(e);
-          }
+    this.managerVM.invoke("removeListenerFromMemberMXBean", () -> {
+      ManagementService service = this.managementTestRule.getManagementService();
+      DistributedSystemMXBean distributedSystemMXBean = service.getDistributedSystemMXBean();
+
+      await().until(() -> assertThat(distributedSystemMXBean.listMemberObjectNames()).hasSize(5));
+
+      for (ObjectName objectName : distributedSystemMXBean.listMemberObjectNames()) {
+        NotificationHubTestListener listener = new NotificationHubTestListener();
+        try {
+          ManagementFactory.getPlatformMBeanServer().removeNotificationListener(objectName,
+              listener); // because new instance!!
+        } catch (ListenerNotFoundException e) {
+          // TODO: [old] apparently there is never a notification listener on any these mbeans at
+          // this point [fix this]
+          // fix this test so it doesn't hit these unexpected exceptions -- getLogWriter().error(e);
         }
       }
     });
 
-    for (VM vm : managedNodeList) {
-      vm.invoke(new SerializableRunnable("Check Hub Listeners clean up") {
-
-        public void run() {
-          Cache cache = getCache();
-          SystemManagementService service = (SystemManagementService) getManagementService();
-          NotificationHub hub = service.getNotificationHub();
-          hub.cleanUpListeners();
-          assertEquals(0, hub.getListenerObjectMap().size());
-
-          Iterator<ObjectName> it = notificationListenerMap.keySet().iterator();
-          while (it.hasNext()) {
-            ObjectName objectName = it.next();
-            NotificationListener listener = notificationListenerMap.get(objectName);
-            try {
-              mbeanServer.removeNotificationListener(objectName, listener);
-              fail("Found Listeners inspite of clearing them");
-            } catch (ListenerNotFoundException e) {
-              // Expected Exception Do nothing
-            } catch (InstanceNotFoundException e) {
-              LogWriterUtils.getLogWriter().error(e);
-            }
-          }
+    for (VM memberVM : this.memberVMs) {
+      memberVM.invoke("verifyNotificationHubListenersWereRemoved", () -> {
+        SystemManagementService service = this.managementTestRule.getSystemManagementService();
+        NotificationHub notificationHub = service.getNotificationHub();
+        notificationHub.cleanUpListeners();
+        assertThat(notificationHub.getListenerObjectMap()).isEmpty();
+
+        for (ObjectName objectName : notificationListenerMap.keySet()) {
+          NotificationListener listener = notificationListenerMap.get(objectName);
+          assertThatThrownBy(() -> ManagementFactory.getPlatformMBeanServer()
+              .removeNotificationListener(objectName, listener))
+                  .isExactlyInstanceOf(ListenerNotFoundException.class);
         }
       });
     }
@@ -486,404 +336,229 @@ public class DistributedSystemDUnitTest extends ManagementTestBase {
 
   /**
    * Tests each and every operations that is defined on the MemberMXBean
-   * 
-   * @throws Exception
    */
   @Test
   public void testAlert() throws Exception {
-    VM managingNode = getManagingNode();
-
-    createManagementCache(managingNode);
-    startManagingNode(managingNode);
-    addAlertListener(managingNode);
-    resetAlertCounts(managingNode);
-
-    final DistributedMember managingMember = getMember(managingNode);
-
-
+    this.managementTestRule.createManager(this.managerVM);
+    addAlertListener(this.managerVM);
+    resetAlertCounts(this.managerVM);
 
-    warnLevelAlert(managingNode);
-    severeLevelAlert(managingNode);
-    checkAlertCount(managingNode, 1, 0);
-    resetAlertCounts(managingNode);
+    DistributedMember managerDistributedMember =
+        this.managementTestRule.getDistributedMember(this.managerVM);
 
-    for (VM vm : getManagedNodeList()) {
+    generateWarningAlert(this.managerVM);
+    generateSevereAlert(this.managerVM);
+    verifyAlertCount(this.managerVM, 1, 0);
+    resetAlertCounts(this.managerVM);
 
-      createCache(vm);
-      // Default is severe ,So only Severe level alert is expected
+    for (VM memberVM : this.memberVMs) {
+      this.managementTestRule.createMember(memberVM);
 
-      ensureLoggerState(vm, managingMember, Alert.SEVERE);
-
-      warnLevelAlert(vm);
-      severeLevelAlert(vm);
+      verifyAlertAppender(memberVM, managerDistributedMember, Alert.SEVERE);
 
+      generateWarningAlert(memberVM);
+      generateSevereAlert(memberVM);
     }
-    checkAlertCount(managingNode, 3, 0);
-    resetAlertCounts(managingNode);
-    setAlertLevel(managingNode, AlertDetails.getAlertLevelAsString(Alert.WARNING));
 
+    verifyAlertCount(this.managerVM, 3, 0);
+    resetAlertCounts(this.managerVM);
+    setAlertLevel(this.managerVM, AlertDetails.getAlertLevelAsString(Alert.WARNING));
 
-    for (VM vm : getManagedNodeList()) {
-      // warning and severe alerts both are to be checked
-      ensureLoggerState(vm, managingMember, Alert.WARNING);
-      warnLevelAlert(vm);
-      severeLevelAlert(vm);
+    for (VM memberVM : this.memberVMs) {
+      verifyAlertAppender(memberVM, managerDistributedMember, Alert.WARNING);
+      generateWarningAlert(memberVM);
+      generateSevereAlert(memberVM);
     }
 
-    checkAlertCount(managingNode, 3, 3);
-
-    resetAlertCounts(managingNode);
+    verifyAlertCount(this.managerVM, 3, 3);
 
-    setAlertLevel(managingNode, AlertDetails.getAlertLevelAsString(Alert.OFF));
+    resetAlertCounts(this.managerVM);
 
-    for (VM vm : getManagedNodeList()) {
-      ensureLoggerState(vm, managingMember, Alert.OFF);
-      warnLevelAlert(vm);
-      severeLevelAlert(vm);
-    }
-    checkAlertCount(managingNode, 0, 0);
-    resetAlertCounts(managingNode);
+    setAlertLevel(this.managerVM, AlertDetails.getAlertLevelAsString(Alert.OFF));
 
-    for (VM vm : getManagedNodeList()) {
-      closeCache(vm);
+    for (VM memberVM : this.memberVMs) {
+      verifyAlertAppender(memberVM, managerDistributedMember, Alert.OFF);
+      generateWarningAlert(memberVM);
+      generateSevereAlert(memberVM);
     }
 
-    closeCache(managingNode);
-
+    verifyAlertCount(this.managerVM, 0, 0);
   }
 
-  @SuppressWarnings("serial")
-  public void checkAlertCount(VM vm1, final int expectedSevereAlertCount,
-      final int expectedWarningAlertCount) throws Exception {
-    {
-      vm1.invoke(new SerializableCallable("Check Alert Count") {
-
-        public Object call() throws Exception {
-          final AlertNotifListener nt = AlertNotifListener.getInstance();
-          Wait.waitForCriterion(new WaitCriterion() {
-            public String description() {
-              return "Waiting for all alerts to reach the Managing Node";
-            }
-
-            public boolean done() {
-              if (expectedSevereAlertCount == nt.getseverAlertCount()
-                  && expectedWarningAlertCount == nt.getWarnigAlertCount()) {
-                return true;
-              } else {
-                return false;
-              }
-
-            }
-
-          }, MAX_WAIT, 500, true);
-
-          return null;
-        }
-      });
-
-    }
+  private void verifyAlertAppender(final VM memberVM, final DistributedMember member,
+      final int alertLevel) {
+    memberVM.invoke("verifyAlertAppender",
+        () -> await().until(
+            () -> assertThat(AlertAppender.getInstance().hasAlertListener(member, alertLevel))
+                .isTrue()));
   }
 
+  private void verifyAlertCount(final VM managerVM, final int expectedSevereAlertCount,
+      final int expectedWarningAlertCount) {
+    managerVM.invoke("verifyAlertCount", () -> {
+      AlertNotificationListener listener = AlertNotificationListener.getInstance();
 
-
-  @SuppressWarnings("serial")
-  public void setAlertLevel(VM vm1, final String alertLevel) throws Exception {
-    {
-      vm1.invoke(new SerializableCallable("Set Alert level") {
-
-        public Object call() throws Exception {
-          GemFireCacheImpl cache = GemFireCacheImpl.getInstance();
-          ManagementService service = getManagementService();
-          DistributedSystemMXBean bean = service.getDistributedSystemMXBean();
-          assertNotNull(bean);
-          bean.changeAlertLevel(alertLevel);
-
-          return null;
-        }
-      });
-
-    }
+      await().until(
+          () -> assertThat(listener.getSevereAlertCount()).isEqualTo(expectedSevereAlertCount));
+      await().until(
+          () -> assertThat(listener.getWarningAlertCount()).isEqualTo(expectedWarningAlertCount));
+    });
   }
 
-  @SuppressWarnings("serial")
-  public void warnLevelAlert(VM vm1) throws Exception {
-    {
-      vm1.invoke(new SerializableCallable("Warning level Alerts") {
-
-        public Object call() throws Exception {
-          final IgnoredException warnEx =
-              IgnoredException.addIgnoredException(WARNING_LEVEL_MESSAGE);
-          logger.warn(WARNING_LEVEL_MESSAGE);
-          warnEx.remove();
-          return null;
-        }
-      });
-
-    }
+  private void setAlertLevel(final VM managerVM, final String alertLevel) {
+    managerVM.invoke("setAlertLevel", () -> {
+      ManagementService service = this.managementTestRule.getManagementService();
+      DistributedSystemMXBean distributedSystemMXBean = service.getDistributedSystemMXBean();
+      distributedSystemMXBean.changeAlertLevel(alertLevel);
+    });
   }
 
-
-  @SuppressWarnings("serial")
-  public void resetAlertCounts(VM vm1) throws Exception {
-    {
-      vm1.invoke(new SerializableCallable("Reset Alert Count") {
-
-        public Object call() throws Exception {
-          AlertNotifListener nt = AlertNotifListener.getInstance();
-          nt.resetCount();
-          return null;
-        }
-      });
-
-    }
+  private void generateWarningAlert(final VM anyVM) {
+    anyVM.invoke("generateWarningAlert", () -> {
+      IgnoredException ignoredException =
+          IgnoredException.addIgnoredException(WARNING_LEVEL_MESSAGE);
+      logger.warn(WARNING_LEVEL_MESSAGE);
+      ignoredException.remove();
+    });
   }
 
-  @SuppressWarnings("serial")
-  public void severeLevelAlert(VM vm1) throws Exception {
-    {
-      vm1.invoke(new SerializableCallable("Severe Level Alert") {
-
-        public Object call() throws Exception {
-          // add expected exception strings
-
-          final IgnoredException severeEx =
-              IgnoredException.addIgnoredException(SEVERE_LEVEL_MESSAGE);
-          logger.fatal(SEVERE_LEVEL_MESSAGE);
-          severeEx.remove();
-          return null;
-        }
-      });
-
-    }
+  private void resetAlertCounts(final VM managerVM) {
+    managerVM.invoke("resetAlertCounts", () -> {
+      AlertNotificationListener listener = AlertNotificationListener.getInstance();
+      listener.resetCount();
+    });
   }
 
-  @SuppressWarnings("serial")
-  public void addAlertListener(VM vm1) throws Exception {
-    {
-      vm1.invoke(new SerializableCallable("Add Alert Listener") {
-
-        public Object call() throws Exception {
-          GemFireCacheImpl cache = GemFireCacheImpl.getInstance();
-          ManagementService service = getManagementService();
-          DistributedSystemMXBean bean = service.getDistributedSystemMXBean();
-          AlertNotifListener nt = AlertNotifListener.getInstance();
-          nt.resetCount();
-
-          NotificationFilter notificationFilter = new NotificationFilter() {
-            @Override
-            public boolean isNotificationEnabled(Notification notification) {
-              return notification.getType().equals(JMXNotificationType.SYSTEM_ALERT);
-            }
-
-          };
+  private void generateSevereAlert(final VM anyVM) {
+    anyVM.invoke("generateSevereAlert", () -> {
+      IgnoredException ignoredException =
+          IgnoredException.addIgnoredException(SEVERE_LEVEL_MESSAGE);
+      logger.fatal(SEVERE_LEVEL_MESSAGE);
+      ignoredException.remove();
+    });
+  }
 
-          mbeanServer.addNotificationListener(MBeanJMXAdapter.getDistributedSystemName(), nt,
-              notificationFilter, null);
+  private void addAlertListener(final VM managerVM) {
+    managerVM.invoke("addAlertListener", () -> {
+      AlertNotificationListener listener = AlertNotificationListener.getInstance();
+      listener.resetCount();
 
-          return null;
-        }
-      });
+      NotificationFilter notificationFilter = (Notification notification) -> notification.getType()
+          .equals(JMXNotificationType.SYSTEM_ALERT);
 
-    }
+      ManagementFactory.getPlatformMBeanServer().addNotificationListener(
+          MBeanJMXAdapter.getDistributedSystemName(), listener, notificationFilter, null);
+    });
   }
 
   /**
    * Check aggregate related functions and attributes
-   * 
-   * @param vm1
-   * @throws Exception
    */
-  @SuppressWarnings("serial")
-  public void checkAggregate(VM vm1) throws Exception {
-    {
-      vm1.invoke(new SerializableCallable("Chech Aggregate Attributes") {
-
-        public Object call() throws Exception {
-          GemFireCacheImpl cache = GemFireCacheImpl.getInstance();
-
-          ManagementService service = getManagementService();
-
-          final DistributedSystemMXBean bean = service.getDistributedSystemMXBean();
-          assertNotNull(service.getDistributedSystemMXBean());
-
-          Wait.waitForCriterion(new WaitCriterion() {
-            public String description() {
-              return "Waiting All members to intitialize DistributedSystemMBean expect 5 but found "
-                  + bean.getMemberCount();
-            }
-
-            public boolean done() {
-              // including locator
-              if (bean.getMemberCount() == 5) {
-                return true;
-              } else {
-                return false;
-              }
-
-            }
-
-          }, MAX_WAIT, 500, true);
-
-
+  private void verifyDistributedSystemMXBean(final VM managerVM) {
+    managerVM.invoke("verifyDistributedSystemMXBean", () -> {
+      ManagementService service = this.managementTestRule.getManagementService();
+      DistributedSystemMXBean distributedSystemMXBean = service.getDistributedSystemMXBean();
 
-          final Set<DistributedMember> otherMemberSet =
-              cache.getDistributionManager().getOtherNormalDistributionManagerIds();
-          Iterator<DistributedMember> memberIt = otherMemberSet.iterator();
-          while (memberIt.hasNext()) {
-            DistributedMember member = memberIt.next();
-            LogWriterUtils.getLogWriter().info("JVM Metrics For Member " + member.getId() + ":"
-                + bean.showJVMMetrics(member.getId()));
-            LogWriterUtils.getLogWriter().info("OS Metrics For Member " + member.getId() + ":"
-                + bean.showOSMetrics(member.getId()));
-          }
+      await().until(() -> assertThat(distributedSystemMXBean.getMemberCount()).isEqualTo(5));
 
-          return null;
-        }
-      });
-
-    }
+      Set<DistributedMember> otherMemberSet = this.managementTestRule.getOtherNormalMembers();
+      for (DistributedMember member : otherMemberSet) {
+        // TODO: need assertions? JVMMetrics and OSMetrics
+      }
+    });
   }
 
-  @SuppressWarnings("serial")
-  public void addNotificationListener(VM vm1) throws Exception {
-    {
-      vm1.invoke(new SerializableCallable("Add Notification Listener") {
-
-        public Object call() throws Exception {
-          GemFireCacheImpl cache = GemFireCacheImpl.getInstance();
-          ManagementService service = getManagementService();
-          DistributedSystemMXBean bean = service.getDistributedSystemMXBean();
-          assertNotNull(bean);
-          TestDistributedSystemNotif nt = new TestDistributedSystemNotif();
-          mbeanServer.addNotificationListener(MBeanJMXAdapter.getDistributedSystemName(), nt, null,
-              null);
-
-          return null;
-        }
-      });
+  private void addNotificationListener(final VM managerVM) {
+    managerVM.invoke("addNotificationListener", () -> {
+      ManagementService service = this.managementTestRule.getManagementService();
+      DistributedSystemMXBean distributedSystemMXBean = service.getDistributedSystemMXBean();
+      assertThat(distributedSystemMXBean).isNotNull();
 
-    }
+      DistributedSystemNotificationListener listener = new DistributedSystemNotificationListener();
+      ManagementFactory.getPlatformMBeanServer().addNotificationListener(
+          MBeanJMXAdapter.getDistributedSystemName(), listener, null, null);
+    });
   }
 
+  private void shutDownAll(final VM managerVM) {
+    managerVM.invoke("shutDownAll", () -> {
+      ManagementService service = this.managementTestRule.getManagementService();
+      DistributedSystemMXBean distributedSystemMXBean = service.getDistributedSystemMXBean();
+      distributedSystemMXBean.shutDownAllMembers();
 
-
-  @SuppressWarnings("serial")
-  public void shutDownAll(VM vm1) throws Exception {
-    {
-      vm1.invoke(new SerializableCallable("Shut Down All") {
-
-        public Object call() throws Exception {
-          GemFireCacheImpl cache = GemFireCacheImpl.getInstance();
-          ManagementService service = getManagementService();
-          DistributedSystemMXBean bean = service.getDistributedSystemMXBean();
-          assertNotNull(service.getDistributedSystemMXBean());
-          bean.shutDownAllMembers();
-          Wait.pause(2000);
-          assertEquals(cache.getDistributedSystem().getAllOtherMembers().size(), 1);
-          return null;
-        }
-      });
-
-    }
+      await().until(() -> assertThat(this.managementTestRule.getOtherNormalMembers()).hasSize(0));
+    });
   }
 
+  private void verifyFetchMemberObjectName(final VM managerVM, final int memberCount) {
+    managerVM.invoke("verifyFetchMemberObjectName", () -> {
+      ManagementService service = this.managementTestRule.getManagementService();
+      DistributedSystemMXBean distributedSystemMXBean = service.getDistributedSystemMXBean();
 
+      await().until(
+          () -> assertThat(distributedSystemMXBean.listMemberObjectNames()).hasSize(memberCount));
 
-  @SuppressWarnings("serial")
-  public void checkNavigationAPIs(VM vm1) throws Exception {
-    {
-      vm1.invoke(new SerializableCallable("Check Navigation APIS") {
-
-        public Object call() throws Exception {
-          GemFireCacheImpl cache = GemFireCacheImpl.getInstance();
-          ManagementService service = getManagementService();
-          final DistributedSystemMXBean bean = service.getDistributedSystemMXBean();
-
-          assertNotNull(service.getDistributedSystemMXBean());
-
-          waitForAllMembers(4);
-
-          for (int i = 0; i < bean.listMemberObjectNames().length; i++) {
-            LogWriterUtils.getLogWriter()
-                .info("ObjectNames Of the Mmeber" + bean.listMemberObjectNames()[i]);
-          }
-
-
-          ObjectName thisMemberName = MBeanJMXAdapter.getMemberMBeanName(
-              InternalDistributedSystem.getConnectedInstance().getDistributedMember().getId());
-
-          ObjectName memberName = bean.fetchMemberObjectName(
-              InternalDistributedSystem.getConnectedInstance().getDistributedMember().getId());
-          assertEquals(thisMemberName, memberName);
-
-          return null;
-        }
-      });
-
-    }
+      String memberId = this.managementTestRule.getDistributedMember().getId();
+      ObjectName thisMemberName = MBeanJMXAdapter.getMemberMBeanName(memberId);
+      ObjectName memberName = distributedSystemMXBean.fetchMemberObjectName(memberId);
+      assertThat(memberName).isEqualTo(thisMemberName);
+    });
   }
 
+  private ConditionFactory await() {
+    return Awaitility.await().atMost(2, MINUTES);
+  }
 
-  /**
-   * Notification handler
-   * 
-   * 
-   */
-  private static class TestDistributedSystemNotif implements NotificationListener {
+  private static class DistributedSystemNotificationListener implements NotificationListener {
 
     @Override
-    public void handleNotification(Notification notification, Object handback) {
-      assertNotNull(notification);
+    public void handleNotification(final Notification notification, final Object handback) {
+      assertThat(notification).isNotNull();
     }
-
   }
 
-  /**
-   * Notification handler
-   * 
-   * 
-   */
-  private static class AlertNotifListener implements NotificationListener {
+  private static class AlertNotificationListener implements NotificationListener {
+
+    private static AlertNotificationListener listener = new AlertNotificationListener();
+
+    private int warningAlertCount = 0;
 
-    private static AlertNotifListener listener = new AlertNotifListener();
+    private int severeAlertCount = 0;
 
-    public static AlertNotifListener getInstance() {
+    static AlertNotificationListener getInstance() { // TODO: get rid of singleton
       return listener;
     }
 
-    private int warnigAlertCount = 0;
+    @Override
+    public synchronized void handleNotification(final Notification notification,
+        final Object handback) {
+      assertThat(notification).isNotNull();
 
-    private int severAlertCount = 0;
+      Map<String, String> notificationUserData = (Map<String, String>) notification.getUserData();
 
-    @Override
-    public synchronized void handleNotification(Notification notification, Object handback) {
-      assertNotNull(notification);
-      logger.info("Notification received {}", notification);
-      Map<String, String> notifUserData = (Map<String, String>) notification.getUserData();
-      if (notifUserData.get(JMXNotificationUserData.ALERT_LEVEL).equalsIgnoreCase("warning")) {
-        assertEquals(WARNING_LEVEL_MESSAGE, notification.getMessage());
-        ++warnigAlertCount;
+      if (notificationUserData.get(JMXNotificationUserData.ALERT_LEVEL)
+          .equalsIgnoreCase("warning")) {
+        assertThat(notification.getMessage()).isEqualTo(WARNING_LEVEL_MESSAGE);
+        warningAlertCount++;
       }
-      if (notifUserData.get(JMXNotificationUserData.ALERT_LEVEL).equalsIgnoreCase("severe")) {
-        assertEquals(SEVERE_LEVEL_MESSAGE, notification.getMessage());
-        ++severAlertCount;
+      if (notificationUserData.get(JMXNotificationUserData.ALERT_LEVEL)
+          .equalsIgnoreCase("severe")) {
+        assertThat(notification.getMessage()).isEqualTo(SEVERE_LEVEL_MESSAGE);
+        severeAlertCount++;
       }
     }
 
-    public void resetCount() {
-      warnigAlertCount = 0;
-
-      severAlertCount = 0;
+    void resetCount() {
+      warningAlertCount = 0;
+      severeAlertCount = 0;
     }
 
-    public int getWarnigAlertCount() {
-      return warnigAlertCount;
+    int getWarningAlertCount() {
+      return warningAlertCount;
     }
 
-    public int getseverAlertCount() {
-      return severAlertCount;
+    int getSevereAlertCount() {
+      return severeAlertCount;
     }
-
   }
-
 }


Mime
View raw message