@Test
  public void testAckConflation() throws Exception {
    final AttributesFactory factory = new AttributesFactory();
    factory.setScope(Scope.DISTRIBUTED_NO_ACK);
    factory.setEnableAsyncConflation(true);
    final Region r = createRootRegion("slowrec", factory.create());
    final Region ar = createAckRegion(false, true);
    ar.create("ackKey", "ackValue");

    final DMStats stats = getSystem().getDistributionManager().getStats();

    // create receiver in vm0 with queuing enabled
    Properties p = new Properties();
    p.setProperty(ASYNC_DISTRIBUTION_TIMEOUT, "2");
    doCreateOtherVm(p, false);

    forceQueuing(r);
    {
      // make sure ack does not hang
      // make sure two ack updates do not conflate but are both queued
      long startQueuedMsgs = stats.getAsyncQueuedMsgs();
      long startConflatedMsgs = stats.getAsyncConflatedMsgs();
      Thread t = new Thread(() -> ar.put("ackKey", "ackValue"));
      t.start();
      Thread t2 = new Thread(() -> ar.put("ackKey", "ackValue"));
      t2.start();
      // give threads a chance to get queued
      try {
        Thread.sleep(100);
      } catch (InterruptedException ignore) {
        fail("interrupted");
      }
      forceQueueFlush();
      ThreadUtils.join(t, 2 * 1000);
      ThreadUtils.join(t2, 2 * 1000);
      long endQueuedMsgs = stats.getAsyncQueuedMsgs();
      long endConflatedMsgs = stats.getAsyncConflatedMsgs();
      assertEquals(startConflatedMsgs, endConflatedMsgs);
      // queue should be flushed by the time we get an ack
      assertEquals(endQueuedMsgs, stats.getAsyncDequeuedMsgs());
      assertEquals(startQueuedMsgs + 2, endQueuedMsgs);
    }
  }
make sure ack does not hang make sure two ack updates do not conflate but are both queued
geode-core/src/distributedTest/java/org/apache/geode/cache30/SlowRecDUnitTest.java