private void doTestDisconnectCleanup() throws Exception {
    final AttributesFactory factory = new AttributesFactory();
    factory.setScope(DISTRIBUTED_NO_ACK);
    final Region r = createRootRegion("slowrec", factory.create());
    final DistributionManager dm = getSystem().getDistributionManager();
    final DMStats stats = dm.getStats();
    // set others before vm0 connects
    final Set others = dm.getOtherDistributionManagerIds();
    long initialQueuedMsgs = stats.getAsyncQueuedMsgs();
    final long initialQueues = stats.getAsyncQueues();

    // create receiver in vm0 with queuing enabled
    final Properties p = new Properties();
    p.setProperty(ASYNC_DISTRIBUTION_TIMEOUT, "5");
    p.setProperty(ASYNC_QUEUE_TIMEOUT, "86400000"); // max value
    p.setProperty(ASYNC_MAX_QUEUE_SIZE, "1024"); // max value

    getOtherVm().invoke(new CacheSerializableRunnable("Create other vm") {
      @Override
      public void run2() throws CacheException {
        getSystem(p);
        AttributesFactory af = new AttributesFactory();
        af.setScope(DISTRIBUTED_NO_ACK);
        af.setDataPolicy(REPLICATE);

        doTestDisconnectCleanup_Listener = new ControlListener();
        af.setCacheListener(doTestDisconnectCleanup_Listener);
        createRootRegion("slowrec", af.create());
      }
    });

    // put vm0 cache listener into wait
    getLogWriter().info("[testDisconnectCleanup] about to put vm0 into wait");
    int millisToWait = 1000 * 60 * 5; // 5 minutes
    r.put(KEY_WAIT, millisToWait);
    r.put(KEY_DISCONNECT, KEY_DISCONNECT);

    // build up queue size
    getLogWriter().info("[testDisconnectCleanup] building up queue size...");
    final Object key = "key";
    final int socketBufferSize = getSystem().getConfig().getSocketBufferSize();
    final int VALUE_SIZE = socketBufferSize * 3;
    // final int VALUE_SIZE = 1024 * 1024 ; // 1 MB
    final byte[] value = new byte[VALUE_SIZE];

    int count = 0;
    final long abortMillis = currentTimeMillis() + millisToWait;
    while (stats.getAsyncQueuedMsgs() == initialQueuedMsgs) {
      count++;
      r.put(key, value);
      assertFalse(currentTimeMillis() >= abortMillis);
    }

    getLogWriter().info("[testDisconnectCleanup] After " + count + " puts of size "
        + VALUE_SIZE + " slowrec mode kicked in with queue size=" + stats.getAsyncQueueSize());

    while (stats.getAsyncQueuedMsgs() < 10 || stats.getAsyncQueueSize() < VALUE_SIZE * 10) {
      count++;
      r.put(key, value);
      assertFalse(currentTimeMillis() >= abortMillis);
    }
    assertTrue(stats.getAsyncQueuedMsgs() >= 10);

    while (stats.getAsyncQueues() < 1) {
      pause(100);
      assertFalse(currentTimeMillis() >= abortMillis);
    }

    getLogWriter()
        .info("[testDisconnectCleanup] After " + count + " puts of size " + VALUE_SIZE
            + " queue size has reached " + stats.getAsyncQueueSize()
            + " bytes and number of queues is " + stats.getAsyncQueues() + ".");

    assertTrue(stats.getAsyncQueueSize() >= (VALUE_SIZE * 5));
    assertEquals(initialQueues + 1, stats.getAsyncQueues());

    // assert vm0 is still connected
    assertTrue(dm.getOtherDistributionManagerIds().size() > others.size());

    // send notify to vm0
    getLogWriter().info("[testDisconnectCleanup] wake up vm0");
    getOtherVm().invoke(new SerializableRunnable("Wake up other vm") {
      @Override
      public void run() {
        synchronized (doTestDisconnectCleanup_Listener.CONTROL_LOCK) {
          doTestDisconnectCleanup_Listener.CONTROL_LOCK.notifyAll();
        }
      }
    });

    // make sure we lost a connection to vm0
    getLogWriter().info("[testDisconnectCleanup] wait for vm0 to disconnect");
    WaitCriterion ev = new WaitCriterion() {
      @Override
      public boolean done() {
        return dm.getOtherDistributionManagerIds().size() <= others.size();
      }

      @Override
      public String description() {
        return "waiting for disconnect";
      }
    };
    GeodeAwaitility.await().untilAsserted(ev);
    assertEquals(others, dm.getOtherDistributionManagerIds());

    // check free memory... perform wait loop with System.gc
    getLogWriter().info("[testDisconnectCleanup] wait for queue cleanup");
    ev = new WaitCriterion() {
      @Override
      public boolean done() {
        if (stats.getAsyncQueues() <= initialQueues) {
          return true;
        }
        getRuntime().gc();
        return false;
      }

      @Override
      public String description() {
        return "waiting for queue cleanup";
      }
    };
    GeodeAwaitility.await().untilAsserted(ev);
    assertEquals(initialQueues, stats.getAsyncQueues());
  }
Make sure a disconnect causes queue memory to be released.
geode-core/src/distributedTest/java/org/apache/geode/cache30/SlowRecDUnitTest.java