private void doTestPartialMessage() throws Exception {
    final AttributesFactory factory = new AttributesFactory();
    factory.setScope(Scope.DISTRIBUTED_NO_ACK);
    factory.setEnableAsyncConflation(true);
    final Region r = createRootRegion("slowrec", factory.create());
    final DistributionManager dm = getSystem().getDistributionManager();
    final DMStats stats = dm.getStats();

    // set others before vm0 connects
    long initialQueuedMsgs = stats.getAsyncQueuedMsgs();

    // create receiver in vm0 with queuing enabled
    final Properties p = new Properties();
    p.setProperty(ASYNC_DISTRIBUTION_TIMEOUT, String.valueOf(1000 * 4)); // 4 sec
    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(Scope.DISTRIBUTED_NO_ACK);
        af.setDataPolicy(DataPolicy.REPLICATE);

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

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

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

    int count = 0;
    while (stats.getAsyncQueuedMsgs() == initialQueuedMsgs) {
      count++;
      r.put(key, value, count);
    }

    final int partialId = count;
    assertEquals(0, stats.getAsyncConflatedMsgs());

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

    Wait.pause(2000);

    // conflate 10 times
    while (stats.getAsyncConflatedMsgs() < 10) {
      count++;
      r.put(key, value, count);
      if (count == partialId + 1) {
        assertEquals(initialQueuedMsgs + 2, stats.getAsyncQueuedMsgs());
        assertEquals(0, stats.getAsyncConflatedMsgs());
      } else if (count == partialId + 2) {
        assertEquals(initialQueuedMsgs + 2, stats.getAsyncQueuedMsgs());
        assertEquals(1, stats.getAsyncConflatedMsgs());
      }
    }

    final int conflateId = count;

    final int[] expectedArgs = {partialId, conflateId};

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

    // wait for queue to be flushed
    LogWriterUtils.getLogWriter().info("[testPartialMessage] wait for vm0");
    getOtherVm().invoke(new SerializableRunnable("Wait for other vm") {
      @Override
      public void run() {
        try {
          synchronized (doTestPartialMessage_Listener.CONTROL_LOCK) {
            boolean done = false;
            while (!done) {
              if (doTestPartialMessage_Listener.callbackArguments.size() > 0) {
                CallbackWrapper last =
                    (CallbackWrapper) doTestPartialMessage_Listener.callbackArguments.getLast();
                Integer lastId = (Integer) last.callbackArgument;
                if (lastId == conflateId) {
                  done = true;
                } else {
                  doTestPartialMessage_Listener.CONTROL_LOCK.wait(millisToWait);
                }
              } else {
                doTestPartialMessage_Listener.CONTROL_LOCK.wait(millisToWait);
              }
            }
          }
        } catch (InterruptedException ignore) {
          fail("interrupted");
        }
      }
    });

    // assert values on both listeners
    LogWriterUtils.getLogWriter().info("[testPartialMessage] assert callback arguments");
    getOtherVm().invoke(new SerializableRunnable("Assert callback arguments") {
      @Override
      public void run() {
        synchronized (doTestPartialMessage_Listener.CONTROL_LOCK) {
          LogWriterUtils.getLogWriter()
              .info("[testPartialMessage] " + "doTestPartialMessage_Listener.callbackArguments="
                  + doTestPartialMessage_Listener.callbackArguments);

          assertEquals(doTestPartialMessage_Listener.callbackArguments.size(),
              doTestPartialMessage_Listener.callbackTypes.size());

          int i = 0;
          Iterator argIter = doTestPartialMessage_Listener.callbackArguments.iterator();
          Iterator typeIter = doTestPartialMessage_Listener.callbackTypes.iterator();

          while (argIter.hasNext()) {
            CallbackWrapper wrapper = (CallbackWrapper) argIter.next();
            Integer arg = (Integer) wrapper.callbackArgument;
            typeIter.next(); // Integer type
            if (arg < partialId) {
              continue;
            }
            assertEquals(new Integer(expectedArgs[i]), arg);
            // assertIndexDetailsEquals(CALLBACK_UPDATE_INTEGER, type);
            i++;
          }
        }
      }
    });

  }
Make sure a disconnect causes queue memory to be released.
<p>
[bruce] This test was disabled when the SlowRecDUnitTest was re-enabled in build.xml in the
splitbrainNov07 branch. It had been disabled since June 2006 due to hangs. Some of the tests,
like this one, still need work because the periodically (some quite often) fail.
geode-core/src/distributedTest/java/org/apache/geode/cache30/SlowRecDUnitTest.java