/*
 * 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;

import static java.lang.management.ManagementFactory.getPlatformMBeanServer;
import static java.util.Calendar.MONTH;
import static org.apache.geode.cache.EvictionAction.LOCAL_DESTROY;
import static org.apache.geode.cache.Region.SEPARATOR;
import static org.apache.geode.cache.RegionShortcut.LOCAL;
import static org.apache.geode.cache.RegionShortcut.PARTITION;
import static org.apache.geode.cache.RegionShortcut.PARTITION_REDUNDANT;
import static org.apache.geode.cache.RegionShortcut.REPLICATE;
import static org.apache.geode.distributed.ConfigurationProperties.ENABLE_TIME_STATISTICS;
import static org.apache.geode.distributed.ConfigurationProperties.HTTP_SERVICE_PORT;
import static org.apache.geode.distributed.ConfigurationProperties.JMX_MANAGER;
import static org.apache.geode.distributed.ConfigurationProperties.JMX_MANAGER_PORT;
import static org.apache.geode.distributed.ConfigurationProperties.JMX_MANAGER_START;
import static org.apache.geode.distributed.ConfigurationProperties.STATISTIC_SAMPLING_ENABLED;
import static org.apache.geode.management.internal.MBeanJMXAdapter.getDistributedRegionMbeanName;
import static org.apache.geode.management.internal.MBeanJMXAdapter.getMemberMBeanName;
import static org.apache.geode.management.internal.MBeanJMXAdapter.getRegionMBeanName;
import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
import static org.apache.geode.test.dunit.Disconnect.disconnectAllFromDS;
import static org.apache.geode.test.dunit.Invoke.invokeInEveryVM;
import static org.apache.geode.test.dunit.VM.getVM;
import static org.apache.geode.test.dunit.VM.toArray;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Calendar;
import java.util.Date;
import java.util.HashSet;
import java.util.List;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;

import javax.management.InstanceNotFoundException;
import javax.management.MalformedObjectNameException;
import javax.management.Notification;
import javax.management.NotificationListener;
import javax.management.ObjectName;

import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;

import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheListener;
import org.apache.geode.cache.EntryOperation;
import org.apache.geode.cache.EvictionAttributes;
import org.apache.geode.cache.FixedPartitionAttributes;
import org.apache.geode.cache.FixedPartitionResolver;
import org.apache.geode.cache.PartitionAttributes;
import org.apache.geode.cache.PartitionAttributesFactory;
import org.apache.geode.cache.PartitionResolver;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionAttributes;
import org.apache.geode.cache.RegionFactory;
import org.apache.geode.cache.query.data.Portfolio;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.InternalRegion;
import org.apache.geode.internal.cache.TestObjectSizerImpl;
import org.apache.geode.internal.cache.eviction.EvictionCounters;
import org.apache.geode.management.internal.MBeanJMXAdapter;
import org.apache.geode.management.internal.SystemManagementService;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.rules.CacheRule;
import org.apache.geode.test.dunit.rules.DistributedRule;
import org.apache.geode.test.junit.rules.serializable.SerializableTestName;

/**
 * Distributed tests for {@link RegionMXBean}.
 *
 * <p>
 * This class checks and verifies various data and operations exposed through RegionMXBean
 * interface.
 *
 * <p>
 * Goal of the Test : RegionMBean gets created once region is created. Data like Region Attributes
 * data and stats are of proper value
 */
@SuppressWarnings("serial")
public class RegionManagementDUnitTest implements Serializable {

  private static final AtomicReference<List<Notification>> MEMBER_NOTIFICATIONS =
      new AtomicReference<>();
  private static final AtomicReference<List<Notification>> SYSTEM_NOTIFICATIONS =
      new AtomicReference<>();

  private String regionName;
  private String partitionedRegionName;
  private String subregionName;

  private VM managerVM;

  private VM[] memberVMs;

  @Rule
  public DistributedRule distributedRule = new DistributedRule();

  @Rule
  public CacheRule cacheRule = new CacheRule();

  @Rule
  public SerializableTestName testName = new SerializableTestName();

  @Before
  public void setUp() {
    String uniqueName = getClass().getSimpleName() + "_" + testName.getMethodName();
    regionName = uniqueName + "_region";
    partitionedRegionName = uniqueName + "_partitionedRegion";
    subregionName = uniqueName + "_subregion";

    managerVM = getVM(0);
    VM memberVM1 = getVM(1);
    VM memberVM2 = getVM(2);
    VM memberVM3 = getVM(3);

    memberVM1.invoke(() -> createMember(cacheRule));
    memberVM2.invoke(() -> createMember(cacheRule));
    memberVM3.invoke(() -> createMember(cacheRule));

    managerVM.invoke(() -> createManager(cacheRule));

    memberVMs = toArray(memberVM1, memberVM2, memberVM3);
  }

  @After
  public void tearDown() {
    disconnectAllFromDS();

    MEMBER_NOTIFICATIONS.set(null);
    SYSTEM_NOTIFICATIONS.set(null);
    invokeInEveryVM(() -> MEMBER_NOTIFICATIONS.set(null));
    invokeInEveryVM(() -> SYSTEM_NOTIFICATIONS.set(null));
  }

  /**
   * Tests all Region MBean related Management APIs:
   * <p>
   * a) Notification propagated to member MBean while a region is created<br>
   * b) Creates and check a Distributed Region
   */
  @Test
  public void testDistributedRegion() {
    // Adding notification listener for remote cache memberVMs
    managerVM.invoke(() -> addMemberNotificationListener(3));

    for (VM memberVM : memberVMs) {
      memberVM.invoke(() -> {
        getCache().createRegionFactory(REPLICATE).create(regionName);
        verifyReplicateRegionAfterCreate(regionName);
      });
    }

    managerVM.invoke(() -> verifyRemoteDistributedRegion(3, regionName));

    for (VM memberVM : memberVMs) {
      memberVM.invoke(() -> {
        getCache().getRegion(regionName).close();
        verifyReplicatedRegionAfterClose(regionName);
      });
    }

    managerVM.invoke(() -> verifyProxyCleanup(regionName));

    managerVM.invoke(() -> verifyMemberNotifications(regionName, 3));
  }

  /**
   * Tests all Region MBean related Management APIs:
   * <p>
   * a) Notification propagated to member MBean while a region is created<br>
   * b) Created and check a Partitioned Region
   */
  @Test
  public void testPartitionedRegion() {
    // Adding notification listener for remote cache memberVMs
    managerVM.invoke(() -> addMemberNotificationListener(3));

    for (VM memberVM : memberVMs) {
      memberVM.invoke(() -> {
        getCache().createRegionFactory(PARTITION_REDUNDANT).create(partitionedRegionName);
        verifyPartitionRegionAfterCreate(partitionedRegionName, false);
      });
    }

    managerVM.invoke(() -> verifyRemotePartitionRegion(partitionedRegionName));

    for (VM memberVM : memberVMs) {
      memberVM.invoke(() -> {
        getCache().getRegion(partitionedRegionName).close();
        verifyPartitionRegionAfterClose(partitionedRegionName);
      });
    }

    managerVM.invoke(() -> verifyMemberNotifications(partitionedRegionName, 3));
  }

  /**
   * Tests all Region MBean related Management APIs:
   * <p>
   * a) Notification propagated to member MBean while a region is created<br>
   * b) Creates and check a Fixed Partitioned Region
   */
  @Test
  public void testFixedPRRegionMBean() {
    // Adding notification listener for remote cache memberVMs
    managerVM.invoke(() -> addMemberNotificationListener(3));

    int primaryIndex = 0;
    for (VM memberVM : memberVMs) {
      List<FixedPartitionAttributes> fixedPartitionAttributesList =
          createFixedPartitionList(primaryIndex + 1);
      memberVM.invoke(
          () -> createFixedPartitionRegion(partitionedRegionName, fixedPartitionAttributesList));
      primaryIndex++;
    }

    managerVM.invoke(() -> verifyRemoteFixedPartitionRegion(partitionedRegionName));

    for (VM memberVM : memberVMs) {
      memberVM.invoke(() -> getCache().getRegion(partitionedRegionName).close());
    }

    managerVM.invoke(() -> verifyMemberNotifications(partitionedRegionName, 3));
  }

  /**
   * Tests a Distributed Region at Managing Node side while region is created in a member node
   * asynchronously.
   */
  @Test
  public void testRegionAggregate() {
    // Adding notification listener for remote cache memberVMs
    managerVM.invoke(this::addSystemNotificationListener);

    for (VM memberVM : memberVMs) {
      memberVM.invoke(() -> {
        getCache().createRegionFactory(REPLICATE).create(regionName);
      });
    }

    managerVM.invoke(() -> {
      verifyDistributedMBean(regionName, 3);
      getCache().createRegionFactory(REPLICATE).create(regionName);
      verifyDistributedMBean(regionName, 4);
    });

    for (VM memberVM : memberVMs) {
      memberVM.invoke(() -> getCache().getRegion(regionName).close());
    }

    managerVM.invoke(() -> {
      verifyProxyCleanup(regionName);

      verifyDistributedMBean(regionName, 1);
      getCache().getRegion(regionName).close();
      verifyDistributedMBean(regionName, 0);

      // TODO: GEODE-1930: verifySystemNotifications is too flaky and needs to be fixed
      // verifySystemNotifications(regionName, 3);
    });
  }

  @Test
  public void testNavigationAPIS() {
    for (VM memberVM : memberVMs) {
      memberVM.invoke(() -> {
        getCache().createRegionFactory(REPLICATE).create(regionName);
        getCache().createRegionFactory(PARTITION_REDUNDANT).create(partitionedRegionName);
      });
    }

    managerVM.invoke(() -> {
      getCache().createRegionFactory(REPLICATE).create(regionName);
      getCache().createRegionFactory(PARTITION_REDUNDANT).create(partitionedRegionName);
    });

    List<String> memberIds = new ArrayList<>();
    for (VM memberVM : memberVMs) {
      memberIds.add(
          memberVM.invoke(() -> getCache().getDistributedSystem().getDistributedMember().getId()));
    }

    managerVM.invoke(() -> verifyNavigationApis(memberIds));
  }

  @Test
  public void testSubRegions() {
    for (VM memberVM : memberVMs) {
      memberVM.invoke(() -> {
        Region region = getCache().createRegionFactory(LOCAL).create(regionName);
        getCache().createRegionFactory(LOCAL).createSubregion(region, subregionName);
      });
    }

    for (VM memberVM : memberVMs) {
      memberVM.invoke(() -> verifyRegionMXBeanIsNotNull(regionName, subregionName));
    }

    for (VM memberVM : memberVMs) {
      memberVM.invoke(() -> {
        getCache().getRegion(regionName).close();
        verifyRegionMXBeanIsNull(regionName, subregionName);
      });
    }
  }

  @Test
  public void testSpecialRegions() {
    memberVMs[0].invoke(() -> {
      RegionFactory<?, Portfolio> regionFactory = getCache().createRegionFactory(LOCAL);
      regionFactory.setValueConstraint(Portfolio.class);

      regionFactory.create("p-os");
      regionFactory.create("p_os");
    });

    managerVM.invoke(() -> {
      awaitDistributedRegionMXBean("p-os", 1);
      awaitDistributedRegionMXBean("p_os", 1);
    });
  }

  @Test
  public void testLruStats() {
    for (VM memberVM : memberVMs) {
      memberVM.invoke(() -> createDiskRegion(regionName));
    }

    managerVM.invoke(() -> verifyEntrySize(regionName, 3));
  }

  private InternalCache getCache() {
    return cacheRule.getCache();
  }

  private ManagementService getManagementService() {
    return ManagementService.getManagementService(cacheRule.getCache());
  }

  private SystemManagementService getSystemManagementService() {
    return (SystemManagementService) getManagementService();
  }

  private Set<DistributedMember> getOtherMembers() {
    Set<DistributedMember> allMembers =
        new HashSet<>(getCache().getDistributionManager().getNormalDistributionManagerIds());
    allMembers.remove(getCache().getDistributionManager().getId());
    return allMembers;
  }

  private String toPath(final String regionName) {
    return SEPARATOR + regionName;
  }

  private String toPath(final String regionName, final String subregionName) {
    return SEPARATOR + regionName + SEPARATOR + subregionName;
  }

  private void createDiskRegion(final String regionName) {
    EvictionAttributes evictionAttributes = EvictionAttributes.createLRUMemoryAttributes(20,
        new TestObjectSizerImpl(), LOCAL_DESTROY);

    RegionFactory<Integer, Object> regionFactory = getCache().createRegionFactory(LOCAL);
    regionFactory.setEvictionAttributes(evictionAttributes);

    Region<Integer, Object> region = regionFactory.create(regionName);

    EvictionCounters lruStats = ((InternalRegion) region).getEvictionController().getCounters();
    assertThat(lruStats).isNotNull();

    RegionMXBean regionMXBean = getManagementService().getLocalRegionMBean(toPath(regionName));
    assertThat(regionMXBean).isNotNull();

    for (int total = 0; total < 10; total++) {
      int[] array = new int[250];
      array[0] = total;
      region.put(total, array);
    }
    assertThat(regionMXBean.getEntrySize()).isGreaterThan(0);
  }

  private void createFixedPartitionRegion(final String regionName,
      final List<FixedPartitionAttributes> fixedPartitionAttributesList) {
    PartitionAttributesFactory<?, ?> partitionAttributesFactory = new PartitionAttributesFactory();
    partitionAttributesFactory.setRedundantCopies(2);
    partitionAttributesFactory.setTotalNumBuckets(12);

    for (FixedPartitionAttributes fixedPartitionAttributes : fixedPartitionAttributesList) {
      partitionAttributesFactory.addFixedPartitionAttributes(fixedPartitionAttributes);
    }

    partitionAttributesFactory.setPartitionResolver(new QuarterlyFixedPartitionResolver<>());

    RegionFactory<?, ?> regionFactory = getCache().createRegionFactory(PARTITION);
    regionFactory.setPartitionAttributes(partitionAttributesFactory.create());

    Region<?, ?> region = regionFactory.create(regionName);

    SystemManagementService service = getSystemManagementService();
    RegionMXBean regionMXBean = service.getLocalRegionMBean(toPath(regionName));
    RegionAttributes regionAttributes = region.getAttributes();

    PartitionAttributesData partitionAttributesData = regionMXBean.listPartitionAttributes();
    verifyPartitionData(regionAttributes, partitionAttributesData, true);

    FixedPartitionAttributesData[] fixedPartitionAttributesData =
        regionMXBean.listFixedPartitionAttributes();
    assertThat(fixedPartitionAttributesData).isNotNull();
    assertThat(fixedPartitionAttributesData).hasSize(3);

    for (FixedPartitionAttributesData aFixedPartitionAttributesData : fixedPartitionAttributesData) {
      // TODO: add real assertions for FixedPartitionAttributesData
      // LogWriterUtils.getLogWriter().info("<ExpectedString> Fixed PR Data is " +
      // fixedPartitionAttributesData[i] + "</ExpectedString> ");
    }
  }

  private List<FixedPartitionAttributes> createFixedPartitionList(final int primaryIndex) {
    List<FixedPartitionAttributes> fixedPartitionAttributesList = new ArrayList<>();
    if (primaryIndex == 1) {
      fixedPartitionAttributesList
          .add(FixedPartitionAttributes.createFixedPartition("Q1", true, 3));
      fixedPartitionAttributesList.add(FixedPartitionAttributes.createFixedPartition("Q2", 3));
      fixedPartitionAttributesList.add(FixedPartitionAttributes.createFixedPartition("Q3", 3));
    }
    if (primaryIndex == 2) {
      fixedPartitionAttributesList.add(FixedPartitionAttributes.createFixedPartition("Q1", 3));
      fixedPartitionAttributesList
          .add(FixedPartitionAttributes.createFixedPartition("Q2", true, 3));
      fixedPartitionAttributesList.add(FixedPartitionAttributes.createFixedPartition("Q3", 3));
    }
    if (primaryIndex == 3) {
      fixedPartitionAttributesList.add(FixedPartitionAttributes.createFixedPartition("Q1", 3));
      fixedPartitionAttributesList.add(FixedPartitionAttributes.createFixedPartition("Q2", 3));
      fixedPartitionAttributesList
          .add(FixedPartitionAttributes.createFixedPartition("Q3", true, 3));
    }
    return fixedPartitionAttributesList;
  }

  private void addMemberNotificationListener(final int expectedMembers)
      throws InstanceNotFoundException {
    Set<DistributedMember> otherMemberSet = getOtherMembers();
    assertThat(otherMemberSet).hasSize(expectedMembers);

    SystemManagementService service = getSystemManagementService();

    List<Notification> notifications = new ArrayList<>();
    MEMBER_NOTIFICATIONS.set(notifications);

    for (DistributedMember member : otherMemberSet) {
      MemberNotificationListener listener = new MemberNotificationListener(notifications);
      ObjectName objectName = service.getMemberMBeanName(member);
      awaitMemberMXBeanProxy(objectName);

      getPlatformMBeanServer().addNotificationListener(objectName, listener, null, null);
    }
  }

  /**
   * Add a Notification listener to DistributedSystemMBean which should gather all the notifications
   * which are propagated through all individual MemberMBeans Hence Region created/destroyed should
   * be visible to this listener
   */
  private void addSystemNotificationListener() throws InstanceNotFoundException {
    awaitDistributedSystemMXBean();

    List<Notification> notifications = new ArrayList<>();
    SYSTEM_NOTIFICATIONS.set(notifications);

    DistributedSystemNotificationListener listener =
        new DistributedSystemNotificationListener(notifications);
    ObjectName objectName = MBeanJMXAdapter.getDistributedSystemName();
    getPlatformMBeanServer().addNotificationListener(objectName, listener, null, null);
  }

  private void awaitMemberCount(final int expectedCount) {
    DistributedSystemMXBean distributedSystemMXBean = awaitDistributedSystemMXBean();
    await()
        .untilAsserted(
            () -> assertThat(distributedSystemMXBean.getMemberCount()).isEqualTo(expectedCount));
  }

  private DistributedRegionMXBean awaitDistributedRegionMXBean(final String regionName,
      final int memberCount) {
    SystemManagementService service = getSystemManagementService();

    await().untilAsserted(
        () -> assertThat(service.getDistributedRegionMXBean(toPath(regionName))).isNotNull());
    await()
        .untilAsserted(() -> assertThat(
            service.getDistributedRegionMXBean(toPath(regionName)).getMemberCount())
                .isEqualTo(memberCount));

    return service.getDistributedRegionMXBean(toPath(regionName));
  }

  private RegionMXBean awaitRegionMXBeanProxy(final DistributedMember member,
      final String regionName) {
    SystemManagementService service = getSystemManagementService();
    ObjectName objectName = service.getRegionMBeanName(member, toPath(regionName));
    String alias = "awaiting RegionMXBean proxy for " + member;

    await(alias)
        .untilAsserted(
            () -> assertThat(service.getMBeanProxy(objectName, RegionMXBean.class)).isNotNull());

    return service.getMBeanProxy(objectName, RegionMXBean.class);
  }

  private RegionMXBean awaitRegionMXBeanProxy(final ObjectName objectName) {
    SystemManagementService service = getSystemManagementService();

    await()
        .untilAsserted(
            () -> assertThat(service.getMBeanProxy(objectName, RegionMXBean.class)).isNotNull());

    return service.getMBeanProxy(objectName, RegionMXBean.class);
  }

  private MemberMXBean awaitMemberMXBeanProxy(final DistributedMember member) {
    SystemManagementService service = getSystemManagementService();
    ObjectName objectName = service.getMemberMBeanName(member);
    String alias = "awaiting MemberMXBean proxy for " + member;

    await(alias)
        .untilAsserted(
            () -> assertThat(service.getMBeanProxy(objectName, MemberMXBean.class)).isNotNull());

    return service.getMBeanProxy(objectName, MemberMXBean.class);
  }

  private MemberMXBean awaitMemberMXBeanProxy(final ObjectName objectName) {
    SystemManagementService service = getSystemManagementService();
    await()
        .untilAsserted(
            () -> assertThat(service.getMBeanProxy(objectName, MemberMXBean.class)).isNotNull());
    return service.getMBeanProxy(objectName, MemberMXBean.class);
  }

  private DistributedSystemMXBean awaitDistributedSystemMXBean() {
    ManagementService service = getSystemManagementService();

    await().untilAsserted(() -> assertThat(service.getDistributedSystemMXBean()).isNotNull());

    return service.getDistributedSystemMXBean();
  }

  private void verifyEntrySize(String regionName, final int expectedMembers) {
    DistributedRegionMXBean distributedRegionMXBean =
        awaitDistributedRegionMXBean(regionName, expectedMembers);
    assertThat(distributedRegionMXBean).isNotNull();
    assertThat(distributedRegionMXBean.getEntrySize()).isGreaterThan(0);
  }

  private void verifyRegionMXBeanIsNotNull(final String regionName, final String subregionName) {
    RegionMXBean regionMXBean =
        getManagementService().getLocalRegionMBean(toPath(regionName, subregionName));
    assertThat(regionMXBean).isNotNull();
  }

  private void verifyRegionMXBeanIsNull(final String regionName, final String subregionName) {
    RegionMXBean regionMXBean =
        getManagementService().getLocalRegionMBean(toPath(regionName, subregionName));
    assertThat(regionMXBean).isNull();
  }

  private void verifyNavigationApis(final List<String> memberIds) throws Exception {
    ManagementService service = getManagementService();
    assertThat(service.getDistributedSystemMXBean()).isNotNull();

    // With the DUnit framework there is a locator, a manager and 3 members
    awaitMemberCount(5);

    DistributedSystemMXBean distributedSystemMXBean = service.getDistributedSystemMXBean();
    assertThat(distributedSystemMXBean.listDistributedRegionObjectNames()).hasSize(2);

    assertThat(
        distributedSystemMXBean.fetchDistributedRegionObjectName(toPath(partitionedRegionName)))
            .isNotNull();
    assertThat(distributedSystemMXBean.fetchDistributedRegionObjectName(toPath(regionName)))
        .isNotNull();

    ObjectName actualName =
        distributedSystemMXBean.fetchDistributedRegionObjectName(toPath(partitionedRegionName));
    ObjectName expectedName = getDistributedRegionMbeanName(toPath(partitionedRegionName));
    assertThat(actualName).isEqualTo(expectedName);

    actualName = distributedSystemMXBean.fetchDistributedRegionObjectName(toPath(regionName));
    expectedName = getDistributedRegionMbeanName(toPath(regionName));
    assertThat(actualName).isEqualTo(expectedName);

    for (String memberId : memberIds) {
      ObjectName objectName = getMemberMBeanName(memberId);
      awaitMemberMXBeanProxy(objectName);

      ObjectName[] objectNames = distributedSystemMXBean.fetchRegionObjectNames(objectName);
      assertThat(objectNames).isNotNull();
      assertThat(objectNames).hasSize(2);

      List<ObjectName> listOfNames = Arrays.asList(objectNames);

      expectedName = getRegionMBeanName(memberId, toPath(partitionedRegionName));
      assertThat(listOfNames).contains(expectedName);

      expectedName = getRegionMBeanName(memberId, toPath(regionName));
      assertThat(listOfNames).contains(expectedName);
    }

    for (String memberId : memberIds) {
      ObjectName objectName = getMemberMBeanName(memberId);
      awaitMemberMXBeanProxy(objectName);

      expectedName = getRegionMBeanName(memberId, toPath(partitionedRegionName));
      awaitRegionMXBeanProxy(expectedName);

      actualName =
          distributedSystemMXBean.fetchRegionObjectName(memberId, toPath(partitionedRegionName));
      assertThat(actualName).isEqualTo(expectedName);

      expectedName = getRegionMBeanName(memberId, toPath(regionName));
      awaitRegionMXBeanProxy(expectedName);

      actualName = distributedSystemMXBean.fetchRegionObjectName(memberId, toPath(regionName));
      assertThat(actualName).isEqualTo(expectedName);
    }
  }

  private void verifyMemberNotifications(final String regionName, final int expectedMembers) {
    await()
        .untilAsserted(() -> assertThat(MEMBER_NOTIFICATIONS.get()).hasSize(expectedMembers * 2));

    int regionCreatedCount = 0;
    int regionDestroyedCount = 0;
    for (Notification notification : MEMBER_NOTIFICATIONS.get()) {
      if (JMXNotificationType.REGION_CREATED.equals(notification.getType())) {
        regionCreatedCount++;
        assertThat(notification.getMessage()).contains(regionName);
      } else if (JMXNotificationType.REGION_CLOSED.equals(notification.getType())) {
        regionDestroyedCount++;
        assertThat(notification.getMessage()).contains(regionName);
      } else {
        fail("Unexpected notification type: " + notification.getType());
      }
    }

    assertThat(regionCreatedCount).isEqualTo(expectedMembers);
    assertThat(regionDestroyedCount).isEqualTo(expectedMembers);

    // <[javax.management.Notification[source=10.118.33.232(17632)<v1>-32770][type=gemfire.distributedsystem.cache.region.created][message=Region
    // Created With Name /MANAGEMENT_TEST_REGION],
    // javax.management.Notification[source=10.118.33.232(17633)<v2>-32771][type=gemfire.distributedsystem.cache.region.created][message=Region
    // Created With Name /MANAGEMENT_TEST_REGION],
    // javax.management.Notification[source=10.118.33.232(17634)<v3>-32772][type=gemfire.distributedsystem.cache.region.created][message=Region
    // Created With Name /MANAGEMENT_TEST_REGION],
    // javax.management.Notification[source=10.118.33.232(17632)<v1>-32770][type=gemfire.distributedsystem.cache.region.closed][message=Region
    // Destroyed/Closed With Name /MANAGEMENT_TEST_REGION],
    // javax.management.Notification[source=10.118.33.232(17633)<v2>-32771][type=gemfire.distributedsystem.cache.region.closed][message=Region
    // Destroyed/Closed With Name /MANAGEMENT_TEST_REGION],
    // javax.management.Notification[source=10.118.33.232(17634)<v3>-32772][type=gemfire.distributedsystem.cache.region.closed][message=Region
    // Destroyed/Closed With Name /MANAGEMENT_TEST_REGION]]>
  }

  /**
   * Please don't delete verifySystemNotifications. We need to improve verification in this test
   * class and fix whatever flakiness is in this method.
   */
  @SuppressWarnings("unused")
  private void verifySystemNotifications(final String regionName, final int expectedMembers) {
    assertThat(SYSTEM_NOTIFICATIONS.get()).isNotNull();
    assertThat(SYSTEM_NOTIFICATIONS.get()).hasSize(expectedMembers + 2); // 2 for the manager


    int regionCreatedCount = 0;
    int regionDestroyedCount = 0;
    for (Notification notification : SYSTEM_NOTIFICATIONS.get()) {
      if (JMXNotificationType.REGION_CREATED.equals(notification.getType())) {
        regionCreatedCount++;
        assertThat(notification.getMessage()).contains(regionName);
      } else if (JMXNotificationType.REGION_CLOSED.equals(notification.getType())) {
        regionDestroyedCount++;
        assertThat(notification.getMessage()).contains(regionName);
      } else {
        fail("Unexpected notification type: " + notification.getType());
      }
    }

    assertThat(regionCreatedCount).isEqualTo(1); // just the manager
    assertThat(regionDestroyedCount).isEqualTo(expectedMembers + 1); // all 3 members + manager

    // <[javax.management.Notification[source=192.168.1.72(18496)<v27>-32770][type=gemfire.distributedsystem.cache.region.created][message=Region
    // Created With Name /MANAGEMENT_TEST_REGION],
    // javax.management.Notification[source=192.168.1.72(18497)<v28>-32771][type=gemfire.distributedsystem.cache.region.closed][message=Region
    // Destroyed/Closed With Name /MANAGEMENT_TEST_REGION],
    // javax.management.Notification[source=192.168.1.72(18498)<v29>-32772][type=gemfire.distributedsystem.cache.region.closed][message=Region
    // Destroyed/Closed With Name /MANAGEMENT_TEST_REGION],
    // javax.management.Notification[source=192.168.1.72(18499)<v30>-32773][type=gemfire.distributedsystem.cache.region.closed][message=Region
    // Destroyed/Closed With Name /MANAGEMENT_TEST_REGION],
    // javax.management.Notification[source=192.168.1.72(18496)<v27>-32770][type=gemfire.distributedsystem.cache.region.closed][message=Region
    // Destroyed/Closed With Name /MANAGEMENT_TEST_REGION]]>
  }

  private void verifyProxyCleanup(final String regionName) {
    SystemManagementService service = getSystemManagementService();

    Set<DistributedMember> otherMemberSet = getOtherMembers();
    for (DistributedMember member : otherMemberSet) {
      String alias = "Waiting for the proxy to get deleted at managing node";
      await(alias).untilAsserted(
          () -> assertThat(
              service.getMBeanProxy(service.getRegionMBeanName(member, toPath(regionName)),
                  RegionMXBean.class)).isNull());
    }
  }

  private void verifyRemoteDistributedRegion(final int expectedMembers, final String regionName) {
    Set<DistributedMember> otherMemberSet = getOtherMembers();
    assertThat(otherMemberSet).hasSize(expectedMembers);

    for (DistributedMember member : otherMemberSet) {
      RegionMXBean regionMXBean = awaitRegionMXBeanProxy(member, regionName);

      RegionAttributesData regionAttributesData = regionMXBean.listRegionAttributes();
      assertThat(regionAttributesData).isNotNull();

      MembershipAttributesData membershipAttributesData = regionMXBean.listMembershipAttributes();
      assertThat(membershipAttributesData).isNotNull();

      EvictionAttributesData evictionAttributesData = regionMXBean.listEvictionAttributes();
      assertThat(evictionAttributesData).isNotNull();
    }

    DistributedRegionMXBean distributedRegionMXBean =
        awaitDistributedRegionMXBean(regionName, expectedMembers);

    assertThat(distributedRegionMXBean).isNotNull();
    assertThat(distributedRegionMXBean.getFullPath()).isEqualTo(toPath(regionName));
  }

  private void verifyDistributedMBean(final String regionName, final int expectedMembers) {
    if (expectedMembers == 0) {
      ManagementService service = getManagementService();
      String alias = "Waiting for the proxy to get deleted at managing node";
      await(alias)
          .untilAsserted(
              () -> assertThat(service.getDistributedRegionMXBean(toPath(regionName))).isNull());
      return;
    }

    DistributedRegionMXBean distributedRegionMXBean =
        awaitDistributedRegionMXBean(regionName, expectedMembers);

    assertThat(distributedRegionMXBean.getFullPath()).isEqualTo(toPath(regionName));
    assertThat(distributedRegionMXBean.getMemberCount()).isEqualTo(expectedMembers);
    assertThat(distributedRegionMXBean.getMembers()).hasSize(expectedMembers);

    // Check Stats related Data
    // LogWriterUtils.getLogWriter().info("<ExpectedString> CacheListenerCallsAvgLatency is " +
    // distributedRegionMXBean.getCacheListenerCallsAvgLatency() + "</ExpectedString> ");
    // LogWriterUtils.getLogWriter().info("<ExpectedString> CacheWriterCallsAvgLatency is " +
    // distributedRegionMXBean.getCacheWriterCallsAvgLatency() + "</ExpectedString> ");
    // LogWriterUtils.getLogWriter().info("<ExpectedString> CreatesRate is " +
    // distributedRegionMXBean.getCreatesRate() + "</ExpectedString> ");
  }

  private void verifyRemotePartitionRegion(final String regionName) {
    Set<DistributedMember> otherMemberSet = getOtherMembers();

    for (DistributedMember member : otherMemberSet) {
      RegionMXBean regionMXBean = awaitRegionMXBeanProxy(member, regionName);
