/*
 * 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.
 */
package org.apache.geode.cache30;

import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.geode.cache.Region.SEPARATOR;
import static org.apache.geode.internal.lang.ThrowableUtils.hasCauseMessage;
import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
import static org.apache.geode.test.dunit.Invoke.invokeInEveryVM;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.assertj.core.api.Assertions.fail;
import static org.assertj.core.api.Assumptions.assumeThat;
import static org.hamcrest.Matchers.equalTo;

import java.io.ByteArrayInputStream;
import java.io.DataInput;
import java.io.DataInputStream;
import java.io.DataOutput;
import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.stream.Stream;

import org.awaitility.core.ConditionTimeoutException;
import org.junit.After;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;

import org.apache.geode.DataSerializable;
import org.apache.geode.DataSerializer;
import org.apache.geode.Instantiator;
import org.apache.geode.InvalidDeltaException;
import org.apache.geode.cache.AttributesFactory;
import org.apache.geode.cache.AttributesMutator;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheEvent;
import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.CacheListener;
import org.apache.geode.cache.CacheLoader;
import org.apache.geode.cache.CacheLoaderException;
import org.apache.geode.cache.CacheTransactionManager;
import org.apache.geode.cache.CacheWriter;
import org.apache.geode.cache.CacheWriterException;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.EntryEvent;
import org.apache.geode.cache.EntryNotFoundException;
import org.apache.geode.cache.ExpirationAction;
import org.apache.geode.cache.ExpirationAttributes;
import org.apache.geode.cache.InterestPolicy;
import org.apache.geode.cache.LoaderHelper;
import org.apache.geode.cache.Operation;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionAttributes;
import org.apache.geode.cache.RegionEvent;
import org.apache.geode.cache.RegionFactory;
import org.apache.geode.cache.RegionShortcut;
import org.apache.geode.cache.Scope;
import org.apache.geode.cache.SubscriptionAttributes;
import org.apache.geode.cache.TimeoutException;
import org.apache.geode.cache.TransactionEvent;
import org.apache.geode.cache.TransactionId;
import org.apache.geode.cache.TransactionListener;
import org.apache.geode.cache.partition.PartitionRegionHelper;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.cache.util.CacheListenerAdapter;
import org.apache.geode.cache.util.TxEventTestUtil;
import org.apache.geode.distributed.ConfigurationProperties;
import org.apache.geode.distributed.internal.DMStats;
import org.apache.geode.distributed.internal.DistributionConfig;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.AvailablePortHelper;
import org.apache.geode.internal.HeapDataOutputStream;
import org.apache.geode.internal.InternalDataSerializer;
import org.apache.geode.internal.InternalInstantiator;
import org.apache.geode.internal.cache.DiskStoreImpl;
import org.apache.geode.internal.cache.EntryExpiryTask;
import org.apache.geode.internal.cache.ExpiryTask;
import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.cache.InitialImageOperation;
import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.internal.cache.PartitionedRegion;
import org.apache.geode.internal.cache.RegionEntry;
import org.apache.geode.internal.cache.TXManagerImpl;
import org.apache.geode.internal.cache.TXStateProxy;
import org.apache.geode.internal.cache.Token;
import org.apache.geode.internal.cache.TombstoneService;
import org.apache.geode.internal.cache.persistence.DiskRecoveryStore;
import org.apache.geode.internal.cache.versions.RegionVersionVector;
import org.apache.geode.internal.cache.versions.VersionSource;
import org.apache.geode.internal.cache.versions.VersionTag;
import org.apache.geode.internal.offheap.MemoryAllocatorImpl;
import org.apache.geode.internal.offheap.StoredObject;
import org.apache.geode.internal.serialization.KnownVersion;
import org.apache.geode.test.dunit.AsyncInvocation;
import org.apache.geode.test.dunit.DistributedTestUtils;
import org.apache.geode.test.dunit.SerializableRunnable;
import org.apache.geode.test.dunit.SerializableRunnableIF;
import org.apache.geode.test.dunit.ThreadUtils;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.Wait;
import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;
import org.apache.geode.test.dunit.internal.JUnit4DistributedTestCase;

/**
 * Abstract superclass of {@link Region} tests that involve more than one VM.
 */
public abstract class MultiVMRegionTestCase extends RegionTestCase {
  private static final long POLL_INTERVAL_MILLIS = 50;

  private static final int PUT_RANGE_1_START = 1;
  private static final int PUT_RANGE_1_END = 5;
  private static final int PUT_RANGE_2_START = 6;
  private static final int PUT_RANGE_2_END = 10;

  private static TestCacheListener testCacheListener;
  private static TestCacheLoader testCacheLoader;
  private static TestCacheWriter testCacheWriter;

  private static int afterCreates;

  static LocalRegion CCRegion;
  static int distributedSystemID;
  protected VM vm0;
  protected VM vm1;
  protected VM vm2;
  protected VM vm3;


  private static <K, V> TestCacheListener<K, V> listener() {
    return testCacheListener;
  }

  private static <K, V> void setListener(TestCacheListener<K, V> listener) {
    testCacheListener = listener;
  }

  protected static <K, V> TestCacheLoader<K, V> loader() {
    return testCacheLoader;
  }

  protected static <K, V> void setLoader(TestCacheLoader<K, V> loader) {
    testCacheLoader = loader;
  }

  private static <K, V> TestCacheWriter<K, V> writer() {
    return testCacheWriter;
  }

  private static <K, V> void setWriter(TestCacheWriter<K, V> writer) {
    testCacheWriter = writer;
  }

  private static void cleanup() {
    testCacheListener = null;
    testCacheLoader = null;
    testCacheWriter = null;
    CCRegion = null;
  }

  @Before
  public void setup() {
    vm0 = VM.getVM(0);
    vm1 = VM.getVM(1);
    vm2 = VM.getVM(2);
    vm3 = VM.getVM(3);
  }

  @After
  public void caseTearDown() {
    disconnectAllFromDS();
  }

  @SuppressWarnings("RedundantThrows") // Subclasses may want to throw
  @Override
  protected final void postTearDownRegionTestCase() throws Exception {
    cleanup();
    invokeInEveryVM(MultiVMRegionTestCase::cleanup);
  }

  @Override
  public Properties getDistributedSystemProperties() {
    Properties properties = super.getDistributedSystemProperties();
    properties.put(ConfigurationProperties.SERIALIZABLE_OBJECT_FILTER,
        "org.apache.geode.cache30.MultiVMRegionTestCase$DeltaValue");
    return properties;
  }

  /**
   * This is a for the ConcurrentMap operations. 4 VMs are used to create the region and operations
   * are performed on one of the nodes
   */
  @Test
  public void testConcurrentOperations() {

    // create the VM(0 - 4)
    for (int i = 0; i < 4; i++) {
      VM vm = VM.getVM(i);
      vm.invoke("createRegion", () -> {
        Cache cache1 = getCache();
        RegionAttributes<?, ?> regionAttribs = getRegionAttributes();
        cache1.createRegionFactory(regionAttribs).create("R1");
      });
    }

    concurrentMapTest(SEPARATOR + "R1");
  }

  /**
   * Do putIfAbsent(), replace(Object, Object), replace(Object, Object, Object), remove(Object,
   * Object) operations
   */
  private void concurrentMapTest(final String rName) {

    vm0.invoke("doConcurrentMapOperations", () -> {
      Cache cache = getCache();
      final Region<String, String> pr = cache.getRegion(rName);
      assertThat(pr).describedAs(rName + " not created").isNotNull();

      // test successful putIfAbsent
      for (int i = PUT_RANGE_1_START; i <= PUT_RANGE_1_END; i++) {
        Object putResult = pr.putIfAbsent(Integer.toString(i), Integer.toString(i));
        assertThat(putResult).describedAs("Expected null, but got " + putResult + " for key " + i)
            .isNull();
      }
      int size = pr.size();
      assertThat(size).describedAs("Size doesn't return expected value")
          .isEqualTo(PUT_RANGE_1_END);
      assertThat(pr.isEmpty()).describedAs("isEmpty doesn't return proper state of the region")
          .isFalse();

      // test unsuccessful putIfAbsent
      for (int i = PUT_RANGE_1_START; i <= PUT_RANGE_1_END; i++) {
        Object putResult = pr.putIfAbsent(Integer.toString(i), Integer.toString(i + 1));
        assertThat(putResult).describedAs("for i=" + i).isEqualTo(Integer.toString(i));
        assertThat(pr.get(Integer.toString(i))).describedAs("for i=" + i).isEqualTo(
            Integer.toString(i));
      }
      size = pr.size();
      assertThat(size).describedAs("Size doesn't return expected value")
          .isEqualTo(PUT_RANGE_1_END);
      assertThat(pr.isEmpty()).describedAs("isEmpty doesn't return proper state of the region")
          .isFalse();

      // test successful replace(key, oldValue, newValue)
      for (int i = PUT_RANGE_1_START; i <= PUT_RANGE_1_END; i++) {
        boolean replaceSucceeded =
            pr.replace(Integer.toString(i), Integer.toString(i), "replaced" + i);
        assertThat(replaceSucceeded).describedAs("for i=" + i).isTrue();
        assertThat(pr.get(Integer.toString(i))).describedAs("for i=" + i).isEqualTo(
            ("replaced" + i));
      }
      size = pr.size();
      assertThat(size).describedAs("Size doesn't return expected value").isEqualTo(
          PUT_RANGE_1_END);
      assertThat(pr.isEmpty()).describedAs("isEmpty doesn't return proper state of the region")
          .isFalse();

      // test unsuccessful replace(key, oldValue, newValue)
      for (int i = PUT_RANGE_1_START; i <= PUT_RANGE_2_END; i++) {
        boolean replaceSucceeded = pr.replace(Integer.toString(i), Integer.toString(i), // wrong
            // expected
            // old
            // value
            "not" + i);
        assertThat(replaceSucceeded).describedAs("for i=" + i).isFalse();
        Object expected1 = i <= PUT_RANGE_1_END ? "replaced" + i : null;
        assertThat(pr.get(Integer.toString(i))).describedAs("for i=" + i).isEqualTo(
            expected1);
      }
      size = pr.size();
      assertThat(size).describedAs("Size doesn't return expected value").isEqualTo(
          PUT_RANGE_1_END);
      assertThat(pr.isEmpty()).describedAs("isEmpty doesn't return proper state of the region")
          .isFalse();

      // test successful replace(key, value)
      for (int i = PUT_RANGE_1_START; i <= PUT_RANGE_1_END; i++) {
        Object replaceResult = pr.replace(Integer.toString(i), "twice replaced" + i);
        assertThat(replaceResult).describedAs("for i=" + i).isEqualTo(("replaced" + i));
        assertThat(pr.get(Integer.toString(i))).describedAs("for i=" + i).isEqualTo(
            ("twice replaced" + i));
      }
      size = pr.size();
      assertThat(size).describedAs("Size doesn't return expected value").isEqualTo(
          PUT_RANGE_1_END);
      assertThat(pr.isEmpty()).describedAs("isEmpty doesn't return proper state of the region")
          .isFalse();

      // test unsuccessful replace(key, value)
      for (int i = PUT_RANGE_2_START; i <= PUT_RANGE_2_END; i++) {
        Object replaceResult = pr.replace(Integer.toString(i), "thrice replaced" + i);
        assertThat(replaceResult).describedAs("for i=" + i).isNull();
        assertThat(pr.get(Integer.toString(i))).describedAs("for i=" + i).isNull();
      }
      size = pr.size();
      assertThat(size).describedAs("Size doesn't return expected value").isEqualTo(
          PUT_RANGE_1_END);
      assertThat(pr.isEmpty()).describedAs("isEmpty doesn't return proper state of the region")
          .isFalse();

      // test unsuccessful remove(key, value)
      for (int i = PUT_RANGE_1_START; i <= PUT_RANGE_2_END; i++) {
        boolean removeResult = pr.remove(Integer.toString(i), Integer.toString(-i));
        assertThat(removeResult).describedAs("for i=" + i).isFalse();
        Object expected1 =
            i <= PUT_RANGE_1_END ? "twice replaced" + i : null;
        assertThat(pr.get(Integer.toString(i))).describedAs("for i=" + i).isEqualTo(
            expected1);
      }
      size = pr.size();
      assertThat(size).describedAs("Size doesn't return expected value").isEqualTo(
          PUT_RANGE_1_END);
      assertThat(pr.isEmpty()).describedAs("isEmpty doesn't return proper state of the region")
          .isFalse();

      // test successful remove(key, value)
      for (int i = PUT_RANGE_1_START; i <= PUT_RANGE_1_END; i++) {
        boolean removeResult = pr.remove(Integer.toString(i), "twice replaced" + i);
        assertThat(removeResult).describedAs("for i=" + i).isTrue();
        assertThat(pr.get(Integer.toString(i))).describedAs("for i=" + i).isNull();
      }
      size = pr.size();
      assertThat(size).describedAs("Size doesn't return expected value").isEqualTo(0);
      assertThat(pr.isEmpty()).describedAs("isEmpty doesn't return proper state of the region")
          .isTrue();
    });


    /*
     * destroy the Region.
     */
    vm0.invoke("destroyRegionOp", () -> {
      Cache cache = getCache();
      Region<?, ?> pr = cache.getRegion(rName);
      assertThat(pr).describedAs("Region already destroyed.").isNotNull();
      pr.destroyRegion();
      assertThat(pr.isDestroyed()).describedAs("Region isDestroyed false").isTrue();
      assertThat(cache.getRegion(rName)).describedAs("Region not destroyed.").isNull();
    });
  }

  /**
   * Tests that doing a {@link Region#put put} in a distributed region one VM updates the value in
   * another VM.
   */
  @Test
  public void testDistributedUpdate() {
    boolean condition = getRegionAttributes().getScope().isDistributed();
    assertThat(condition).isTrue();

    final String name = getUniqueName();

    vm0.invoke("Create Region", () -> {
      createRegion(name);
    });
    vm1.invoke("Create Region", () -> {
      createRegion(name);
    });

    final Object key = "KEY";
    final Object oldValue = "OLD_VALUE";
    final Object newValue = "NEW_VALUE";

    vm0.invoke("Put key/value", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      region.put(key, oldValue);
      flushIfNecessary(region);
    });
    vm1.invoke("Put key/value", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      region.put(key, oldValue);
      flushIfNecessary(region);
    });

    vm0.invoke("Update", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      region.put(key, newValue);
      flushIfNecessary(region);
    });

    vm1.invoke("Validate update", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      Region.Entry entry = region.getEntry(key);
      assertThat(entry).isNotNull();
      assertThat(entry.getValue()).isEqualTo(newValue);
    });
  }

  /**
   * Verifies that distributed updates are delivered in order.
   *
   * <p>
   * Note that this test does not make sense for regions that are {@link Scope#DISTRIBUTED_NO_ACK}
   * for which we do not guarantee the ordering of updates for a single producer/single consumer.
   *
   * @throws Exception if the test fails
   */
  @Test
  public void testOrderedUpdates() throws Exception {
    assumeThat(getRegionAttributes().getScope()).isNotEqualTo(Scope.DISTRIBUTED_NO_ACK);

    final String regionName = getUniqueName();
    final String key = "KEY";
    final int lastValue = 10;

    vm0.invoke("Create region", () -> {
      createRegion(regionName);
    });

    vm1.invoke("Create region and region entry", () -> {
      Region<String, Integer> region = createRegion(regionName);
      region.create(key, null);
    });

    vm1.invoke("Set listener", () -> {
      Region<String, Integer> region = getRootRegion().getSubregion(regionName);
      region.setUserAttribute(new LinkedBlockingQueue<>());

      region.getAttributesMutator().addCacheListener(new CacheListenerAdapter<String, Integer>() {
        @Override
        public void afterUpdate(EntryEvent<String, Integer> event) {
          Region<String, Integer> region = event.getRegion();
          @SuppressWarnings("unchecked")
          BlockingQueue<Integer> queue = (BlockingQueue<Integer>) region.getUserAttribute();
          int value = event.getNewValue();
          try {
            queue.put(value);
          } catch (InterruptedException e) {
            throw new RuntimeException(e);
          }
        }
      });
      flushIfNecessary(region);
    });

    AsyncInvocation<Void> verify = vm1.invokeAsync("Verify", () -> {
      Region<String, Integer> region = getRootRegion().getSubregion(regionName);
      @SuppressWarnings("unchecked")
      BlockingQueue<Integer> queue = (BlockingQueue<Integer>) region.getUserAttribute();
      for (int i = 0; i <= lastValue; i++) {
        try {
          int value = queue.take();
          assertThat(value).isEqualTo(i);
        } catch (InterruptedException e) {
          throw new RuntimeException(e);
        }
      }
    });

    AsyncInvocation<Void> populate = vm0.invokeAsync("Populate", () -> {
      Region<String, Integer> region = getRootRegion().getSubregion(regionName);
      for (int i = 0; i <= lastValue; i++) {
        region.put(key, i);
      }
    });

    populate.await();
    verify.await();
  }

  /**
   * Tests that doing a distributed get results in a <code>netSearch</code>.
   */
  @Test
  public void testDistributedGet() {
    assertThat(getRegionAttributes().getScope().isDistributed()).isTrue();

    final String name = getUniqueName();
    final Object key = "KEY";
    final Object value = "VALUE";

    vm0.invoke("Populate region", () -> {
      Region<Object, Object> region = createRegion(name);
      region.put(key, value);
    });

    vm1.invoke("Distributed get", () -> {
      Region<Object, Object> region = createRegion(name);
      assertThat(region.get(key)).isEqualTo(value);
    });
  }

  /**
   * Tests that doing a {@link Region#put put} on a distributed region in one VM does not effect a
   * region in a different VM that does not have that key defined.
   *
   * @throws Exception if the test fails
   */
  @Test
  public void testDistributedPutNoUpdate() throws Exception {
    assertThat(getRegionAttributes().getScope().isDistributed()).isTrue();

    final String name = getUniqueName();

    vm0.invoke("Create Region", () -> {
      createRegion(name);
    });
    vm1.invoke("Create Region", () -> {
      createRegion(name);
    });

    Thread.sleep(250);

    final Object key = "KEY";
    final Object value = "VALUE";

    vm0.invoke("Put key/value", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      region.put(key, value);
    });

    Thread.sleep(250);

    vm1.invoke("Verify no update", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      Region.Entry entry = region.getEntry(key);
      if (getRegionAttributes().getDataPolicy().withReplication()
          || getRegionAttributes().getPartitionAttributes() != null) {
        assertThat(region.get(key)).isEqualTo(value);
      } else {
        assertThat(entry).isNull();
      }
    });
  }

  /**
   * Two VMs create a region. One populates a region entry. The other VM defines that entry. The
   * first VM updates the entry. The second VM should see the updated value.
   */
  @Test
  public void testDefinedEntryUpdated() {
    final String name = getUniqueName();
    final Object key = "KEY";
    final Object oldValue = "OLD_VALUE";
    final Object newValue = "NEW_VALUE";

    vm0.invoke("Create Region", () -> {
      createRegion(name);
    });
    vm1.invoke("Create Region", () -> {
      createRegion(name);
    });

    vm0.invoke("Create and populate", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      region.put(key, oldValue);
    });

    vm1.invoke("Define entry", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      if (!getRegionAttributes().getDataPolicy().withReplication()) {
        region.create(key, null);
      }
    });

    vm0.invoke("Update entry", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      region.put(key, newValue);
    });

    vm1.invoke("Get entry", repeatingIfNecessary(() -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      assertThat(region.get(key)).isEqualTo(newValue);
    }));
  }


  /**
   * Tests that {@linkplain Region#destroy destroying} an entry is propagated to all VMs that define
   * that entry.
   */
  @Test
  public void testDistributedDestroy() {
    assertThat(getRegionAttributes().getScope().isDistributed()).isTrue();

    final String name = getUniqueName();

    SerializableRunnable create = new CacheSerializableRunnable() {
      @Override
      public void run2() throws CacheException {
        Region<Object, Object> region = createRegion(name);
        assertThat(region.isDestroyed()).isFalse();
        Region root = region.getParentRegion();
        assertThat(root.isDestroyed()).isFalse();
      }
    };

    vm0.invoke("Create Region", create);
    vm1.invoke("Create Region", create);
    vm2.invoke("Create Region", create);

    final Object key = "KEY";
    final Object value = "VALUE";

    SerializableRunnable put = new CacheSerializableRunnable() {
      @Override
      public void run2() throws CacheException {
        Region<Object, Object> region = getRootRegion().getSubregion(name);
        region.put(key, value);
        assertThat(region.isDestroyed()).isFalse();
        assertThat(region.getParentRegion().isDestroyed()).isFalse();
        flushIfNecessary(region);
      }
    };

    vm0.invoke("Put key/value", put);
    vm1.invoke("Put key/value", put);
    vm2.invoke("Put key/value", put);

    SerializableRunnable verifyPut = new CacheSerializableRunnable() {
      @Override
      public void run2() throws CacheException {
        Region<?, ?> root = getRootRegion();
        assertThat(root.isDestroyed()).isFalse();
        Region<Object, Object> region = root.getSubregion(name);
        assertThat(region.isDestroyed()).isFalse();
        assertThat(region.getEntry(key).getValue()).isEqualTo(value);
      }
    };

    vm0.invoke("Verify Put", verifyPut);
    vm1.invoke("Verify Put", verifyPut);
    vm2.invoke("Verify Put", verifyPut);

    vm0.invoke("Destroy Entry", () -> {
      Region<?, ?> region = getRootRegion().getSubregion(name);
      region.destroy(key);
      flushIfNecessary(region);
    });

    CacheSerializableRunnable verifyDestroy =
        new CacheSerializableRunnable() {
          @Override
          public void run2() throws CacheException {
            Region<?, ?> root = getRootRegion();
            assertThat(root.isDestroyed()).isFalse();
            Region<Object, Object> region = root.getSubregion(name);
            assertThat(region.isDestroyed()).isFalse();
            assertThat(region.getEntry(key)).isNull();
          }
        };
    vm0.invoke("Verify entry destruction", verifyDestroy);
    vm1.invoke("Verify entry destruction", verifyDestroy);
    vm2.invoke("Verify entry destruction", verifyDestroy);
  }

  /**
   * Tests that {@linkplain Region#destroy destroying} a region is propagated to all VMs that define
   * that region.
   */
  @Test
  public void testDistributedRegionDestroy() {
    assertThat(getRegionAttributes().getScope().isDistributed()).isTrue();

    final String name = getUniqueName();

    invokeInEveryVM("Create Region", () -> {
      createRegion(name);
    });

    vm0.invoke("Destroy Region", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      region.destroyRegion();
      flushIfNecessary(region);
    });

    invokeInEveryVM("Verify region destruction", () -> await()
        .until(() -> getRootRegion().getSubregion(name) == null));
  }

  /**
   * Tests that a {@linkplain Region#localDestroy} does not effect other VMs that define that
   * entry.
   *
   * @throws Exception if the test fails
   */
  @Test
  public void testLocalDestroy() throws Exception {
    assumeThat(supportsLocalDestroyAndLocalInvalidate()).isTrue();

    // test not valid for persistBackup region since they have to be
    // mirrored KEYS_VALUES
    if (getRegionAttributes().getDataPolicy().withPersistence()) {
      return;
    }

    assertThat(getRegionAttributes().getScope().isDistributed()).isTrue();

    final String name = getUniqueName();

    vm0.invoke("Create Region", () -> {
      createRegion(name);
    });
    vm1.invoke("Create Region", () -> {
      createRegion(name);
    });

    Thread.sleep(250);

    final Object key = "KEY";
    final Object value = "VALUE";

    vm0.invoke("Put key/value", () -> {
      Region<Object, Object> region1 = getRootRegion().getSubregion(name);
      region1.put(key, value);
    });
    vm1.invoke("Put key/value", () -> {
      Region<Object, Object> region1 = getRootRegion().getSubregion(name);
      region1.put(key, value);
    });

    Thread.sleep(250);

    vm0.invoke("Local Destroy Entry", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      region.localDestroy(key);
    });

    Thread.sleep(250);

    vm1.invoke("Verify entry existence", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      assertThat(region.getEntry(key)).isNotNull();
    });
  }

  /**
   * Tests that a {@link Region#localDestroyRegion} is not propagated to other VMs that define that
   * region.
   *
   * @throws Exception if the test fails
   */
  @Test
  public void testLocalRegionDestroy() throws Exception {
    assertThat(getRegionAttributes().getScope().isDistributed()).isTrue();

    final String name = getUniqueName();

    vm0.invoke("Create Region", () -> {
      createRegion(name);
    });
    vm1.invoke("Create Region", () -> {
      createRegion(name);
    });

    Thread.sleep(250);

    vm0.invoke("Local Destroy Region",
        () -> getRootRegion().getSubregion(name).localDestroyRegion());

    Thread.sleep(250);

    vm1.invoke("Verify region existence", () -> {
      assertThat(getRootRegion().getSubregion(name)).isNotNull();
    });
  }

  /**
   * Tests that {@linkplain Region#invalidate invalidating} an entry is propagated to all VMs that
   * define that entry.
   */
  @Test
  public void testDistributedInvalidate() {
    assertThat(getRegionAttributes().getScope().isDistributed()).isTrue();

    final String name = getUniqueName();

    vm0.invoke("Create Region", () -> {
      createRegion(name);
    });
    vm1.invoke("Create Region", () -> {
      createRegion(name);
    });

    // vm2 is on a different gemfire system
    vm2.invoke("Create Region", () -> {
      createRegion(name);
    });

    final Object key = "KEY";
    final Object value = "VALUE";

    SerializableRunnable put = new CacheSerializableRunnable() {
      @Override
      public void run2() throws CacheException {
        Region<Object, Object> region = getRootRegion().getSubregion(name);
        region.put(key, value);
        flushIfNecessary(region);
      }
    };

    vm0.invoke("Put key/value", put);
    vm1.invoke("Put key/value", put);
    vm2.invoke("Put key/value", put);

    vm0.invoke("Invalidate Entry", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      region.invalidate(key);
      flushIfNecessary(region);
    });

    CacheSerializableRunnable verify = new CacheSerializableRunnable() {
      @Override
      public void run2() throws CacheException {
        Region<Object, Object> region = getRootRegion().getSubregion(name);
        Region.Entry entry = region.getEntry(key);
        assertThat(entry).isNotNull();
        if (entry.getValue() != null) {
          // changed from severe to fine because it is possible
          // for this to return non-null on d-no-ack
          // that is was invokeRepeatingIfNecessary is called
          logger.debug("invalidated entry has value of " + entry.getValue());
        }
        assertThat(entry.getValue()).isNull();
      }
    };

    vm1.invoke("Verify entry invalidation", verify);
    vm2.invoke("Verify entry invalidation", verify);
  }

  /**
   * Tests that {@linkplain Region#invalidate invalidating} an entry in multiple VMs does not cause
   * any problems.
   */
  @Test
  public void testDistributedInvalidate4() {
    assertThat(getRegionAttributes().getScope().isDistributed()).isTrue();

    final String name = getUniqueName();
    final Object key = "KEY";
    final Object value = "VALUE";

    int vmCount = VM.getVMCount();
    for (int i = 0; i < vmCount; i++) {
      VM vm0 = VM.getVM(i);
      vm0.invoke("Create Region", () -> {
        createRegion(name);
      });
    }

    for (int i = 0; i < vmCount; i++) {
      VM vm = VM.getVM(i);
      vm.invoke("put entry", () -> {
        Region<Object, Object> region = getRootRegion().getSubregion(name);
        region.put(key, value);
        flushIfNecessary(region);
      });
    }

    for (int i = 0; i < vmCount; i++) {
      VM vm = VM.getVM(i);
      vm.invoke("Invalidate Entry", () -> {
        Region<Object, Object> region = getRootRegion().getSubregion(name);
        region.invalidate(key);
        flushIfNecessary(region);
      });
    }

    for (int i = 0; i < vmCount; i++) {
      VM vm0 = VM.getVM(i);
      vm0.invoke("Verify entry invalidation", () -> {
        Region<Object, Object> region = getRootRegion().getSubregion(name);
        Region.Entry entry = region.getEntry(key);
        assertThat(entry).isNotNull();
        assertThat(entry.getValue()).isNull();
      });
    }
  }

  /**
   * Tests that {@linkplain Region#invalidateRegion invalidating} a region is propagated to all VMs
   * that define that entry.
   */
  @Test
  public void testDistributedRegionInvalidate() {
    assumeThat(supportsSubregions()).isTrue();

    final String name = getUniqueName();
    final String subname = "sub";
    final boolean useSubs = getRegionAttributes().getPartitionAttributes() == null;

    invokeInEveryVM("Create Region", () -> {
      Region<Object, Object> region1;
      region1 = createRegion(name);
      if (useSubs) {
        getCache().createRegionFactory(region1.getAttributes()).createSubregion(region1, subname);
      }
    });

    final Object key = "KEY";
    final Object value = "VALUE";
    final Object key2 = "KEY2";
    final Object value2 = "VALUE2";

    invokeInEveryVM("Put key/value", () -> {
      Region<Object, Object> region1 = getRootRegion().getSubregion(name);
      region1.put(key, value);
      region1.put(key2, value2);
      flushIfNecessary(region1);

      if (useSubs) {
        Region<Object, Object> subregion = region1.getSubregion(subname);
        subregion.put(key, value);
        subregion.put(key2, value2);
        flushIfNecessary(subregion);
      }
    });

    vm0.invoke("Invalidate Region", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      region.invalidateRegion();
    });

    invokeInEveryVM(repeatingIfNecessary(() -> {
