@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());
  }
Override as needed to add to the configuration, such as regionFactory.setOffHeap(boolean).
geode-core/src/distributedTest/java/org/apache/geode/internal/cache/wan/asyncqueue/ParallelAsyncEventListenerDistributedTest.java