/*
 * 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.internal.cache.wan.asyncqueue;

import static java.util.Collections.synchronizedSet;
import static java.util.concurrent.TimeUnit.MINUTES;
import static java.util.concurrent.TimeUnit.SECONDS;
import static java.util.stream.Collectors.toSet;
import static org.apache.geode.cache.FixedPartitionAttributes.createFixedPartition;
import static org.apache.geode.cache.RegionShortcut.PARTITION;
import static org.apache.geode.cache.RegionShortcut.REPLICATE;
import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
import static org.apache.geode.test.dunit.VM.getCurrentVMNum;
import static org.apache.geode.test.dunit.VM.getVM;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;

import java.io.File;
import java.io.IOException;
import java.io.Serializable;
import java.io.UncheckedIOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.IntStream;

import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.junit.runners.Parameterized.Parameter;
import org.junit.runners.Parameterized.Parameters;
import org.junit.runners.Parameterized.UseParametersRunnerFactory;

import org.apache.geode.cache.Cache;
import org.apache.geode.cache.Declarable;
import org.apache.geode.cache.DiskStoreFactory;
import org.apache.geode.cache.EntryOperation;
import org.apache.geode.cache.FixedPartitionAttributes;
import org.apache.geode.cache.FixedPartitionResolver;
import org.apache.geode.cache.PartitionAttributesFactory;
import org.apache.geode.cache.PartitionResolver;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionFactory;
import org.apache.geode.cache.asyncqueue.AsyncEvent;
import org.apache.geode.cache.asyncqueue.AsyncEventListener;
import org.apache.geode.cache.asyncqueue.AsyncEventQueue;
import org.apache.geode.cache.asyncqueue.AsyncEventQueueFactory;
import org.apache.geode.cache.asyncqueue.internal.AsyncEventQueueStats;
import org.apache.geode.cache.asyncqueue.internal.InternalAsyncEventQueue;
import org.apache.geode.cache.control.RebalanceFactory;
import org.apache.geode.cache.control.RebalanceOperation;
import org.apache.geode.cache.control.RebalanceResults;
import org.apache.geode.cache.control.ResourceManager;
import org.apache.geode.cache.partition.PartitionRegionHelper;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.PartitionedRegion;
import org.apache.geode.internal.cache.RegionQueue;
import org.apache.geode.internal.cache.control.InternalResourceManager;
import org.apache.geode.internal.cache.control.InternalResourceManager.ResourceObserver;
import org.apache.geode.internal.cache.control.InternalResourceManager.ResourceObserverAdapter;
import org.apache.geode.internal.cache.partitioned.BecomePrimaryBucketMessage;
import org.apache.geode.internal.cache.partitioned.BecomePrimaryBucketMessage.BecomePrimaryBucketResponse;
import org.apache.geode.internal.cache.wan.AsyncEventQueueConfigurationException;
import org.apache.geode.internal.cache.wan.GatewaySenderEventImpl;
import org.apache.geode.internal.cache.wan.InternalGatewaySender;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.rules.CacheRule;
import org.apache.geode.test.dunit.rules.DistributedRestoreSystemProperties;
import org.apache.geode.test.dunit.rules.DistributedRule;
import org.apache.geode.test.junit.categories.AEQTest;
import org.apache.geode.test.junit.rules.serializable.SerializableTemporaryFolder;
import org.apache.geode.test.junit.runners.CategoryWithParameterizedRunnerFactory;

/**
 * Extracted from {@link AsyncEventListenerDistributedTest}.
 */
@Category(AEQTest.class)
@RunWith(Parameterized.class)
@UseParametersRunnerFactory(CategoryWithParameterizedRunnerFactory.class)
@SuppressWarnings("serial")
public class ParallelAsyncEventListenerDistributedTest implements Serializable {

  @Parameter
  public int dispatcherThreadCount;

  @Parameters(name = "dispatcherThreadCount={0}")
  public static Iterable<Integer> dispatcherThreadCounts() {
    return Arrays.asList(1, 3);
  }

  @Rule
  public DistributedRule distributedRule = new DistributedRule();

  @Rule
  public CacheRule cacheRule = new CacheRule();

  @Rule
  public DistributedRestoreSystemProperties restoreSystemProperties =
      new DistributedRestoreSystemProperties();

  @Rule
  public SerializableTemporaryFolder temporaryFolder = new SerializableTemporaryFolder();

  private String partitionedRegionName;
  private String childPartitionedRegionName;
  private String replicateRegionName;
  private String asyncEventQueueId;

  private VM vm0;
  private VM vm1;
  private VM vm2;
  private VM vm3;

  @Before
  public void setUp() throws Exception {
    vm0 = getVM(0);
    vm1 = getVM(1);
    vm2 = getVM(2);
    vm3 = getVM(3);

    String className = getClass().getSimpleName();
    partitionedRegionName = className + "_PR";
    childPartitionedRegionName = className + "_PR_child";
    replicateRegionName = className + "_RR";

    asyncEventQueueId = className;
  }

  /**
   * Override as needed to add to the configuration, such as off-heap-memory-size.
   */
  protected Properties getDistributedSystemProperties() {
    return new Properties();
  }

  /**
   * Override as needed to add to the configuration, such as regionFactory.setOffHeap(boolean).
   */
  protected RegionFactory<?, ?> configureRegion(RegionFactory<?, ?> regionFactory) {
    return regionFactory;
  }

  @Test // parallel, PartitionedRegion, FixedPartitionAttributes
  public void testParallelAsyncEventQueueWithFixedPartition() {
    vm0.invoke(this::createCache);
    vm1.invoke(this::createCache);

    vm0.invoke(() -> createAsyncEventQueue(asyncEventQueueId, new SpyAsyncEventListener(), false,
        100, dispatcherThreadCount, 100, true));
    vm1.invoke(() -> createAsyncEventQueue(asyncEventQueueId, new SpyAsyncEventListener(), false,
        100, dispatcherThreadCount, 100, true));

    String partition1 = "partition1";
    String partition2 = "partition2";
    List<String> allPartitions = Arrays.asList(partition1, partition2);

    vm0.invoke(() -> createFixedPartitionedRegion(partitionedRegionName, asyncEventQueueId, 0, 16,
        partition1, new SimpleFixedPartitionResolver(allPartitions)));
    vm1.invoke(() -> createFixedPartitionedRegion(partitionedRegionName, asyncEventQueueId, 0, 16,
        partition2, new SimpleFixedPartitionResolver(allPartitions)));

    vm0.invoke(() -> doPuts(partitionedRegionName, 256));

    vm0.invoke(this::waitForAsyncQueueToEmpty);
    vm1.invoke(this::waitForAsyncQueueToEmpty);

    int sizeInVM0 = vm0.invoke(() -> getSpyAsyncEventListener().getEventsMap().size());
    int sizeInVM1 = vm1.invoke(() -> getSpyAsyncEventListener().getEventsMap().size());

    String description = "sizeInVM0=" + sizeInVM0 + ", sizeInVM1=" + sizeInVM1;
    assertThat(sizeInVM0 + sizeInVM1).as(description).isEqualTo(256);
  }

  @Test // parallel, conflation, ReplicateRegion, IntegrationTest
  public void testParallelAsyncEventQueueWithReplicatedRegion() {
    vm0.invoke(() -> {
      createCache();

      createAsyncEventQueue(asyncEventQueueId, new SpyAsyncEventListener(), true, 100,
          dispatcherThreadCount, 100, true);

      assertThatThrownBy(() -> createReplicateRegion(replicateRegionName, asyncEventQueueId))
          .isInstanceOf(AsyncEventQueueConfigurationException.class)
          .hasMessageContaining("can not be used with replicated region");
    });
  }

  @Test // parallel, PartitionedRegion
  public void testParallelAsyncEventQueue() {
    vm0.invoke(this::createCache);
    vm1.invoke(this::createCache);
    vm2.invoke(this::createCache);
    vm3.invoke(this::createCache);

    vm0.invoke(() -> createAsyncEventQueue(asyncEventQueueId, new SpyAsyncEventListener(), false,
        100, dispatcherThreadCount, 100, true));
    vm1.invoke(() -> createAsyncEventQueue(asyncEventQueueId, new SpyAsyncEventListener(), false,
        100, dispatcherThreadCount, 100, true));
    vm2.invoke(() -> createAsyncEventQueue(asyncEventQueueId, new SpyAsyncEventListener(), false,
        100, dispatcherThreadCount, 100, true));
    vm3.invoke(() -> createAsyncEventQueue(asyncEventQueueId, new SpyAsyncEventListener(), false,
        100, dispatcherThreadCount, 100, true));

    vm0.invoke(() -> createPartitionedRegion(partitionedRegionName, asyncEventQueueId, 0, 16));
    vm1.invoke(() -> createPartitionedRegion(partitionedRegionName, asyncEventQueueId, 0, 16));
    vm2.invoke(() -> createPartitionedRegion(partitionedRegionName, asyncEventQueueId, 0, 16));
    vm3.invoke(() -> createPartitionedRegion(partitionedRegionName, asyncEventQueueId, 0, 16));

    vm0.invoke(() -> doPuts(partitionedRegionName, 256));

    vm0.invoke(this::waitForAsyncQueueToEmpty);
    vm1.invoke(this::waitForAsyncQueueToEmpty);
    vm2.invoke(this::waitForAsyncQueueToEmpty);
    vm3.invoke(this::waitForAsyncQueueToEmpty);

    int sizeInVM0 = vm0.invoke(() -> getSpyAsyncEventListener().getEventsMap().size());
    int sizeInVM1 = vm1.invoke(() -> getSpyAsyncEventListener().getEventsMap().size());
    int sizeInVM2 = vm2.invoke(() -> getSpyAsyncEventListener().getEventsMap().size());
    int sizeInVM3 = vm3.invoke(() -> getSpyAsyncEventListener().getEventsMap().size());

    String description = "sizeInVM0=" + sizeInVM0 + ", sizeInVM1=" + sizeInVM1 + ", sizeInVM2="
        + sizeInVM2 + ", sizeInVM3=" + sizeInVM3;
    assertThat(sizeInVM0 + sizeInVM1 + sizeInVM2 + sizeInVM3).as(description).isEqualTo(256);
  }

  @Test // parallel, PartitionedRegion
  public void testParallelAsyncEventQueueSize() {
    vm0.invoke(this::createCache);
    vm1.invoke(this::createCache);
    vm2.invoke(this::createCache);
    vm3.invoke(this::createCache);

    vm0.invoke(() -> createAsyncEventQueue(asyncEventQueueId, new SpyAsyncEventListener(), false,
        100, dispatcherThreadCount, 100, true));
    vm1.invoke(() -> createAsyncEventQueue(asyncEventQueueId, new SpyAsyncEventListener(), false,
        100, dispatcherThreadCount, 100, true));
    vm2.invoke(() -> createAsyncEventQueue(asyncEventQueueId, new SpyAsyncEventListener(), false,
        100, dispatcherThreadCount, 100, true));
    vm3.invoke(() -> createAsyncEventQueue(asyncEventQueueId, new SpyAsyncEventListener(), false,
        100, dispatcherThreadCount, 100, true));

    vm0.invoke(() -> createPartitionedRegion(partitionedRegionName, asyncEventQueueId, 0, 16));
    vm1.invoke(() -> createPartitionedRegion(partitionedRegionName, asyncEventQueueId, 0, 16));
    vm2.invoke(() -> createPartitionedRegion(partitionedRegionName, asyncEventQueueId, 0, 16));
    vm3.invoke(() -> createPartitionedRegion(partitionedRegionName, asyncEventQueueId, 0, 16));

    vm0.invoke(() -> getInternalGatewaySender().pause());
    vm1.invoke(() -> getInternalGatewaySender().pause());
    vm2.invoke(() -> getInternalGatewaySender().pause());
    vm3.invoke(() -> getInternalGatewaySender().pause());

    vm0.invoke(this::waitForDispatcherToPause);
    vm1.invoke(this::waitForDispatcherToPause);
    vm2.invoke(this::waitForDispatcherToPause);
    vm3.invoke(this::waitForDispatcherToPause);

    vm0.invoke(() -> doPuts(partitionedRegionName, 1000));

    assertThat(vm0.invoke(() -> getAsyncEventQueue().size())).isEqualTo(1000);
    assertThat(vm1.invoke(() -> getAsyncEventQueue().size())).isEqualTo(1000);
  }

  @Test // parallel, conflation, PartitionedRegion
  public void testParallelAsyncEventQueueWithConflationEnabled() {
    vm0.invoke(this::createCache);
    vm1.invoke(this::createCache);
    vm2.invoke(this::createCache);
    vm3.invoke(this::createCache);

    vm0.invoke(
        () -> createAsyncEventQueue(asyncEventQueueId, new SpyAsyncEventListener(), true, 100,
            dispatcherThreadCount, 100, true));
    vm1.invoke(
        () -> createAsyncEventQueue(asyncEventQueueId, new SpyAsyncEventListener(), true, 100,
            dispatcherThreadCount, 100, true));
    vm2.invoke(
        () -> createAsyncEventQueue(asyncEventQueueId, new SpyAsyncEventListener(), true, 100,
            dispatcherThreadCount, 100, true));
    vm3.invoke(
        () -> createAsyncEventQueue(asyncEventQueueId, new SpyAsyncEventListener(), true, 100,
            dispatcherThreadCount, 100, true));

    vm0.invoke(() -> createPartitionedRegion(partitionedRegionName, asyncEventQueueId, 0, 16));
    vm1.invoke(() -> createPartitionedRegion(partitionedRegionName, asyncEventQueueId, 0, 16));
    vm2.invoke(() -> createPartitionedRegion(partitionedRegionName, asyncEventQueueId, 0, 16));
    vm3.invoke(() -> createPartitionedRegion(partitionedRegionName, asyncEventQueueId, 0, 16));

    vm0.invoke(() -> getInternalGatewaySender().pause());
    vm1.invoke(() -> getInternalGatewaySender().pause());
    vm2.invoke(() -> getInternalGatewaySender().pause());
    vm3.invoke(() -> getInternalGatewaySender().pause());

    vm0.invoke(this::waitForDispatcherToPause);
    vm1.invoke(this::waitForDispatcherToPause);
    vm2.invoke(this::waitForDispatcherToPause);
    vm3.invoke(this::waitForDispatcherToPause);

    Map<Integer, Integer> keyValues = new HashMap<>();
    for (int i = 0; i < 1000; i++) {
      keyValues.put(i, i);
    }

    vm0.invoke(() -> {
      putGivenKeyValue(partitionedRegionName, keyValues);

      validateParallelAsyncEventQueueSize(keyValues.size());
    });

    Map<Integer, String> updateKeyValues = new HashMap<>();
    for (int i = 0; i < 500; i++) {
      updateKeyValues.put(i, i + "_updated");
    }

    vm0.invoke(() -> {
      putGivenKeyValue(partitionedRegionName, updateKeyValues);

      // no conflation of creates
      waitForParallelAsyncEventQueueSize(keyValues.size() + updateKeyValues.size());

      putGivenKeyValue(partitionedRegionName, updateKeyValues);

      // conflation of updates
      waitForParallelAsyncEventQueueSize(keyValues.size() + updateKeyValues.size());
    });

    vm0.invoke(() -> getInternalGatewaySender().resume());
    vm1.invoke(() -> getInternalGatewaySender().resume());
    vm2.invoke(() -> getInternalGatewaySender().resume());
    vm3.invoke(() -> getInternalGatewaySender().resume());

    vm0.invoke(this::waitForAsyncQueueToEmpty);
    vm1.invoke(this::waitForAsyncQueueToEmpty);
    vm2.invoke(this::waitForAsyncQueueToEmpty);
    vm3.invoke(this::waitForAsyncQueueToEmpty);

    int sizeInVM0 = vm0.invoke(() -> getSpyAsyncEventListener().getEventsMap().size());
    int sizeInVM1 = vm1.invoke(() -> getSpyAsyncEventListener().getEventsMap().size());
    int sizeInVM2 = vm2.invoke(() -> getSpyAsyncEventListener().getEventsMap().size());
    int sizeInVM3 = vm3.invoke(() -> getSpyAsyncEventListener().getEventsMap().size());

    String description = "sizeInVM0=" + sizeInVM0 + ", sizeInVM1=" + sizeInVM1 + ", sizeInVM2="
        + sizeInVM2 + ", sizeInVM3=" + sizeInVM3;
    assertThat(sizeInVM0 + sizeInVM1 + sizeInVM2 + sizeInVM3).as(description)
        .isEqualTo(keyValues.size());
  }

  /**
   * Added to reproduce defect #47213
   *
   * TRAC #47213:
   */
  @Test // parallel, conflation, PartitionedRegion, RegressionTest
  public void testParallelAsyncEventQueueWithConflationEnabled_bug47213() {
    vm0.invoke(this::createCache);
    vm1.invoke(this::createCache);
    vm2.invoke(this::createCache);
    vm3.invoke(this::createCache);

    vm0.invoke(
        () -> createAsyncEventQueue(asyncEventQueueId, new SpyAsyncEventListener(), true, 100,
            dispatcherThreadCount, 100, true));
    vm1.invoke(
        () -> createAsyncEventQueue(asyncEventQueueId, new SpyAsyncEventListener(), true, 100,
            dispatcherThreadCount, 100, true));
    vm2.invoke(
        () -> createAsyncEventQueue(asyncEventQueueId, new SpyAsyncEventListener(), true, 100,
            dispatcherThreadCount, 100, true));
    vm3.invoke(
        () -> createAsyncEventQueue(asyncEventQueueId, new SpyAsyncEventListener(), true, 100,
            dispatcherThreadCount, 100, true));

    vm0.invoke(() -> createPartitionedRegionAndAwaitRecovery(partitionedRegionName,
        asyncEventQueueId, 1, 16));
    vm1.invoke(() -> createPartitionedRegionAndAwaitRecovery(partitionedRegionName,
        asyncEventQueueId, 1, 16));
    vm2.invoke(() -> createPartitionedRegionAndAwaitRecovery(partitionedRegionName,
        asyncEventQueueId, 1, 16));
    vm3.invoke(() -> createPartitionedRegionAndAwaitRecovery(partitionedRegionName,
        asyncEventQueueId, 1, 16));

    vm0.invoke(() -> getInternalGatewaySender().pause());
    vm1.invoke(() -> getInternalGatewaySender().pause());
    vm2.invoke(() -> getInternalGatewaySender().pause());
    vm3.invoke(() -> getInternalGatewaySender().pause());

    vm0.invoke(this::waitForDispatcherToPause);
    vm1.invoke(this::waitForDispatcherToPause);
    vm2.invoke(this::waitForDispatcherToPause);
    vm3.invoke(this::waitForDispatcherToPause);

    Map<Integer, Integer> keyValues = new HashMap<>();
    for (int i = 0; i < 1000; i++) {
      keyValues.put(i, i);
    }

    vm0.invoke(() -> {
      putGivenKeyValue(partitionedRegionName, keyValues);

      waitForParallelAsyncEventQueueSize(keyValues.size());
    });

    Map<Integer, String> updateKeyValues = new HashMap<>();
    for (int i = 0; i < 500; i++) {
      updateKeyValues.put(i, i + "_updated");
    }

    vm0.invoke(() -> {
      putGivenKeyValue(partitionedRegionName, updateKeyValues);
      putGivenKeyValue(partitionedRegionName, updateKeyValues);

      waitForParallelAsyncEventQueueSize(keyValues.size() + updateKeyValues.size());
    });

    vm0.invoke(() -> getInternalGatewaySender().resume());
    vm1.invoke(() -> getInternalGatewaySender().resume());
    vm2.invoke(() -> getInternalGatewaySender().resume());
    vm3.invoke(() -> getInternalGatewaySender().resume());

    vm0.invoke(this::waitForAsyncQueueToEmpty);
    vm1.invoke(this::waitForAsyncQueueToEmpty);
    vm2.invoke(this::waitForAsyncQueueToEmpty);
    vm3.invoke(this::waitForAsyncQueueToEmpty);

    int sizeInVM0 = vm0.invoke(() -> getSpyAsyncEventListener().getEventsMap().size());
    int sizeInVM1 = vm1.invoke(() -> getSpyAsyncEventListener().getEventsMap().size());
    int sizeInVM2 = vm2.invoke(() -> getSpyAsyncEventListener().getEventsMap().size());
    int sizeInVM3 = vm3.invoke(() -> getSpyAsyncEventListener().getEventsMap().size());

    String description = "sizeInVM0=" + sizeInVM0 + ", sizeInVM1=" + sizeInVM1 + ", sizeInVM2="
        + sizeInVM2 + ", sizeInVM3=" + sizeInVM3;
    assertThat(sizeInVM0 + sizeInVM1 + sizeInVM2 + sizeInVM3).as(description)
        .isEqualTo(keyValues.size());
  }

  @Test // parallel, PartitionedRegion, accessor
  public void testParallelAsyncEventQueueWithOneAccessor() {
    vm0.invoke(this::createCache);
    vm1.invoke(this::createCache);
    vm2.invoke(this::createCache);
    vm3.invoke(this::createCache);

    vm0.invoke(() -> createAsyncEventQueue(asyncEventQueueId, new SpyAsyncEventListener(), false,
        100, dispatcherThreadCount, 100, true));
    vm1.invoke(() -> createAsyncEventQueue(asyncEventQueueId, new SpyAsyncEventListener(), false,
        100, dispatcherThreadCount, 100, true));
    vm2.invoke(() -> createAsyncEventQueue(asyncEventQueueId, new SpyAsyncEventListener(), false,
        100, dispatcherThreadCount, 100, true));
    vm3.invoke(() -> createAsyncEventQueue(asyncEventQueueId, new SpyAsyncEventListener(), false,
        100, dispatcherThreadCount, 100, true));

    vm0.invoke(
        () -> createPartitionedRegionAccessor(partitionedRegionName, asyncEventQueueId, 0, 16));

    vm1.invoke(() -> createPartitionedRegion(partitionedRegionName, asyncEventQueueId, 0, 16));
    vm2.invoke(() -> createPartitionedRegion(partitionedRegionName, asyncEventQueueId, 0, 16));
    vm3.invoke(() -> createPartitionedRegion(partitionedRegionName, asyncEventQueueId, 0, 16));

    vm0.invoke(() -> doPuts(partitionedRegionName, 256));

    vm1.invoke(this::waitForAsyncQueueToEmpty);
    vm2.invoke(this::waitForAsyncQueueToEmpty);
    vm3.invoke(this::waitForAsyncQueueToEmpty);

    vm0.invoke(() -> validateSpyAsyncEventListenerEventsMap(0));

    int sizeInVM1 = vm1.invoke(() -> getSpyAsyncEventListener().getEventsMap().size());
    int sizeInVM2 = vm2.invoke(() -> getSpyAsyncEventListener().getEventsMap().size());
    int sizeInVM3 = vm3.invoke(() -> getSpyAsyncEventListener().getEventsMap().size());

    String description =
        "sizeInVM1=" + sizeInVM1 + ", sizeInVM2=" + sizeInVM2 + ", sizeInVM3=" + sizeInVM3;
    assertThat(sizeInVM1 + sizeInVM2 + sizeInVM3).as(description).isEqualTo(256);
  }

  @Test // parallel, persistent, PartitionedRegion
  public void testParallelAsyncEventQueueWithPersistence() {
    vm0.invoke(this::createCache);
    vm1.invoke(this::createCache);
    vm2.invoke(this::createCache);
    vm3.invoke(this::createCache);

    vm0.invoke(() -> createPersistentAsyncEventQueue(asyncEventQueueId, new SpyAsyncEventListener(),
        false, 100, createDiskStoreName(asyncEventQueueId), false, dispatcherThreadCount, 100, true,
        true));
    vm1.invoke(() -> createPersistentAsyncEventQueue(asyncEventQueueId, new SpyAsyncEventListener(),
        false, 100, createDiskStoreName(asyncEventQueueId), false, dispatcherThreadCount, 100, true,
        true));
    vm2.invoke(() -> createPersistentAsyncEventQueue(asyncEventQueueId, new SpyAsyncEventListener(),
        false, 100, createDiskStoreName(asyncEventQueueId), false, dispatcherThreadCount, 100, true,
        true));
    vm3.invoke(() -> createPersistentAsyncEventQueue(asyncEventQueueId, new SpyAsyncEventListener(),
        false, 100, createDiskStoreName(asyncEventQueueId), false, dispatcherThreadCount, 100, true,
        true));

    vm0.invoke(() -> createPartitionedRegion(partitionedRegionName, asyncEventQueueId, 0, 16));
    vm1.invoke(() -> createPartitionedRegion(partitionedRegionName, asyncEventQueueId, 0, 16));
    vm2.invoke(() -> createPartitionedRegion(partitionedRegionName, asyncEventQueueId, 0, 16));
    vm3.invoke(() -> createPartitionedRegion(partitionedRegionName, asyncEventQueueId, 0, 16));

    vm0.invoke(() -> doPuts(partitionedRegionName, 256));

    vm0.invoke(this::waitForAsyncQueueToEmpty);
    vm1.invoke(this::waitForAsyncQueueToEmpty);
    vm2.invoke(this::waitForAsyncQueueToEmpty);
    vm3.invoke(this::waitForAsyncQueueToEmpty);

    int sizeInVM0 = vm0.invoke(() -> getSpyAsyncEventListener().getEventsMap().size());
    int sizeInVM1 = vm1.invoke(() -> getSpyAsyncEventListener().getEventsMap().size());
    int sizeInVM2 = vm2.invoke(() -> getSpyAsyncEventListener().getEventsMap().size());
    int sizeInVM3 = vm3.invoke(() -> getSpyAsyncEventListener().getEventsMap().size());

    String description = "sizeInVM0=" + sizeInVM0 + ", sizeInVM1=" + sizeInVM1 + ", sizeInVM2="
        + sizeInVM2 + ", sizeInVM3=" + sizeInVM3;
    assertThat(sizeInVM0 + sizeInVM1 + sizeInVM2 + sizeInVM3).as(description).isEqualTo(256);
  }

  /**
   * Test case to test possibleDuplicates. vm0 & vm1 are hosting the PR. vm1 is killed so the
   * buckets hosted by it are shifted to vm0.
   */
  @Test // parallel, PartitionedRegion, possibleDuplicates
  public void testParallelAsyncEventQueueHA_Scenario1() throws InterruptedException {
    vm0.invoke(this::createCache);
    vm1.invoke(this::createCache);

    vm0.invoke(() -> createAsyncEventQueue(asyncEventQueueId, new GatewaySenderAsyncEventListener(),
        false, 5, dispatcherThreadCount, 100, true));
    vm1.invoke(() -> createAsyncEventQueue(asyncEventQueueId, new GatewaySenderAsyncEventListener(),
        false, 5, dispatcherThreadCount, 100, true));

    vm0.invoke(() -> createPartitionedRegionAndAwaitRecovery(partitionedRegionName,
        asyncEventQueueId, 1, 16));
    vm1.invoke(() -> createPartitionedRegionAndAwaitRecovery(partitionedRegionName,
        asyncEventQueueId, 1, 16));

    vm0.invoke(() -> getInternalGatewaySender().pause());
    vm1.invoke(() -> getInternalGatewaySender().pause());

    vm0.invoke(this::waitForDispatcherToPause);
    vm1.invoke(this::waitForDispatcherToPause);

    vm0.invoke(() -> doPuts(partitionedRegionName, 80));

    Set<Integer> primaryBucketsInVM1 =
        vm0.invoke(() -> getAllLocalPrimaryBucketIds(partitionedRegionName));
    Set<Integer> primaryBucketsInVM2 =
        vm1.invoke(() -> getAllLocalPrimaryBucketIds(partitionedRegionName));

    assertThat(primaryBucketsInVM1.size() + primaryBucketsInVM2.size()).isEqualTo(16);

    vm1.invoke(() -> getCache().close());

    vm0.invoke(() -> {
      // give some time for rebalancing to happen
      SECONDS.sleep(2); // TODO: change to await rebalancing to complete

      getInternalGatewaySender().resume();

      waitForAsyncQueueToEmpty();

      validatePossibleDuplicateEvents(primaryBucketsInVM2, 5);
    });
  }

  /**
   * Test case to test possibleDuplicates. vm0 & vm1 are hosting the PR. vm1 is killed and
   * subsequently vm2 is brought up. Buckets are now rebalanced between vm0 & vm2.
   */
  @Test // parallel, PartitionedRegion, possibleDuplicates
  public void testParallelAsyncEventQueueHA_Scenario2() {
    vm0.invoke(this::createCache);
    vm1.invoke(this::createCache);

    vm0.invoke(() -> createAsyncEventQueue(asyncEventQueueId, new GatewaySenderAsyncEventListener(),
        false, 5, dispatcherThreadCount, 100, true));
    vm1.invoke(() -> createAsyncEventQueue(asyncEventQueueId, new GatewaySenderAsyncEventListener(),
        false, 5, dispatcherThreadCount, 100, true));

    vm0.invoke(() -> createPartitionedRegionAndAwaitRecovery(partitionedRegionName,
        asyncEventQueueId, 1, 16));
    vm1.invoke(() -> createPartitionedRegionAndAwaitRecovery(partitionedRegionName,
        asyncEventQueueId, 1, 16));

    vm0.invoke(() -> getInternalGatewaySender().pause());
    vm1.invoke(() -> getInternalGatewaySender().pause());

    vm0.invoke(() -> doPuts(partitionedRegionName, 80));

    Set<Integer> primaryBucketsInVM1 =
        vm1.invoke(() -> getAllLocalPrimaryBucketIds(partitionedRegionName));
    assertThat(primaryBucketsInVM1).isNotEmpty();

    // before shutdown vm1, both vm0 and vm1 should have 40 events in primary queue
    vm0.invoke(() -> validateAsyncEventQueueStats(40, 40, 80, 80, 0));
    vm1.invoke(() -> validateAsyncEventQueueStats(40, 40, 80, 80, 0));

    vm1.invoke(() -> getCache().close());

    vm2.invoke(() -> {
      createCache();
      createAsyncEventQueue(asyncEventQueueId, new GatewaySenderAsyncEventListener(), false, 5,
          dispatcherThreadCount, 100, true);

      // vm2 will move some primary buckets from vm0, but vm0's primary queue size did not reduce
      getInternalGatewaySender().pause();
      createPartitionedRegionAndAwaitRecovery(partitionedRegionName, asyncEventQueueId, 1, 16);
    });

    Set<Integer> primaryBucketsInVM0 =
        vm0.invoke(() -> getAllLocalPrimaryBucketIds(partitionedRegionName));
    assertThat(primaryBucketsInVM0).isNotEmpty();

    Set<Integer> primaryBucketsInVM2 =
        vm2.invoke(() -> getAllLocalPrimaryBucketIds(partitionedRegionName));
    assertThat(primaryBucketsInVM2).isNotEmpty();

    vm0.invoke(() -> validateAsyncEventQueueStats(40, 40, 80, 80, 0));
    vm2.invoke(() -> validateAsyncEventQueueStats(40, 40, 0, 0, 0));

    vm0.invoke(() -> getInternalGatewaySender().resume());
    vm2.invoke(() -> getInternalGatewaySender().resume());

    vm0.invoke(this::waitForAsyncQueueToEmpty);
    vm2.invoke(this::waitForAsyncQueueToEmpty);

    vm2.invoke(() -> validatePossibleDuplicateEvents(primaryBucketsInVM2, 5));
  }

  /**
   * Test case to test possibleDuplicates. vm0 & vm1 are hosting the PR. vm2 is brought up and
   * rebalancing is triggered so the buckets get balanced among vm0, vm1 & vm2.
   */
  @Test // parallel, PartitionedRegion, possibleDuplicates
  public void testParallelAsyncEventQueueHA_Scenario3() {
    vm0.invoke(this::createCache);
    vm1.invoke(this::createCache);

    vm0.invoke(() -> createAsyncEventQueue(asyncEventQueueId, new GatewaySenderAsyncEventListener(),
        false, 5, dispatcherThreadCount, 100, true));
    vm1.invoke(() -> createAsyncEventQueue(asyncEventQueueId, new GatewaySenderAsyncEventListener(),
        false, 5, dispatcherThreadCount, 100, true));

    vm0.invoke(() -> createPartitionedRegionAndAwaitRecovery(partitionedRegionName,
        asyncEventQueueId, 1, 16));
    vm1.invoke(() -> createPartitionedRegionAndAwaitRecovery(partitionedRegionName,
        asyncEventQueueId, 1, 16));

    vm0.invoke(() -> getInternalGatewaySender().pause());
    vm1.invoke(() -> getInternalGatewaySender().pause());

    vm0.invoke(this::waitForDispatcherToPause);
    vm1.invoke(this::waitForDispatcherToPause);

    vm0.invoke(() -> doPuts(partitionedRegionName, 80));

    vm2.invoke(() -> {
      createCache();
      createAsyncEventQueue(asyncEventQueueId, new GatewaySenderAsyncEventListener(), false, 5,
          dispatcherThreadCount, 100, true);
      createPartitionedRegionAndAwaitRecovery(partitionedRegionName, asyncEventQueueId, 1, 16);
    });

    vm0.invoke(this::doRebalance);

    Set<Integer> primaryBucketsInVM3 =
        vm2.invoke(() -> getAllLocalPrimaryBucketIds(partitionedRegionName));

    vm0.invoke(() -> getInternalGatewaySender().resume());
    vm1.invoke(() -> getInternalGatewaySender().resume());

    vm0.invoke(this::waitForAsyncQueueToEmpty);
    vm1.invoke(this::waitForAsyncQueueToEmpty);
    vm2.invoke(this::waitForAsyncQueueToEmpty);

    vm2.invoke(() -> validatePossibleDuplicateEvents(primaryBucketsInVM3, 5));
  }

  /**
   * Added for defect #50364 Can't colocate region that has AEQ with a region that does not have
   * that same AEQ
   *
   * TRAC #50364:
   */
  @Test // parallel, PartitionedRegion, colocated, RegressionTest, IntegrationTest
  public void testParallelAsyncEventQueueAttachedToChildRegionButNotToParentRegion() {
    vm2.invoke(() -> {
      // create cache on node
      createCache();

      // create AsyncEventQueue on node
      createAsyncEventQueue(asyncEventQueueId, new SpyAsyncEventListener(), false, 10,
          dispatcherThreadCount, 100, true);

      // create leader (parent) PR on node
      createPartitionedRegionWithRecoveryDelay(partitionedRegionName, 0, 0, 100);

      // create colocated (child) PR on node
      createColocatedPartitionedRegion(childPartitionedRegionName, asyncEventQueueId,
          0, 100, partitionedRegionName);

      // do puts in colocated (child) PR on node
      doPuts(childPartitionedRegionName, 1000);

      // wait for AsyncEventQueue to get empty on node
      waitForAsyncQueueToEmpty();

      // verify the events in listener
      assertThat(getSpyAsyncEventListener().getEventsMap().size()).isEqualTo(1000);
    });
  }

  @Test // parallel, Rebalancing, PartitionedRegion
  public void testParallelAsyncEventQueueMoveBucketAndMoveItBackDuringDispatching() {
    vm0.invoke(this::createCache);
    vm1.invoke(this::createCache);

    DistributedMember memberInVM0 =
        vm0.invoke(() -> getCache().getDistributedSystem().getDistributedMember());
    DistributedMember memberInVM1 =
        vm1.invoke(() -> getCache().getDistributedSystem().getDistributedMember());

    vm0.invoke(() -> {
      createAsyncEventQueue(asyncEventQueueId,
