@Test
  public void testDACKLoadedMessage() throws Exception {
    final CacheTransactionManager txMgr = getCache().getCacheTransactionManager();
    final String rgnName = getUniqueName();
    AttributesFactory factory = new AttributesFactory();
    factory.setScope(Scope.DISTRIBUTED_ACK);
    factory.setEarlyAck(false);
    factory.setCacheLoader(new CacheLoader() {
      @Override
      public Object load(LoaderHelper helper) {
        return "val" + helper.getArgument();
      }

      @Override
      public void close() {}
    });
    Region rgn = getCache().createRegion(rgnName, factory.create());

    Invoke.invokeInEveryVM(new SerializableRunnable("testDACKLoadedMessage: intial configuration") {
      @Override
      public void run() {
        try {
          AttributesFactory factory2 = new AttributesFactory();
          factory2.setScope(Scope.DISTRIBUTED_ACK);
          factory2.setEarlyAck(false);
          // factory.setDataPolicy(DataPolicy.REPLICATE);
          factory2.setMirrorType(MirrorType.KEYS);
          getCache().createRegion(rgnName, factory2.create());
        } catch (CacheException e) {
          Assert.fail("While creating region", e);
        }
      }
    });

    // Confirm the standard case
    txMgr.begin();
    rgn.put("key1", "val1");
    txMgr.commit();
    assertEquals("val1", rgn.getEntry("key1").getValue());

    Invoke
        .invokeInEveryVM(new SerializableRunnable("testDACKLoadedMessage: confirm standard case") {
          @Override
          public void run() {
            Region rgn1 = getCache().getRegion(rgnName);
            assertEquals("val1", rgn1.getEntry("key1").getValue());
          }
        });

    // Confirm loaded value case
    txMgr.begin();
    rgn.get("key2", 2);
    txMgr.commit();
    assertEquals("val2", rgn.getEntry("key2").getValue());

    Invoke
        .invokeInEveryVM(new SerializableRunnable("testDACKLoadedMessage: confirm standard case") {
          @Override
          public void run() {
            Region rgn1 = getCache().getRegion(rgnName);
            assertEquals("val2", rgn1.getEntry("key2").getValue());
          }
        });

    // This should use the ack w/ the lockid
    txMgr.begin();
    rgn.put("key3", "val3");
    rgn.get("key4", 4);
    txMgr.commit();

    Invoke
        .invokeInEveryVM(new SerializableRunnable("testDACKLoadedMessage: confirm standard case") {
          @Override
          public void run() {
            Region rgn1 = getCache().getRegion(rgnName);
            assertEquals("val3", rgn1.getEntry("key3").getValue());
            assertEquals("val4", rgn1.getEntry("key4").getValue());
          }
        });

  }
Test distributed ack transactions that consist only of data from loaded values
geode-core/src/distributedTest/java/org/apache/geode/cache30/TXDistributedDUnitTest.java