/*
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
 * agreements. See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance with the License. You may obtain a
 * copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */

/**
 * Test various distributed aspects of transactions, e.g. locking/reservation symantics that do not
 * need multiple Region configurations. For those tests see <code>MultiVMRegionTestCase</code>.
 *
 *
 * @since GemFire 4.0
 * @see MultiVMRegionTestCase
 *
 */
package org.apache.geode.cache30;

import static org.apache.geode.distributed.ConfigurationProperties.ENABLE_NETWORK_PARTITION_DETECTION;
import static org.apache.geode.distributed.ConfigurationProperties.LOG_LEVEL;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import java.io.Serializable;
import java.util.HashSet;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;

import org.junit.Ignore;
import org.junit.Test;

import org.apache.geode.SystemFailure;
import org.apache.geode.cache.AttributesFactory;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.CacheLoader;
import org.apache.geode.cache.CacheTransactionManager;
import org.apache.geode.cache.CommitConflictException;
import org.apache.geode.cache.CommitIncompleteException;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.DiskAccessException;
import org.apache.geode.cache.LoaderHelper;
import org.apache.geode.cache.MirrorType;
import org.apache.geode.cache.Operation;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionAttributes;
import org.apache.geode.cache.Scope;
import org.apache.geode.distributed.internal.ResourceEvent;
import org.apache.geode.distributed.internal.ResourceEventsListener;
import org.apache.geode.distributed.internal.locks.DLockBatch;
import org.apache.geode.distributed.internal.locks.DLockService;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.distributed.internal.membership.api.MembershipManagerHelper;
import org.apache.geode.internal.cache.CommitReplyException;
import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.cache.InternalRegionFactory;
import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.internal.cache.RegionEntry;
import org.apache.geode.internal.cache.TXManagerImpl;
import org.apache.geode.internal.cache.TXState;
import org.apache.geode.internal.cache.TXStateInterface;
import org.apache.geode.internal.cache.TXStateProxyImpl;
import org.apache.geode.internal.cache.locks.TXLockBatch;
import org.apache.geode.internal.cache.locks.TXLockService;
import org.apache.geode.internal.cache.locks.TXLockServiceImpl;
import org.apache.geode.test.awaitility.GeodeAwaitility;
import org.apache.geode.test.dunit.Assert;
import org.apache.geode.test.dunit.Host;
import org.apache.geode.test.dunit.IgnoredException;
import org.apache.geode.test.dunit.Invoke;
import org.apache.geode.test.dunit.LogWriterUtils;
import org.apache.geode.test.dunit.SerializableCallable;
import org.apache.geode.test.dunit.SerializableRunnable;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.WaitCriterion;
import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;


public class TXDistributedDUnitTest extends JUnit4CacheTestCase {

  protected <K, V> RegionAttributes<K, V> getRegionAttributes() {
    return getRegionAttributes(Scope.DISTRIBUTED_ACK);
  }

  protected <K, V> RegionAttributes<K, V> getRegionAttributes(Scope scope) {
    AttributesFactory<K, V> factory = new AttributesFactory<>();
    factory.setScope(scope);
    if (scope.isDistributedAck()) {
      factory.setEarlyAck(false);
    }
    return factory.create();
  }

  /**
   * Test a remote grantor
   */
  @Test
  public void testRemoteGrantor() throws Exception {
    IgnoredException.addIgnoredException("killing members ds");
    final CacheTransactionManager txMgr = getCache().getCacheTransactionManager();
    final String rgnName = getUniqueName();
    Region rgn = getCache().createRegion(rgnName, getRegionAttributes());
    rgn.create("key", null);

    Invoke.invokeInEveryVM(new SerializableRunnable("testRemoteGrantor: initial configuration") {
      @Override
      public void run() {
        try {
          Region rgn1 = getCache().createRegion(rgnName, getRegionAttributes());
          rgn1.put("key", "val0");
        } catch (CacheException e) {
          Assert.fail("While creating region", e);
        }
      }
    });

    Host host = Host.getHost(0);
    VM vm0 = host.getVM(0);
    // VM vm1 = host.getVM(1);
    // VM vm2 = host.getVM(2);

    vm0.invoke(new SerializableRunnable("testRemoteGrantor: remote grantor init") {
      @Override
      public void run() {
        try {
          Region rgn1 = getCache().getRegion(rgnName);
          final CacheTransactionManager txMgr2 = getCache().getCacheTransactionManager();
          txMgr2.begin();
          rgn1.put("key", "val1");
          txMgr2.commit();
          assertNotNull(TXLockService.getDTLS());
          assertTrue(TXLockService.getDTLS().isLockGrantor());
        } catch (CacheException e) {
          fail("While performing first transaction");
        }
      }
    });

    // fix for bug 38843 causes the DTLS to be created in every TX participant
    assertNotNull(TXLockService.getDTLS());
    assertFalse(TXLockService.getDTLS().isLockGrantor());
    assertEquals("val1", rgn.getEntry("key").getValue());

    vm0.invoke(new SerializableRunnable("Disconnect from DS, remote grantor death") {
      @Override
      public void run() {
        try {
          MembershipManagerHelper.crashDistributedSystem(getSystem());
        } finally {
          // Allow getCache() to re-establish a ds connection
          closeCache();
        }
      }
    });

    // Make this VM the remote Grantor
    txMgr.begin();
    rgn.put("key", "val2");
    txMgr.commit();
    assertNotNull(TXLockService.getDTLS());
    assertTrue(TXLockService.getDTLS().isLockGrantor());

    SerializableRunnable remoteComm =
        new SerializableRunnable("testRemoteGrantor: remote grantor commit") {
          @Override
          public void run() {
            try {
              Cache c = getCache();
              CacheTransactionManager txMgr2 = c.getCacheTransactionManager();
              Region rgn1 = c.getRegion(rgnName);
              if (rgn1 == null) {
                // This block should only execute on VM0
                rgn1 = c.createRegion(rgnName, getRegionAttributes());
              }

              txMgr2.begin();
              rgn1.put("key", "val3");
              txMgr2.commit();

              if (TXLockService.getDTLS() != null) {
                assertTrue(!TXLockService.getDTLS().isLockGrantor());
              }
            } catch (CacheException e) {
              Assert.fail("While creating region", e);
            }
          }
        };
    Invoke.invokeInEveryVM(remoteComm);
    // vm1.invoke(remoteComm);
    // vm2.invoke(remoteComm);

    assertNotNull(TXLockService.getDTLS());
    assertTrue(TXLockService.getDTLS().isLockGrantor());
    assertEquals("val3", rgn.getEntry("key").getValue());
    rgn.destroyRegion();
  }

  /**
   * Test the internal callbacks used for what else... testing
   */
  @Test
  public void testInternalCallbacks() throws Exception {
    final CacheTransactionManager txMgr = getCache().getCacheTransactionManager();
    final String rgnName1 = getUniqueName() + "_1";
    final String rgnName2 = getUniqueName() + "_2";
    final String rgnName3 = getUniqueName() + "_3";
    Region rgn1 = getCache().createRegion(rgnName1, getRegionAttributes());

    Host host = Host.getHost(0);
    VM vm0 = host.getVM(0);
    VM vm1 = host.getVM(1);

    SerializableRunnable createRgn =
        new SerializableRunnable("testInternalCallbacks: initial configuration") {
          @Override
          public void run() {
            try {
              Region rgn1a = getCache().createRegion(rgnName1, getRegionAttributes());
              Region rgn2 = getCache().createRegion(rgnName2, getRegionAttributes());
              Region rgn3 =
                  getCache().createRegion(rgnName3, getRegionAttributes(Scope.DISTRIBUTED_NO_ACK));
              rgn1a.create("key", null);
              rgn2.create("key", null);
              rgn3.create("key", null);
            } catch (CacheException e) {
              Assert.fail("While creating region", e);
            }
          }
        };
    vm0.invoke(createRgn);
    vm1.invoke(createRgn);

    // Standard commit check
    txMgr.begin();
    rgn1.put("key", "value0");
    txMgr.commit();
    SerializableRunnable checkRgn1 =
        new SerializableRunnable("testInternalCallbacks: check rgn1 valus") {
          @Override
          public void run() {
            Region rgn1a = getCache().getRegion(rgnName1);
            assertNotNull(rgn1a);
            assertEquals("value0", rgn1a.getEntry("key").getValue());
          }
        };
    vm0.invoke(checkRgn1);
    vm1.invoke(checkRgn1);

    {
      final byte[] cbSensors = {0, 0, 0, 0, 0, 0, 0, 0, 0};
      txMgr.begin();
      ((TXStateProxyImpl) ((TXManagerImpl) txMgr).getTXState()).forceLocalBootstrap();
      setInternalCallbacks(((TXManagerImpl) txMgr).getTXState(), cbSensors);
      rgn1.put("key", "value1");
      txMgr.commit();
      for (int i = cbSensors.length - 3; i >= 0; --i) {
        assertEquals("Internal callback " + i + " was not called the expected number of times!",
            (byte) 1, cbSensors[i]);
      }
      for (int i = cbSensors.length - 1; i > cbSensors.length - 3; --i) {
        assertEquals(
            "Internal \"during\" callback " + i + " invoked an unexpected number of times!",
            (byte) 2, cbSensors[i]);
      }
    }
    SerializableRunnable checkRgn1Again =
        new SerializableRunnable("testInternalCallbacks: validate remote values") {
          @Override
          public void run() {
            Region rgn1a = getCache().getRegion(rgnName1);
            assertNotNull(rgn1a);
            assertEquals("value1", rgn1a.getEntry("key").getValue());
          }
        };
    vm0.invoke(checkRgn1Again);
    vm1.invoke(checkRgn1Again);

    // Try 2 regions
    Region rgn2 = getCache().createRegion(rgnName2, getRegionAttributes());
    txMgr.begin();
    rgn1.put("key", "value2");
    rgn2.put("key", "value2");
    txMgr.commit();
    SerializableRunnable checkRgn12 =
        new SerializableRunnable("testInternalCallbacks: check rgn1 valus") {
          @Override
          public void run() {
            Region rgn1a = getCache().getRegion(rgnName1);
            assertNotNull(rgn1a);
            assertEquals("value2", rgn1a.getEntry("key").getValue());
            Region rgn2a = getCache().getRegion(rgnName2);
            assertNotNull(rgn2a);
            assertEquals("value2", rgn2a.getEntry("key").getValue());
          }
        };
    vm0.invoke(checkRgn12);
    vm1.invoke(checkRgn12);

    {
      final byte[] cbSensors = {0, 0, 0, 0, 0, 0, 0, 0, 0};
      txMgr.begin();
      ((TXStateProxyImpl) ((TXManagerImpl) txMgr).getTXState()).forceLocalBootstrap();
      setInternalCallbacks(((TXManagerImpl) txMgr).getTXState(), cbSensors);
      rgn1.put("key", "value3");
      rgn2.put("key", "value3");
      txMgr.commit();

      for (int i = cbSensors.length - 3; i >= 0; i--) {
        assertEquals("Internal callback " + i + " was not called the expected number of times!",
            (byte) 1, cbSensors[i]);
      }
      for (int i = cbSensors.length - 1; i > cbSensors.length - 3; --i) {
        assertEquals(
            "Internal \"during\" callback " + i + " invoked an unexpected number of times!",
            (byte) 2, cbSensors[i]);
      }
    }
    SerializableRunnable checkRgn12Again =
        new SerializableRunnable("testInternalCallbacks: validate both regions remote values") {
          @Override
          public void run() {
            Region rgn1a = getCache().getRegion(rgnName1);
            assertNotNull(rgn1a);
            assertEquals("value3", rgn1a.getEntry("key").getValue());
            Region rgn2a = getCache().getRegion(rgnName2);
            assertNotNull(rgn2a);
            assertEquals("value3", rgn2a.getEntry("key").getValue());
          }
        };
    vm0.invoke(checkRgn12Again);
    vm1.invoke(checkRgn12Again);

    // Try a third region (D_NO_ACK)
    Region rgn3 = getCache().createRegion(rgnName3, getRegionAttributes(Scope.DISTRIBUTED_NO_ACK));
    txMgr.begin();
    rgn1.put("key", "value4");
    rgn2.put("key", "value4");
    rgn3.put("key", "value4");
    txMgr.commit();
    SerializableRunnable checkRgn123 =
        new SerializableRunnable("testInternalCallbacks: check rgn1 valus") {
          @Override
          public void run() {
            Region rgn1a = getCache().getRegion(rgnName1);
            assertNotNull(rgn1a);
            assertEquals("value4", rgn1a.getEntry("key").getValue());
            Region rgn2a = getCache().getRegion(rgnName2);
            assertNotNull(rgn2a);
            assertEquals("value4", rgn2a.getEntry("key").getValue());
            Region rgn3a = getCache().getRegion(rgnName3);
            assertNotNull(rgn3a);
            assertEquals("value4", rgn3a.getEntry("key").getValue());
          }
        };
    vm0.invoke(checkRgn123);
    vm1.invoke(checkRgn123);

    {
      final byte[] cbSensors = {0, 0, 0, 0, 0, 0, 0, 0, 0};
      txMgr.begin();
      ((TXStateProxyImpl) ((TXManagerImpl) txMgr).getTXState()).forceLocalBootstrap();
      setInternalCallbacks(((TXManagerImpl) txMgr).getTXState(), cbSensors);

      rgn1.put("key", "value5");
      rgn2.put("key", "value5");
      rgn3.put("key", "value5");
      txMgr.commit();

      for (int i = cbSensors.length - 3; i >= 0; i--) {
        assertEquals("Internal callback " + i + " was not called the expected number of times!",
            (byte) 1, cbSensors[i]);
      }
      for (int i = cbSensors.length - 1; i > cbSensors.length - 3; --i) {
        assertEquals(
            "Internal \"during\" callback " + i + " invoked an unexpected number of times!",
            (byte) 2, cbSensors[i]);
      }
    }
    SerializableRunnable checkRgn123Again =
        new SerializableRunnable("testInternalCallbacks: validate both regions remote values") {
          @Override
          public void run() {
            Region rgn1a = getCache().getRegion(rgnName1);
            assertNotNull(rgn1a);
            assertEquals("value5", rgn1a.getEntry("key").getValue());
            Region rgn2a = getCache().getRegion(rgnName2);
            assertNotNull(rgn2a);
            assertEquals("value5", rgn2a.getEntry("key").getValue());
            Region rgn3a = getCache().getRegion(rgnName3);
            assertNotNull(rgn3a);
            assertEquals("value5", rgn3a.getEntry("key").getValue());
          }
        };
    vm0.invoke(checkRgn123Again);
    vm1.invoke(checkRgn123Again);

    rgn1.destroyRegion();
    rgn2.destroyRegion();
  }

  static void setInternalCallbacks(TXStateInterface txp, final byte[] cbSensors) {
    ((TXStateProxyImpl) txp).forceLocalBootstrap();
    TXState tx = (TXState) ((TXStateProxyImpl) txp).getRealDeal(null, null);
    assertEquals(9, cbSensors.length);
    tx.setAfterReservation(() -> cbSensors[0]++);
    tx.setAfterConflictCheck(() -> cbSensors[1]++);
    tx.setAfterApplyChanges(() -> cbSensors[2]++);
    tx.setAfterReleaseLocalLocks(() -> cbSensors[3]++);
    tx.setAfterIndividualSend(() -> cbSensors[4]++);
    tx.setAfterIndividualCommitProcess(() -> cbSensors[5]++);
    tx.setAfterSend(() -> cbSensors[6]++);
    tx.setDuringIndividualSend(() -> cbSensors[7]++);
    tx.setDuringIndividualCommitProcess(() -> cbSensors[8]++);
  }

  /**
   * Test distributed ack transactions that consist only of data from loaded values
   */
  @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());
          }
        });

  }

  @Override
  public Properties getDistributedSystemProperties() {
    Properties p = super.getDistributedSystemProperties();
    p.put(LOG_LEVEL, LogWriterUtils.getDUnitLogLevel());
    p.put(ENABLE_NETWORK_PARTITION_DETECTION, "false");
    return p;
  }

  @Test
  public void testHighAvailabilityFeatures() throws Exception {
    IgnoredException.addIgnoredException("DistributedSystemDisconnectedException");
    final String rgnName = getUniqueName();
    AttributesFactory factory = new AttributesFactory();
    factory.setScope(Scope.DISTRIBUTED_ACK);
    factory.setEarlyAck(false);
    Region rgn = getCache().createRegion(rgnName, factory.create());
    Invoke.invokeInEveryVM(
        new SerializableRunnable("testHighAvailabilityFeatures: intial region configuration") {
          @Override
          public void run() {
            try {
              AttributesFactory factory2 = new AttributesFactory();
              factory2.setScope(Scope.DISTRIBUTED_ACK);
              factory2.setEarlyAck(false);
              factory2.setDataPolicy(DataPolicy.REPLICATE);
              getCache().createRegion(rgnName, factory2.create());
            } catch (CacheException e) {
              Assert.fail("While creating region", e);
            }
          }
        });

    // create entries
    rgn.put("key0", "val0_0");
    rgn.put("key1", "val1_0");

    Host host = Host.getHost(0);
    // This test assumes that there are at least three VMs; the origin and two recipients
    assertTrue(host.getVMCount() >= 3);
    final VM originVM = host.getVM(0);

    // Test that there is no commit after a partial commit message
    // send (only sent to a minority of the recipients)
    originVM.invoke(new SerializableRunnable("Flakey DuringIndividualSend Transaction") {
      @Override
      public void run() {
        final Region rgn1 = getCache().getRegion(rgnName);
        assertNotNull(rgn1);
        try {
          final CacheTransactionManager txMgr2 = getCache().getCacheTransactionManager();
          final CacheTransactionManager txMgrImpl = txMgr2;

          txMgr2.begin();
          // 1. setup an internal callback on originVM that will call
          // disconnectFromDS() on the 2nd duringIndividualSend
          // call.
          ((TXStateProxyImpl) ((TXManagerImpl) txMgrImpl).getTXState()).forceLocalBootstrap();
          TXState txState = (TXState) ((TXStateProxyImpl) ((TXManagerImpl) txMgrImpl).getTXState())
              .getRealDeal(null, null);
          txState.setDuringIndividualSend(new Runnable() {
            private int numCalled = 0;

            @Override
            public synchronized void run() {
              ++numCalled;
              rgn1.getCache().getLogger()
                  .info("setDuringIndividualSend Runnable called " + numCalled + " times");
              if (numCalled > 1) {
                MembershipManagerHelper.crashDistributedSystem(getSystem());
              }
            }
          });
          rgn1.put("key0", "val0_1");
          rgn1.put("key1", "val1_1");
          // 2. commit a transaction in originVM, it will disconnect from the DS
          txMgr2.commit();
        } catch (VirtualMachineError e) {
          SystemFailure.initiateFailure(e);
          throw e;
        } catch (Throwable e) {
          rgn1.getCache().getLogger().warning("Ignoring Exception", e);
        } finally {
          // Allow this VM to re-connect to the DS upon getCache() call
          closeCache();
        }
      }
    });
    // 3. verify on all VMs that the transaction was not committed
    final SerializableRunnable noChangeValidator =
        new SerializableRunnable("testHighAvailabilityFeatures: validate no change in Region") {
          @Override
          public void run() {
            Region rgn1 = getCache().getRegion(rgnName);
            if (rgn1 == null) {
              // Expect a null region from originVM
              try {
                AttributesFactory factory2 = new AttributesFactory();
                factory2.setScope(Scope.DISTRIBUTED_ACK);
                factory2.setEarlyAck(false);
                factory2.setDataPolicy(DataPolicy.REPLICATE);
                rgn1 = getCache().createRegion(rgnName, factory2.create());
              } catch (CacheException e) {
                Assert.fail("While creating region", e);
              }
            }
            Region.Entry re = rgn1.getEntry("key0");
            assertNotNull(re);
            assertEquals("val0_0", re.getValue());
            re = rgn1.getEntry("key1");
            assertNotNull(re);
            assertEquals("val1_0", re.getValue());
          }
        };
    Invoke.invokeInEveryVM(noChangeValidator);

    // Test that there is no commit after sending to all recipients
    // but prior to sending the "commit process" message
    originVM.invoke(new SerializableRunnable("Flakey AfterIndividualSend Transaction") {
      @Override
      public void run() {
        final Region rgn1 = getCache().getRegion(rgnName);
        assertNotNull(rgn1);
        try {
          final CacheTransactionManager txMgr2 = getCache().getCacheTransactionManager();
          final CacheTransactionManager txMgrImpl = txMgr2;

          txMgr2.begin();
          // 1. setup an internal callback on originVM that will call
          // disconnectFromDS() on AfterIndividualSend
          ((TXStateProxyImpl) ((TXManagerImpl) txMgrImpl).getTXState()).forceLocalBootstrap();
          TXState txState = (TXState) ((TXStateProxyImpl) ((TXManagerImpl) txMgrImpl).getTXState())
              .getRealDeal(null, null);
          txState.setAfterIndividualSend(new Runnable() {
            @Override
            public synchronized void run() {
              MembershipManagerHelper.crashDistributedSystem(getSystem());
            }
          });
          rgn1.put("key0", "val0_2");
          rgn1.put("key1", "val1_2");
          // 2. commit a transaction in originVM, it will disconnect from the DS
          txMgr2.commit();
        } catch (VirtualMachineError e) {
          SystemFailure.initiateFailure(e);
          throw e;
        } catch (Throwable e) {
          rgn1.getCache().getLogger().warning("Ignoring Exception", e);
        } finally {
          // Allow this VM to re-connect to the DS upon getCache() call
          closeCache();
        }
      }
    });
    // 3. verify on all VMs, including the origin, that the transaction was not committed
    Invoke.invokeInEveryVM(noChangeValidator);

    // Test commit success upon a single commit process message received.
    originVM.invoke(new SerializableRunnable("Flakey DuringIndividualCommitProcess Transaction") {
      @Override
      public void run() {
        final Region rgn1 = getCache().getRegion(rgnName);
        assertNotNull(rgn1);
        try {
          final CacheTransactionManager txMgr2 = getCache().getCacheTransactionManager();
          final CacheTransactionManager txMgrImpl = txMgr2;

          txMgr2.begin();

          ((TXStateProxyImpl) ((TXManagerImpl) txMgrImpl).getTXState()).forceLocalBootstrap();
          TXState txState = (TXState) ((TXStateProxyImpl) ((TXManagerImpl) txMgrImpl).getTXState())
              .getRealDeal(null, null);
          // 1. setup an internal callback on originVM that will call
          // disconnectFromDS() on the 2nd internalDuringIndividualCommitProcess
          // call.
          txState.setDuringIndividualCommitProcess(new Runnable() {
            private int numCalled = 0;

            @Override
            public synchronized void run() {
              ++numCalled;
              rgn1.getCache().getLogger()
                  .info("setDuringIndividualCommitProcess Runnable called " + numCalled + " times");
              if (numCalled > 1) {
                MembershipManagerHelper.crashDistributedSystem(getSystem());
              }
            }
          });
          rgn1.put("key0", "val0_3");
          rgn1.put("key1", "val1_3");
          // 2. commit a transaction in originVM, it will disconnect from the DS
          txMgr2.commit();
        } catch (VirtualMachineError e) {
          SystemFailure.initiateFailure(e);
          throw e;
        } catch (Throwable e) {
          rgn1.getCache().getLogger().warning("Ignoring Exception", e);
        } finally {
          // Allow this VM to re-connect to the DS upon getCache() call
          closeCache();
        }
      }
    });
    // 3. verify on all VMs that the transaction was committed (including the orgin, due to GII)
    SerializableRunnable nonSoloChangeValidator1 = new SerializableRunnable(
        "testHighAvailabilityFeatures: validate v1 non-solo Region changes") {
      @Override
      public void run() {
        Region rgn1 = getCache().getRegion(rgnName);
        if (rgn1 == null) {
          // Expect a null region from originVM
          try {
            AttributesFactory factory2 = new AttributesFactory();
            factory2.setScope(Scope.DISTRIBUTED_ACK);
            factory2.setEarlyAck(false);
            factory2.setDataPolicy(DataPolicy.REPLICATE);
            rgn1 = getCache().createRegion(rgnName, factory2.create());
          } catch (CacheException e) {
            Assert.fail("While creating region", e);
          }
        }
        long giveUp = System.currentTimeMillis() + 10000;
        while (giveUp > System.currentTimeMillis()) {
          try {
            Region.Entry re = rgn1.getEntry("key0");
            assertNotNull(re);
            assertEquals("val0_3", re.getValue());
            re = rgn1.getEntry("key1");
            assertNotNull(re);
            assertEquals("val1_3", re.getValue());
            break;
          } catch (AssertionError e) {
            if (giveUp > System.currentTimeMillis()) {
              throw e;
            }
          }
        }
      }
    };
    Invoke.invokeInEveryVM(nonSoloChangeValidator1);

    // Verify successful solo region commit after duringIndividualSend
    // (same as afterIndividualSend).
    // Create a region that only exists on the origin and another VM
    final String soloRegionName = getUniqueName() + "_solo";
    SerializableRunnable createSoloRegion =
        new SerializableRunnable("testHighAvailabilityFeatures: solo region configuration") {
          @Override
          public void run() {
            try {
              AttributesFactory factory2 = new AttributesFactory();
              factory2.setScope(Scope.DISTRIBUTED_ACK);
              factory2.setEarlyAck(false);
              factory2.setDataPolicy(DataPolicy.REPLICATE);
              Region rgn1 = getCache().createRegion(soloRegionName, factory2.create());
              rgn1.put("soloKey0", "soloVal0_0");
              rgn1.put("soloKey1", "soloVal1_0");
            } catch (CacheException e) {
              Assert.fail("While creating region", e);
            }
          }
        };
    final VM soloRegionVM = host.getVM(1);
    originVM.invoke(createSoloRegion);
    soloRegionVM.invoke(createSoloRegion);
    originVM
        .invoke(new SerializableRunnable("Flakey solo region DuringIndividualSend Transaction") {
          @Override
          public void run() {
            final Region soloRgn = getCache().getRegion(soloRegionName);
            assertNotNull(soloRgn);
            try {
              final CacheTransactionManager txMgr2 = getCache().getCacheTransactionManager();
              final CacheTransactionManager txMgrImpl = txMgr2;

              txMgr2.begin();
              // 1. setup an internal callback on originVM that will call
              // disconnectFromDS() on the 2nd duringIndividualSend
              // call.
              ((TXStateProxyImpl) ((TXManagerImpl) txMgrImpl).getTXState()).forceLocalBootstrap();
              TXState txState =
                  (TXState) ((TXStateProxyImpl) ((TXManagerImpl) txMgrImpl).getTXState())
                      .getRealDeal(null, null);
              txState.setDuringIndividualSend(new Runnable() {
                private int numCalled = 0;

                @Override
                public synchronized void run() {
                  ++numCalled;
                  soloRgn.getCache().getLogger()
                      .info("setDuringIndividualSend Runnable called " + numCalled + " times");
                  if (numCalled > 1) {
                    MembershipManagerHelper.crashDistributedSystem(getSystem());
                  }
                }
              });
              soloRgn.put("soloKey0", "soloVal0_1");
              soloRgn.put("soloKey1", "soloVal1_1");
              // 2. commit a transaction in originVM, it will disconnect from the DS
              txMgr2.commit();
            } catch (VirtualMachineError e) {
              SystemFailure.initiateFailure(e);
              throw e;
            } catch (Throwable e) {
              soloRgn.getCache().getLogger().warning("Ignoring Exception", e);
            } finally {
              // Allow this VM to re-connect to the DS upon getCache() call
              closeCache();
            }
          }
        });
    // 3. verify on the soloRegionVM that the transaction was committed
    final SerializableRunnable soloRegionCommitValidator1 = new SerializableRunnable(
        "testHighAvailabilityFeatures: validate successful v1 commit in solo Region") {
      @Override
      public void run() {
        Region soloRgn = getCache().getRegion(soloRegionName);
        if (soloRgn == null) {
          // Expect a null region from originVM
          try {
            AttributesFactory factory2 = new AttributesFactory();
            factory2.setScope(Scope.DISTRIBUTED_ACK);
            factory2.setEarlyAck(false);
            factory2.setDataPolicy(DataPolicy.REPLICATE);
            soloRgn = getCache().createRegion(soloRegionName, factory2.create());
          } catch (CacheException e) {
            Assert.fail("While creating region ", e);
          }
        }
        Region.Entry re = soloRgn.getEntry("soloKey0");
        assertNotNull(re);
        assertEquals("soloVal0_1", re.getValue());
        re = soloRgn.getEntry("soloKey1");
        assertNotNull(re);
        assertEquals("soloVal1_1", re.getValue());
      }
    };
    originVM.invoke(soloRegionCommitValidator1);
    soloRegionVM.invoke(soloRegionCommitValidator1);
    // verify no change in nonSolo region, re-establish region in originVM
    Invoke.invokeInEveryVM(nonSoloChangeValidator1);

    // Verify no commit for failed send (afterIndividualSend) for solo
    // Region combined with non-solo Region
    originVM.invoke(new SerializableRunnable(
        "Flakey mixed (solo+non-solo) region DuringIndividualSend Transaction") {
      @Override
      public void run() {
        final Region rgn1 = getCache().getRegion(rgnName);
        assertNotNull(rgn1);
        final Region soloRgn = getCache().getRegion(soloRegionName);
        assertNotNull(soloRgn);
        try {
          final CacheTransactionManager txMgr2 = getCache().getCacheTransactionManager();
          final CacheTransactionManager txMgrImpl = txMgr2;

          txMgr2.begin();
          // 1. setup an internal callback on originVM that will call
          // disconnectFromDS() on the afterIndividualSend
          // call.
          ((TXStateProxyImpl) ((TXManagerImpl) txMgrImpl).getTXState()).forceLocalBootstrap();
          TXState txState = (TXState) ((TXStateProxyImpl) ((TXManagerImpl) txMgrImpl).getTXState())
              .getRealDeal(null, null);
          txState.setAfterIndividualSend(new Runnable() {
            @Override
            public synchronized void run() {
              MembershipManagerHelper.crashDistributedSystem(getSystem());
            }
          });

          rgn1.put("key0", "val0_4");
          rgn1.put("key1", "val1_4");
          soloRgn.put("soloKey0", "soloVal0_2");
          soloRgn.put("soloKey1", "soloVal1_2");

          // 2. commit a transaction in originVM, it will disconnect from the DS
          txMgr2.commit();
        } catch (VirtualMachineError e) {
          SystemFailure.initiateFailure(e);
          throw e;
        } catch (Throwable e) {
          rgn1.getCache().getLogger().warning("Ignoring Exception", e);
        } finally {
          // Allow this VM to re-connect to the DS upon getCache() call
          closeCache();
        }
      }
    });
    // Origin and Solo Region VM should be the same as last validation
    originVM.invoke(soloRegionCommitValidator1);
    soloRegionVM.invoke(soloRegionCommitValidator1);
    Invoke.invokeInEveryVM(nonSoloChangeValidator1);

    // Verify commit after sending a single
    // (duringIndividualCommitProcess) commit process for solo Region
    // combined with non-solo Region
    originVM.invoke(new SerializableRunnable(
        "Flakey mixed (solo+non-solo) region DuringIndividualCommitProcess Transaction") {
      @Override
      public void run() {
        final Region rgn1 = getCache().getRegion(rgnName);
        assertNotNull(rgn1);
        final Region soloRgn = getCache().getRegion(soloRegionName);
        assertNotNull(soloRgn);
        try {
          final CacheTransactionManager txMgr2 = getCache().getCacheTransactionManager();
          final CacheTransactionManager txMgrImpl = txMgr2;

          txMgr2.begin();
          // 1. setup an internal callback on originVM that will call
          // disconnectFromDS() on the afterIndividualSend
          // call.
          ((TXStateProxyImpl) ((TXManagerImpl) txMgrImpl).getTXState()).forceLocalBootstrap();
          TXState txState = (TXState) ((TXStateProxyImpl) ((TXManagerImpl) txMgrImpl).getTXState())
              .getRealDeal(null, null);
          txState.setAfterIndividualSend(new Runnable() {
            private int numCalled = 0;

            @Override
            public synchronized void run() {
              ++numCalled;
              rgn1.getCache().getLogger()
                  .info("setDuringIndividualCommitProcess Runnable called " + numCalled + " times");
              if (numCalled > 1) {
                MembershipManagerHelper.crashDistributedSystem(getSystem());
              }
            }
          });

          rgn1.put("key0", "val0_5");
          rgn1.put("key1", "val1_5");
          soloRgn.put("soloKey0", "soloVal0_3");
          soloRgn.put("soloKey1", "soloVal1_3");

          // 2. commit a transaction in originVM, it will disconnect from the DS
          txMgr2.commit();
        } catch (VirtualMachineError e) {
          SystemFailure.initiateFailure(e);
          throw e;
        } catch (Throwable e) {
          rgn1.getCache().getLogger().warning("Ignoring Exception", e);
        } finally {
          // Allow this VM to re-connect to the DS upon getCache() call
          closeCache();
        }
      }
    });
    final SerializableRunnable soloRegionCommitValidator2 = new SerializableRunnable(
        "testHighAvailabilityFeatures: validate successful v2 commit in solo Region") {
      @Override
      public void run() {
        Region soloRgn = getCache().getRegion(soloRegionName);
        if (soloRgn == null) {
          // Expect a null region from originVM
          try {
            AttributesFactory factory2 = new AttributesFactory();
            factory2.setScope(Scope.DISTRIBUTED_ACK);
            factory2.setEarlyAck(false);
            factory2.setDataPolicy(DataPolicy.REPLICATE);
            soloRgn = getCache().createRegion(soloRegionName, factory2.create());
          } catch (CacheException e) {
            Assert.fail("While creating region ", e);
          }
        }
        Region.Entry re = soloRgn.getEntry("soloKey0");
        assertNotNull(re);
        assertEquals("soloVal0_3", re.getValue());
        re = soloRgn.getEntry("soloKey1");
        assertNotNull(re);
        assertEquals("soloVal1_3", re.getValue());
      }
    };
    originVM.invoke(soloRegionCommitValidator2);
    soloRegionVM.invoke(soloRegionCommitValidator2);
    SerializableRunnable nonSoloChangeValidator2 = new SerializableRunnable(
        "testHighAvailabilityFeatures: validate v2 non-solo Region changes") {
      @Override
      public void run() {
        Region rgn1 = getCache().getRegion(rgnName);
        if (rgn1 == null) {
          // Expect a null region from originVM
          try {
            AttributesFactory factory2 = new AttributesFactory();
            factory2.setScope(Scope.DISTRIBUTED_ACK);
            factory2.setEarlyAck(false);
            factory2.setDataPolicy(DataPolicy.REPLICATE);
            rgn1 = getCache().createRegion(rgnName, factory2.create());
          } catch (CacheException e) {
            Assert.fail("While creating region", e);
          }
        }
        Region.Entry re = rgn1.getEntry("key0");
        assertNotNull(re);
        assertEquals("val0_5", re.getValue());
        re = rgn1.getEntry("key1");
        assertNotNull(re);
        assertEquals("val1_5", re.getValue());
      }
    };
    Invoke.invokeInEveryVM(nonSoloChangeValidator2);
  }

  /**
   * A class used in testLockBatchParticipantsUpdate to pause a transaction in the afterResrvation
   * and afterSend states.
   */
  public static class PausibleTX implements Runnable {
    public boolean isRunning = false;
    public String rgnName = null;
    public Cache myCache = null;
    public Object key = null;
    public Object value = null;

    public boolean getIsRunning() {
      return isRunning;
    }

    @Override
    public void run() {
      Region rgn = myCache.getRegion(rgnName);
      final CacheTransactionManager txMgr = myCache.getCacheTransactionManager();
      txMgr.begin();
      ((TXStateProxyImpl) ((TXManagerImpl) txMgr).getTXState()).forceLocalBootstrap();
      TXState txState = (TXState) ((TXStateProxyImpl) ((TXManagerImpl) txMgr).getTXState())
          .getRealDeal(null, null);
      txState.setAfterReservation(() -> {
        try {
          synchronized (PausibleTX.class) {
            isRunning = true;
            // Notify the thread that created this, that we are ready
            PausibleTX.class.notifyAll();
            // Wait for the controller to start a GII and let us proceed
            PausibleTX.class.wait();
          }
        } catch (InterruptedException ie) {
          // PausibleTX.this.myCache.getLogger().info("Why was I interrupted? " + ie);
          fail("interrupted");
        }
      });
      txState.setAfterSend(() -> {
        try {
          synchronized (PausibleTX.class) {
            // Notify the controller that we have sent the TX data (and the
            // update)
            PausibleTX.class.notifyAll();
            // Wait until the controller has determined in fact the update
            // took place
            PausibleTX.class.wait();
          }
        } catch (InterruptedException ie) {
          // PausibleTX.this.myCache.getLogger().info("Why was I interrupted? " + ie);
          fail("interrupted");
        }
      });
      try {
        rgn.put(key, value);
        txMgr.commit();
      } catch (CommitConflictException cce) {
        fail(
            "Did not expect commit conflict exception when sending updates to new members in PausibleTX"
                + cce);
        // } catch (CacheException ce) {
        // fail("Did not expect cache exception when sending updates to new members in PausibleTX" +
        // ce);
      }
    }
  }

  /**
   * Returns the GemFire system ID of the VM on which this method is run
   */
  public static Serializable getSystemId() {
    Serializable ret = null;
    if (getSystemStatic() != null) {
      ret = getSystemStatic().getDistributionManager().getId();
    }
    return ret;
  }

  static HashSet preTXSystemIds;

  public static void setPreTXSystemIds(HashSet ids) {
    TXDistributedDUnitTest.preTXSystemIds = ids;
  }

  static HashSet postTXSystemIds;

  public static void setPostTXSystemIds(HashSet ids) {
    TXDistributedDUnitTest.postTXSystemIds = ids;
  }

  static Serializable txHostId;

  public static void setTXHostSystemId(Serializable id) {
    TXDistributedDUnitTest.txHostId = id;
  }

  /**
   * Test update of lock batch participants (needed when new members are discovered between a
   * commit's locking phase and the applicatoin of the Region's data. See bug 32999
   */
  @Test
  public void testLockBatchParticipantsUpdate() throws Exception {
    final String rgnName = getUniqueName();
    Region rgn = getCache().createRegion(rgnName, getRegionAttributes());
    rgn.create("key", null);

    Host host = Host.getHost(0);
    VM vm0 = host.getVM(0);
    VM vm1 = host.getVM(1);
    VM vm2 = host.getVM(2);
    SerializableRunnable initRegions =
        new SerializableRunnable("testLockBatchParticipantsUpdate: initial configuration") {
          @Override
          public void run() {
            try {
              Region rgn1 = getCache().createRegion(rgnName, getRegionAttributes());
              rgn1.create("key", null);
            } catch (CacheException e) {
              Assert.fail("While creating region", e);
            }
          }
        };
    vm0.invoke(initRegions);
    vm1.invoke(initRegions);
    rgn.put("key", "val1");

    // Connect vm2 also since it may have been shutdown when logPerTest
    // is turned on
    vm2.invoke(new SerializableRunnable("connect vm2 if not connected") {
      @Override
      public void run() {
        getCache();
      }
    });

    // Make VM0 the Grantor
    vm0.invoke(new SerializableRunnable("testLockBatchParticipantsUpdate: remote grantor init") {
      @Override
      public void run() {
        try {
          Region rgn1 = getCache().getRegion(rgnName);
          final CacheTransactionManager txMgr2 = getCache().getCacheTransactionManager();
          assertEquals("val1", rgn1.getEntry("key").getValue());
          txMgr2.begin();
          rgn1.put("key", "val2");
          txMgr2.commit();
          assertNotNull(TXLockService.getDTLS());
          assertTrue(TXLockService.getDTLS().isLockGrantor());
        } catch (CacheException e) {
          fail("While performing first transaction");
        }
      }
    });

    // fix for bug 38843 causes the DTLS to be created in every TX participant
    assertNotNull(TXLockService.getDTLS());
    assertFalse(TXLockService.getDTLS().isLockGrantor());
    assertEquals("val2", rgn.getEntry("key").getValue());

    // Build sets of System Ids and set them up on VM0 for future batch member checks
    HashSet txMembers = new HashSet(4);
    txMembers.add(getSystemId());
    txMembers.add(vm0.invoke(TXDistributedDUnitTest::getSystemId));
    vm0.invoke(() -> TXDistributedDUnitTest.setPreTXSystemIds(txMembers));
    txMembers.add(vm2.invoke(TXDistributedDUnitTest::getSystemId));
    vm0.invoke(() -> TXDistributedDUnitTest.setPostTXSystemIds(txMembers));

    // Don't include the tx host in the batch member set(s)
    Serializable vm1HostId = vm1.invoke(TXDistributedDUnitTest::getSystemId);
    vm0.invoke(() -> TXDistributedDUnitTest.setTXHostSystemId(vm1HostId));

    // Create a TX on VM1 (such that it will ask for locks on VM0) that uses the callbacks
    // to pause and give us time to start a GII process on another VM
    vm1.invoke(new SerializableRunnable(
        "testLockBatchParticipantsUpdate: slow tx (one that detects new member)") {
      @Override
      public void run() {
        // fix for bug 38843 causes the DTLS to be created in every TX participant
        assertNotNull(TXLockService.getDTLS());
        assertFalse(TXLockService.getDTLS().isLockGrantor());

        PausibleTX pauseTXRunnable = new PausibleTX();
        pauseTXRunnable.rgnName = rgnName;
        pauseTXRunnable.myCache = getCache();
        pauseTXRunnable.key = "key";
        pauseTXRunnable.value = "val3";
        new Thread(pauseTXRunnable, "PausibleTX Thread").start();
        synchronized (PausibleTX.class) {
          while (!pauseTXRunnable.getIsRunning()) {
            try {
              PausibleTX.class.wait();
            } catch (InterruptedException ie) {
              fail("Did not expect " + ie);
            }
          }
        }
      }
    });

    // Verify that the lock batch exists VM0 and has the size we expect
    vm0.invoke(new SerializableRunnable(
        "testLockBatchParticipantsUpdate: Verify lock batch exists on VM0 with expected size") {
      @Override
      public void run() {
        getCache().getRegion(rgnName);
        TXLockServiceImpl dtls = (TXLockServiceImpl) TXLockService.getDTLS();
        assertNotNull(dtls);
        assertTrue(dtls.isLockGrantor());
        DLockService dLockSvc = dtls.getInternalDistributedLockService();
        assertNotNull(TXDistributedDUnitTest.txHostId);
        DLockBatch[] batches = dLockSvc.getGrantor()
            .getLockBatches((InternalDistributedMember) TXDistributedDUnitTest.txHostId);
        assertEquals(batches.length, 1);
        TXLockBatch txLockBatch = (TXLockBatch) batches[0];
        assertNotNull(txLockBatch);
        assertNotNull(TXDistributedDUnitTest.preTXSystemIds);
        assertTrue(
            "Members in lock batch " + txLockBatch.getParticipants() + " not the same as "
                + TXDistributedDUnitTest.preTXSystemIds,
            txLockBatch.getParticipants().equals(TXDistributedDUnitTest.preTXSystemIds));
      }
    });

    // Start a GII process on VM2
    vm2.invoke(new SerializableRunnable("testLockBatchParticipantsUpdate: start GII") {
      @Override
      public void run() {
        try {
          AttributesFactory factory = new AttributesFactory();
          factory.setScope(Scope.DISTRIBUTED_ACK);
          factory.setEarlyAck(false);
          factory.setDataPolicy(DataPolicy.REPLICATE);
          getCache().createRegion(rgnName, factory.create());
        } catch (CacheException e) {
          Assert.fail("While creating region", e);
        }
      }
    });

    // Notify TX on VM1 so that it can continue
    vm1.invoke(
        new SerializableRunnable("testLockBatchParticipantsUpdate: Notfiy VM1 TX to continue") {
          @Override
          public void run() {
            synchronized (PausibleTX.class) {
              // Notify VM1 that it should proceed to the TX send
              PausibleTX.class.notifyAll();
              // Wait until VM1 has sent the TX
              try {
                PausibleTX.class.wait();
              } catch (InterruptedException ie) {
                fail("Did not expect " + ie);
              }
            }
          }
        });

    // Verify that the batch on VM0 has added VM2 into the set
    vm0.invoke(new SerializableRunnable(
        "testLockBatchParticipantsUpdate: Verify lock batch contains VM2") {
      @Override
      public void run() {
        getCache().getRegion(rgnName);
        TXLockServiceImpl dtls = (TXLockServiceImpl) TXLockService.getDTLS();
        assertNotNull(dtls);
        assertTrue(dtls.isLockGrantor());
        DLockService dLockSvc = dtls.getInternalDistributedLockService();
        assertNotNull(TXDistributedDUnitTest.txHostId);
        DLockBatch[] batches = dLockSvc.getGrantor()
            .getLockBatches((InternalDistributedMember) TXDistributedDUnitTest.txHostId);
        assertEquals(batches.length, 1);
        TXLockBatch txLockBatch = (TXLockBatch) batches[0];
        assertNotNull(txLockBatch);
        assertNotNull(TXDistributedDUnitTest.preTXSystemIds);
        assertTrue(
            "Members in lock batch " + txLockBatch.getParticipants() + " not the same as "
                + TXDistributedDUnitTest.postTXSystemIds,
            txLockBatch.getParticipants().equals(TXDistributedDUnitTest.postTXSystemIds));
      }
    });
    // fix for bug 38843 causes the DTLS to be created in every TX participant
    assertNotNull(TXLockService.getDTLS());
    assertFalse(TXLockService.getDTLS().isLockGrantor());
    assertEquals("val3", rgn.getEntry("key").getValue());


    // Notify TX on VM1 that it can go ahead and complete the TX
    vm1.invoke(
        new SerializableRunnable("testLockBatchParticipantsUpdate: Notfiy VM1 TX to finish") {
          @Override
          public void run() {
            synchronized (PausibleTX.class) {
              // Notify VM1 that it should finish the TX
              PausibleTX.class.notifyAll();
            }
          }
        });


    rgn.destroyRegion();
  }

  /**
   * Hitachi bug 38809: Applying an exception to a remote VM fails due to an IOException on a Region
   * configured for LRU Overflow
   */
  public static final String TROUBLE_KEY = "GoBoom";

  public static class TXTroubleMaker implements LocalRegion.TestCallable {
    // private final Region r;
    @Override
    public void call(LocalRegion r, Operation op, RegionEntry re) {
      if (TROUBLE_KEY.equals(re.getKey())) {
        throw new DiskAccessException(TROUBLE_KEY, r);
      }
    }
  }

  public static class ShutdownListener implements ResourceEventsListener {
    CountDownLatch latch = new CountDownLatch(1);

    @Override
    public void handleEvent(ResourceEvent event, Object resource) {
      if (event.equals(ResourceEvent.CACHE_REMOVE)) {
        try {
          latch.await();
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }
    }

    public void unblockShutdown() {
      latch.countDown();
    }
  }

  @Ignore("TODO: Disabled for #51260")
  @Test
  public void testRemoteCommitFailure() throws Exception {
    try {
      disconnectAllFromDS();
      final String rgnName1 = getUniqueName() + "_1";
      final String rgnName2 = getUniqueName() + "_2";
      final String diskStoreName = getUniqueName() + "_store";
      Host host = Host.getHost(0);
      VM origin = host.getVM(0);
      VM trouble1 = host.getVM(1);
      VM trouble2 = host.getVM(2);
      VM noTrouble = host.getVM(3);
      CacheSerializableRunnable initRegions =
          new CacheSerializableRunnable("Initialize no trouble regions") {
            @Override
            public void run2() {
              getCache().createDiskStoreFactory().setDiskDirs(getDiskDirs()).create(diskStoreName);
              TXManagerImpl.ALLOW_PERSISTENT_TRANSACTIONS = true;
              AttributesFactory af = new AttributesFactory();
              af.setDataPolicy(DataPolicy.PERSISTENT_REPLICATE);
              af.setScope(Scope.DISTRIBUTED_ACK);
              af.setDiskStoreName(diskStoreName);
              getCache().createRegion(rgnName1, af.create());
              getCache().createRegion(rgnName2, af.create());
            }
          };
      origin.invoke(initRegions);
      noTrouble.invoke(initRegions);
      SerializableRunnable initTroulbeRegions =
          new CacheSerializableRunnable("Initialize regions that cause trouble") {
            @Override
            public void run2() {
              getCache().createDiskStoreFactory().setDiskDirs(getDiskDirs())
                  .create(diskStoreName);
              TXManagerImpl.ALLOW_PERSISTENT_TRANSACTIONS = true;
              InternalRegionFactory factory = getCache().createInternalRegionFactory();
              factory.setDataPolicy(DataPolicy.PERSISTENT_REPLICATE);
              factory.setScope(Scope.DISTRIBUTED_ACK);
              factory.setDiskStoreName(diskStoreName);
              factory.setTestCallable(new TXTroubleMaker());
              factory.create(rgnName1);
              factory.create(rgnName2);
              getCache().getInternalDistributedSystem().addResourceListener(new ShutdownListener());
            }
          };
      trouble1.invoke(initTroulbeRegions);
      trouble2.invoke(initTroulbeRegions);

      SerializableRunnable doTransaction =
          new CacheSerializableRunnable("Run failing transaction") {
            @Override
            public void run2() {
              Cache c = getCache();
              Region r1 = c.getRegion(rgnName1);
              assertNotNull(r1);
              Region r2 = c.getRegion(rgnName2);
              assertNotNull(r2);
              CacheTransactionManager txmgr = c.getCacheTransactionManager();
              txmgr.begin();
              r1.put("k1", "k1");
              r1.put("k2", "k2");
              r1.put(TROUBLE_KEY, TROUBLE_KEY);
              r2.put("k1", "k1");
              r2.put("k2", "k2");
              r2.put(TROUBLE_KEY, TROUBLE_KEY);
              try {
                txmgr.commit();
                fail("Expected an tx incomplete exception");
              } catch (CommitIncompleteException yay) {
                String msg = yay.getMessage();
                // getLogWriter().info("failing exception", yay);
                // Each region on a trouble VM should be mentioned (two regions per trouble VM)
                int ind = 0, match = 0;
                while ((ind = msg.indexOf(rgnName1, ind)) >= 0) {
                  ind++;
                  match++;
                }
                assertEquals(2, match);
                ind = match = 0;
                while ((ind = msg.indexOf(rgnName2, ind)) >= 0) {
                  ind++;
                  match++;
                }
                assertEquals(2, match);
                // DiskAccessExcpetions should be mentioned four times
                ind = match = 0;
                while ((ind = msg.indexOf(DiskAccessException.class.getName(), ind)) >= 0) {
                  ind++;
                  match++;
                }
                assertEquals(4, match);
              }
            }
          };
      IgnoredException ee = null;
      try {
        ee = IgnoredException.addIgnoredException(
            DiskAccessException.class.getName() + "|" + CommitIncompleteException.class.getName()
                + "|" + CommitReplyException.class.getName());
        origin.invoke(doTransaction);
      } finally {
        if (ee != null) {
          ee.remove();
        }
      }

      SerializableCallable allowCacheToShutdown = new SerializableCallable() {
        @Override
        public Object call() throws Exception {
          GemFireCacheImpl cache = (GemFireCacheImpl) getCache();
          List<ResourceEventsListener> listeners =
              cache.getInternalDistributedSystem().getResourceListeners();
          for (ResourceEventsListener l : listeners) {
            if (l instanceof ShutdownListener) {
              ShutdownListener shutListener = (ShutdownListener) l;
              shutListener.unblockShutdown();
            }
          }
          return null;
        }
      };
      trouble1.invoke(allowCacheToShutdown);
      trouble2.invoke(allowCacheToShutdown);

      // Assert proper content on failing VMs
      SerializableRunnable assertTroubledContent =
          new CacheSerializableRunnable("Assert partail commit data") {
            @Override
            public void run2() {
              final Cache c = getCache();
              GeodeAwaitility.await().untilAsserted(new WaitCriterion() {
                @Override
                public boolean done() {
                  return c.getRegion(rgnName1) == null;
                }

                @Override
                public String description() {
                  return null;
                }
              });
              Region r2 = c.getRegion(rgnName2);
              assertNull(r2);
            }
          };
      trouble1.invoke(assertTroubledContent);
      trouble2.invoke(assertTroubledContent);

      // Assert proper content on successful VMs
      SerializableRunnable assertSuccessfulContent =
          new CacheSerializableRunnable("Assert complete commit of data on successful VMs") {
            @Override
            public void run2() {
              Cache c = getCache();
              {
                Region r1 = c.getRegion(rgnName1);
                assertNotNull(r1);
                assertEquals("k1", r1.getEntry("k1").getValue());
                assertEquals("k2", r1.getEntry("k2").getValue());
                assertEquals(TROUBLE_KEY, r1.getEntry(TROUBLE_KEY).getValue());
              }
              {
                Region r2 = c.getRegion(rgnName2);
                assertNotNull(r2);
                assertEquals("k1", r2.getEntry("k1").getValue());
                assertEquals("k2", r2.getEntry("k2").getValue());
                assertEquals(TROUBLE_KEY, r2.getEntry(TROUBLE_KEY).getValue());
              }
            }
          };
      noTrouble.invoke(assertSuccessfulContent);

      // Assert no content on originating VM
      SerializableRunnable assertNoContent =
          new CacheSerializableRunnable("Assert data survives on origin VM") {
            @Override
            public void run2() {
              Cache c = getCache();
              {
                Region r1 = c.getRegion(rgnName1);
                assertNotNull(r1);
                assertNotNull(r1.getEntry("k1"));
                assertNotNull(r1.getEntry("k2"));
                assertNotNull(r1.getEntry(TROUBLE_KEY));
              }
              {
                Region r2 = c.getRegion(rgnName2);
                assertNotNull(r2);
                assertNotNull(r2.getEntry("k1"));
                assertNotNull(r2.getEntry("k2"));
                assertNotNull(r2.getEntry(TROUBLE_KEY));
              }
            }
          };
      origin.invoke(assertNoContent);
    } finally {
      Invoke.invokeInEveryVM(new SerializableCallable() {
        @Override
        public Object call() throws Exception {
          TXManagerImpl.ALLOW_PERSISTENT_TRANSACTIONS = false;
          return null;
        }
      });
    }
  }
}