/*
 * 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.internal.cache;

import static java.util.Arrays.asList;
import static java.util.Collections.emptySet;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.apache.geode.cache.Region.SEPARATOR;
import static org.apache.geode.cache.RegionShortcut.PARTITION;
import static org.apache.geode.cache.RegionShortcut.PARTITION_PERSISTENT;
import static org.apache.geode.cache.RegionShortcut.REPLICATE;
import static org.apache.geode.cache.client.PoolFactory.DEFAULT_READ_TIMEOUT;
import static org.apache.geode.cache.client.PoolFactory.DEFAULT_SUBSCRIPTION_ACK_INTERVAL;
import static org.apache.geode.cache.client.PoolFactory.DEFAULT_SUBSCRIPTION_ENABLED;
import static org.apache.geode.cache.client.PoolFactory.DEFAULT_SUBSCRIPTION_REDUNDANCY;
import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
import static org.apache.geode.internal.cache.AbstractCacheServer.MAXIMUM_TIME_BETWEEN_PINGS_PROPERTY;
import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
import static org.apache.geode.test.awaitility.GeodeAwaitility.getTimeout;
import static org.apache.geode.test.dunit.IgnoredException.addIgnoredException;
import static org.apache.geode.test.dunit.VM.getController;
import static org.apache.geode.test.dunit.VM.getHostName;
import static org.apache.geode.test.dunit.VM.getVM;
import static org.apache.geode.test.dunit.VM.getVMId;
import static org.apache.geode.test.dunit.rules.DistributedRule.getDistributedSystemProperties;
import static org.apache.geode.test.dunit.rules.DistributedRule.getLocators;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.catchThrowable;
import static org.mockito.Mockito.mock;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.File;
import java.io.IOException;
import java.io.Serializable;
import java.net.ConnectException;
import java.net.SocketException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;

import junitparams.Parameters;
import junitparams.naming.TestCaseName;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.junit.runner.RunWith;

import org.apache.geode.DataSerializable;
import org.apache.geode.DataSerializer;
import org.apache.geode.cache.CacheFactory;
import org.apache.geode.cache.CacheWriterException;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.DiskStore;
import org.apache.geode.cache.EntryEvent;
import org.apache.geode.cache.LowMemoryException;
import org.apache.geode.cache.Operation;
import org.apache.geode.cache.PartitionAttributesFactory;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.Region.Entry;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.RegionFactory;
import org.apache.geode.cache.RegionShortcut;
import org.apache.geode.cache.Scope;
import org.apache.geode.cache.TimeoutException;
import org.apache.geode.cache.client.ClientCacheFactory;
import org.apache.geode.cache.client.ClientRegionFactory;
import org.apache.geode.cache.client.ClientRegionShortcut;
import org.apache.geode.cache.client.PoolFactory;
import org.apache.geode.cache.client.PoolManager;
import org.apache.geode.cache.client.ServerConnectivityException;
import org.apache.geode.cache.client.ServerOperationException;
import org.apache.geode.cache.client.internal.InternalClientCache;
import org.apache.geode.cache.persistence.PartitionOfflineException;
import org.apache.geode.cache.query.CqAttributes;
import org.apache.geode.cache.query.CqAttributesFactory;
import org.apache.geode.cache.query.CqEvent;
import org.apache.geode.cache.query.CqListener;
import org.apache.geode.cache.query.CqQuery;
import org.apache.geode.cache.query.SelectResults;
import org.apache.geode.cache.query.Struct;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.cache.util.CacheListenerAdapter;
import org.apache.geode.cache.util.CacheWriterAdapter;
import org.apache.geode.distributed.DistributedSystemDisconnectedException;
import org.apache.geode.distributed.OplogCancelledException;
import org.apache.geode.internal.cache.persistence.DiskRecoveryStore;
import org.apache.geode.internal.cache.tier.sockets.VersionedObjectList;
import org.apache.geode.internal.cache.versions.VersionTag;
import org.apache.geode.test.dunit.AsyncInvocation;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.rules.DistributedExecutorServiceRule;
import org.apache.geode.test.dunit.rules.DistributedRestoreSystemProperties;
import org.apache.geode.test.dunit.rules.DistributedRule;
import org.apache.geode.test.junit.categories.ClientServerTest;
import org.apache.geode.test.junit.categories.ClientSubscriptionTest;
import org.apache.geode.test.junit.rules.serializable.SerializableTemporaryFolder;
import org.apache.geode.test.junit.runners.GeodeParamsRunner;

/**
 * Tests putAll for c/s. Also tests removeAll
 *
 * @since GemFire 5.0.23
 */
@Category({ClientServerTest.class, ClientSubscriptionTest.class})
@RunWith(GeodeParamsRunner.class)
@SuppressWarnings("serial,NumericCastThatLosesPrecision")
public class PutAllClientServerDistributedTest implements Serializable {

  private static final long TIMEOUT_MILLIS_LONG = getTimeout().toMillis();
  private static final int TIMEOUT_MILLIS_INTEGER = (int) TIMEOUT_MILLIS_LONG;
  private static final String TIMEOUT_MILLIS_STRING = String.valueOf(TIMEOUT_MILLIS_LONG);

  private static final int ONE_HUNDRED = 100;
  private static final int ONE_THOUSAND = 1000;
  private static final InternalCache DUMMY_CACHE = mock(InternalCache.class);
  private static final InternalClientCache DUMMY_CLIENT_CACHE = mock(InternalClientCache.class);
  private static final Counter DUMMY_COUNTER = new Counter("dummy");
  private static final Action<Integer> EMPTY_INTEGER_ACTION =
      new Action<>(null, emptyConsumer());

  private static final AtomicReference<TickerData> VALUE = new AtomicReference<>();
  private static final AtomicReference<CountDownLatch> LATCH = new AtomicReference<>();
  private static final AtomicReference<CountDownLatch> BEFORE = new AtomicReference<>();
  private static final AtomicReference<CountDownLatch> AFTER = new AtomicReference<>();
  private static final AtomicReference<Counter> COUNTER = new AtomicReference<>();

  private static final AtomicReference<InternalCache> CACHE = new AtomicReference<>();
  private static final AtomicReference<InternalClientCache> CLIENT_CACHE = new AtomicReference<>();
  private static final AtomicReference<File> DISK_DIR = new AtomicReference<>();

  private String regionName;
  private String hostName;
  private String poolName;
  private String diskStoreName;
  private String keyPrefix;
  private String locators;

  private VM server1;
  private VM server2;
  private VM client1;
  private VM client2;

  @Rule
  public DistributedRule distributedRule = new DistributedRule();
  @Rule
  public DistributedExecutorServiceRule executorServiceRule = new DistributedExecutorServiceRule();
  @Rule
  public DistributedRestoreSystemProperties restoreProps = new DistributedRestoreSystemProperties();
  @Rule
  public SerializableTemporaryFolder temporaryFolder = new SerializableTemporaryFolder();

  @Before
  public void setUp() {
    regionName = getClass().getSimpleName();
    hostName = getHostName();
    poolName = "testPool";
    diskStoreName = "ds1";
    keyPrefix = "prefix-";
    locators = getLocators();

    server1 = getVM(0);
    server2 = getVM(1);
    client1 = getVM(2);
    client2 = getVM(3);

    for (VM vm : asList(getController(), client1, client2, server1, server2)) {
      vm.invoke(() -> {
        VALUE.set(null);
        COUNTER.set(null);
        LATCH.set(new CountDownLatch(0));
        BEFORE.set(new CountDownLatch(0));
        AFTER.set(new CountDownLatch(0));
        CACHE.set(DUMMY_CACHE);
        CLIENT_CACHE.set(DUMMY_CLIENT_CACHE);
        DISK_DIR.set(temporaryFolder.newFolder("diskDir-" + getVMId()).getAbsoluteFile());
        System.setProperty(MAXIMUM_TIME_BETWEEN_PINGS_PROPERTY, TIMEOUT_MILLIS_STRING);
      });
    }

    addIgnoredException(ConnectException.class);
    addIgnoredException(PutAllPartialResultException.class);
    addIgnoredException(RegionDestroyedException.class);
    addIgnoredException(ServerConnectivityException.class);
    addIgnoredException(SocketException.class);
    addIgnoredException("Broken pipe");
    addIgnoredException("Connection reset");
    addIgnoredException("Unexpected IOException");
  }

  @After
  public void tearDown() {
    for (VM vm : asList(getController(), client1, client2, server1, server2)) {
      vm.invoke(() -> {
        countDown(LATCH);
        countDown(BEFORE);
        countDown(AFTER);
        closeClientCache();
        closeCache();
        PoolManager.close();
        DISK_DIR.set(null);
      });
    }
  }

  /**
   * Tests putAll to one server.
   */
  @Test
  public void testOneServer() throws Exception {
    int serverPort = server1.invoke(() -> new ServerBuilder()
        .regionShortcut(REPLICATE)
        .create());

    client1.invoke(() -> new ClientBuilder()
        .serverPorts(serverPort)
        .subscriptionAckInterval()
        .subscriptionEnabled(true)
        .subscriptionRedundancy()
        .create());
    client2.invoke(() -> new ClientBuilder()
        .serverPorts(serverPort)
        .subscriptionAckInterval()
        .subscriptionEnabled(true)
        .subscriptionRedundancy()
        .create());

    client2.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      region.registerInterest("ALL_KEYS");
    });

    client1.invoke(() -> {
      getClientCache()
          .<String, TickerData>createClientRegionFactory(ClientRegionShortcut.LOCAL)
          .create("localsave");

      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      doPutAll(region, "key-", ONE_HUNDRED);
      assertThat(region.size()).isEqualTo(ONE_HUNDRED);
    });

    client1.invoke(() -> {
      BEFORE.set(new CountDownLatch(1));
      AFTER.set(new CountDownLatch(1));
    });

    AsyncInvocation<Void> createCqInClient1 = client1.invokeAsync(() -> {
      // create a CQ for key 10-20
      Region<String, TickerData> localSaveRegion = getClientCache().getRegion("localsave");

      CqAttributesFactory cqAttributesFactory = new CqAttributesFactory();
      cqAttributesFactory.addCqListener(new CountingCqListener<>(localSaveRegion));
      CqAttributes cqAttributes = cqAttributesFactory.create();

      String cqName = "EOInfoTracker";
      String query = String.join(" ",
          "SELECT ALL * FROM " + SEPARATOR + regionName + " ii",
          "WHERE ii.getTicker() >= '10' and ii.getTicker() < '20'");

      CqQuery cqQuery = getClientCache().getQueryService().newCq(cqName, query, cqAttributes);

      SelectResults<Struct> results = cqQuery.executeWithInitialResults();
      List<Struct> resultsAsList = results.asList();
      for (int i = 0; i < resultsAsList.size(); i++) {
        Struct struct = resultsAsList.get(i);
        TickerData tickerData = (TickerData) struct.get("value");
        localSaveRegion.put("key-" + i, tickerData);
      }

      countDown(BEFORE);
      awaitLatch(AFTER);

      cqQuery.close();
    });

    server1.invoke(() -> {
      Region<String, TickerData> region = getCache().getRegion(regionName);
      assertThat(region.size()).isEqualTo(ONE_HUNDRED);

      for (int i = 0; i < ONE_HUNDRED; i++) {
        TickerData tickerData = region.getEntry("key-" + i).getValue();
        assertThat(tickerData.getPrice()).isEqualTo(i);
      }
    });

    // verify CQ is ready
    client1.invoke(() -> {
      awaitLatch(BEFORE);

      Region<String, TickerData> localSaveRegion = getClientCache().getRegion("localsave");
      await().untilAsserted(() -> assertThat(localSaveRegion.size()).isGreaterThan(0));
    });

    // verify registerInterest result at client2
    client2.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      await().untilAsserted(() -> assertThat(region.size()).isEqualTo(ONE_HUNDRED));

      for (int i = 0; i < ONE_HUNDRED; i++) {
        TickerData tickerData = region.getEntry("key-" + i).getValue();
        assertThat(tickerData.getPrice()).isEqualTo(i);
      }

      // then do update for key 10-20 to trigger CQ at server2
      // destroy key 10-14 to simulate create/update mix case
      region.removeAll(asList("key-10", "key-11", "key-12", "key-13", "key-14"));
      assertThat(region.get("key-10")).isNull();
      assertThat(region.get("key-11")).isNull();
      assertThat(region.get("key-12")).isNull();
      assertThat(region.get("key-13")).isNull();
      assertThat(region.get("key-14")).isNull();
    });

    client2.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      Map<String, TickerData> map = new LinkedHashMap<>();
      for (int i = 10; i < 20; i++) {
        map.put("key-" + i, new TickerData(i * 10));
      }
      region.putAll(map);
    });

    // verify CQ result at client1
    client1.invoke(() -> {
      Region<String, TickerData> localSaveRegion = getClientCache().getRegion("localsave");

      for (int i = 10; i < 20; i++) {
        String key = "key-" + i;
        int price = i * 10;
        await().untilAsserted(() -> {
          TickerData tickerData = localSaveRegion.get(key);
          assertThat(tickerData.getPrice()).isEqualTo(price);
        });
      }

      countDown(AFTER);
    });

    createCqInClient1.await();

    // Test Exception handling
    client1.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);

      Throwable thrown = catchThrowable(() -> region.putAll(null));
      assertThat(thrown).isInstanceOf(NullPointerException.class);

      region.localDestroyRegion();

      Map<String, TickerData> puts = new LinkedHashMap<>();
      for (int i = 1; i < 21; i++) {
        puts.put("key-" + i, null);
      }

      thrown = catchThrowable(() -> region.putAll(puts));
      assertThat(thrown).isInstanceOf(RegionDestroyedException.class);

      thrown = catchThrowable(() -> region.removeAll(asList("key-10", "key-11")));
      assertThat(thrown).isInstanceOf(RegionDestroyedException.class);
    });
  }

  /**
   * Tests putAll afterUpdate event contained oldValue.
   */
  @Test
  public void testOldValueInEvent() {
    // set notifyBySubscription=false to test local-invalidates
    int serverPort1 = server1.invoke(() -> new ServerBuilder()
        .regionShortcut(REPLICATE)
        .create());
    int serverPort2 = server2.invoke(() -> new ServerBuilder()
        .regionShortcut(REPLICATE)
        .create());

    client1.invoke(() -> new ClientBuilder()
        .serverPorts(serverPort1)
        .subscriptionAckInterval()
        .subscriptionEnabled(true)
        .subscriptionRedundancy()
        .create());
    client2.invoke(() -> new ClientBuilder()
        .serverPorts(serverPort2)
        .subscriptionAckInterval()
        .subscriptionEnabled(true)
        .subscriptionRedundancy()
        .create());

    client2.invoke(() -> {
      LATCH.set(new CountDownLatch(1));

      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      region.getAttributesMutator()
          .addCacheListener(new ActionCacheListener<>(new Action<>(Operation.UPDATE,
              event -> {
                assertThat(event.getOldValue()).isNotNull();
                VALUE.set(event.getOldValue());
                countDown(LATCH);
              })));
      region.registerInterest("ALL_KEYS");
    });

    client1.invoke(() -> {
      LATCH.set(new CountDownLatch(1));

      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      region.getAttributesMutator()
          .addCacheListener(new ActionCacheListener<>(new Action<>(Operation.UPDATE,
              event -> {
                assertThat(event.getOldValue()).isNotNull();
                VALUE.set(event.getOldValue());
                countDown(LATCH);
              })));

      // create keys
      doPutAll(region, keyPrefix, ONE_HUNDRED);
      assertThat(region.size()).isEqualTo(ONE_HUNDRED);

      // update keys
      doPutAll(region, keyPrefix, ONE_HUNDRED);
      assertThat(region.size()).isEqualTo(ONE_HUNDRED);
    });

    // the local PUTALL_UPDATE event should contain old value
    client1.invoke(() -> {
      awaitLatch(LATCH);
      assertThat(VALUE.get()).isInstanceOf(TickerData.class);
    });

    client2.invoke(() -> {
      awaitLatch(LATCH);
      assertThat(VALUE.get()).isInstanceOf(TickerData.class);
    });
  }

  /**
   * Create PR without redundancy on 2 servers with lucene index. Feed some key s. From a client, do
   * removeAll on keys in server1. During the removeAll, restart server1 and trigger the removeAll
   * to retry. The retried removeAll should return the version tag of tombstones. Do removeAll again
   * on the same key, it should get the version tag again.
   */
  @Test
  public void shouldReturnVersionTagOfTombstoneVersionWhenRemoveAllRetried() {
    // set notifyBySubscription=false to test local-invalidates
    int serverPort1 = server1.invoke(() -> new ServerBuilder()
        .regionShortcut(PARTITION_PERSISTENT)
        .create());

    client1.invoke(() -> new ClientBuilder()
        .serverPorts(serverPort1)
        .create());

    client1.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      doPutAll(region, "key-", ONE_HUNDRED);
      assertThat(region.size()).isEqualTo(ONE_HUNDRED);
    });

    // verify cache server 1, its data is from client
    server1.invoke(() -> {
      assertThat(getCache().getRegion(regionName).size()).isEqualTo(ONE_HUNDRED);
    });

    VersionedObjectList versions = client1.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      VersionedObjectList versionsToReturn = doRemoveAll(region, "key-", ONE_HUNDRED);
      assertThat(region.size()).isZero();
      return versionsToReturn;
    });

    // client1 removeAll again
    VersionedObjectList versionsAfterRetry = client1.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      VersionedObjectList versionsToReturn = doRemoveAll(region, "key-", ONE_HUNDRED);
      assertThat(region.size()).isZero();
      return versionsToReturn;
    });

    // noinspection rawtypes
    List<VersionTag> versionTags = versions.getVersionTags();
    // noinspection rawtypes
    List<VersionTag> versionTagsAfterRetry = versionsAfterRetry.getVersionTags();
    assertThat(versionTags)
        .hasSameSizeAs(versionTagsAfterRetry)
        .containsAll(versionTagsAfterRetry);
  }

  /**
   * Tests putAll and removeAll to 2 servers. Use Case: 1) putAll from a single-threaded client to a
   * replicated region 2) putAll from a multi-threaded client to a replicated region 3)
   */
  @Test
  public void test2Server() throws Exception {
    int serverPort1 = server1.invoke(() -> new ServerBuilder()
        .regionShortcut(REPLICATE)
        .create());
    int serverPort2 = server2.invoke(() -> new ServerBuilder()
        .regionShortcut(REPLICATE)
        .create());

    client1.invoke(() -> new ClientBuilder()
        .registerInterest(true)
        .serverPorts(serverPort1)
        .subscriptionAckInterval()
        .subscriptionEnabled(true)
        .subscriptionRedundancy()
        .create());
    client2.invoke(() -> new ClientBuilder()
        .registerInterest(true)
        .serverPorts(serverPort2)
        .subscriptionAckInterval()
        .subscriptionEnabled(true)
        .subscriptionRedundancy()
        .create());

    // client1 putAll
    client1.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      doPutAll(region, "key-", ONE_HUNDRED);
      assertThat(region.size()).isEqualTo(ONE_HUNDRED);

      for (int i = 0; i < ONE_HUNDRED; i++) {
        TickerData tickerData = region.getEntry("key-" + i).getValue();
        assertThat(tickerData.getPrice()).isEqualTo(i);
      }
    });

    // verify cache server 1
    server1.invoke(() -> {
      Region<String, TickerData> region = getCache().getRegion(regionName);
      region.getAttributesMutator().setCacheWriter(new CountingCacheWriter<>());
      assertThat(region.size()).isEqualTo(ONE_HUNDRED);

      for (int i = 0; i < ONE_HUNDRED; i++) {
        TickerData tickerData = region.getEntry("key-" + i).getValue();
        assertThat(tickerData.getPrice()).isEqualTo(i);
      }
    });

    // verify cache server 2
    server2.invoke(() -> {
      Region<String, TickerData> region = getCache().getRegion(regionName);
      region.getAttributesMutator().setCacheWriter(new CountingCacheWriter<>());
      assertThat(region.size()).isEqualTo(ONE_HUNDRED);

      for (int i = 0; i < ONE_HUNDRED; i++) {
        TickerData tickerData = region.getEntry("key-" + i).getValue();
        assertThat(tickerData.getPrice()).isEqualTo(i);
      }
    });

    // client2 verify putAll
    client2.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      await().untilAsserted(() -> assertThat(region.size()).isEqualTo(ONE_HUNDRED));

      for (int i = 0; i < ONE_HUNDRED; i++) {
        Entry<String, TickerData> regionEntry = region.getEntry("key-" + i);
        assertThat(regionEntry.getValue()).isNull();
      }
    });

    // client1 removeAll
    client1.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      doRemoveAll(region, "key-", ONE_HUNDRED);
      assertThat(region.size()).isZero();
    });

    // verify removeAll cache server 1
    server1.invoke(() -> {
      Region<String, TickerData> region = getCache().getRegion(regionName);
      assertThat(region.size()).isZero();

      CountingCacheWriter<String, TickerData> countingCacheWriter =
          (CountingCacheWriter<String, TickerData>) region.getAttributes().getCacheWriter();
      assertThat(countingCacheWriter.getDestroys()).isEqualTo(ONE_HUNDRED);
    });

    // verify removeAll cache server 2
    server2.invoke(() -> {
      Region<String, TickerData> region = getCache().getRegion(regionName);
      assertThat(region.size()).isZero();

      CountingCacheWriter<String, TickerData> countingCacheWriter =
          (CountingCacheWriter<String, TickerData>) region.getAttributes().getCacheWriter();
      // beforeDestroys are only triggered at server1 since removeAll is submitted from client1
      assertThat(countingCacheWriter.getDestroys()).isZero();
    });

    // client2 verify removeAll
    client2.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      await().untilAsserted(() -> assertThat(region.size()).isZero());
    });

    // async putAll1 from client1
    AsyncInvocation<Void> putAll1InClient1 = client1.invokeAsync(
        () -> doPutAll(getClientCache().getRegion(regionName), "async1key-", ONE_HUNDRED));

    // async putAll2 from client1
    AsyncInvocation<Void> putAll2InClient1 = client1.invokeAsync(
        () -> doPutAll(getClientCache().getRegion(regionName), "async2key-", ONE_HUNDRED));

    putAll1InClient1.await();
    putAll2InClient1.await();

    // verify client 1 for async keys
    client1.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      assertThat(region.size()).isEqualTo(ONE_HUNDRED * 2);

      long timeStamp1 = 0;
      long timeStamp2 = 0;
      for (int i = 0; i < ONE_HUNDRED; i++) {
        TickerData tickerData = region.getEntry("async1key-" + i).getValue();
        assertThat(tickerData.getPrice()).isEqualTo(i);
        assertThat(tickerData.getTimeStamp()).isGreaterThanOrEqualTo(timeStamp1);
        timeStamp1 = tickerData.getTimeStamp();

        tickerData = region.getEntry("async2key-" + i).getValue();
        assertThat(tickerData.getPrice()).isEqualTo(i);
        assertThat(tickerData.getTimeStamp()).isGreaterThanOrEqualTo(timeStamp2);
        timeStamp2 = tickerData.getTimeStamp();
      }
    });

    // verify cache server 1 for async keys
    server1.invoke(() -> {
      Region<String, TickerData> region = getCache().getRegion(regionName);
      assertThat(region.size()).isEqualTo(ONE_HUNDRED * 2);

      long timeStamp1 = 0;
      long timeStamp2 = 0;
      for (int i = 0; i < ONE_HUNDRED; i++) {
        TickerData tickerData = region.getEntry("async1key-" + i).getValue();
        assertThat(tickerData.getPrice()).isEqualTo(i);
        assertThat(tickerData.getTimeStamp()).isGreaterThanOrEqualTo(timeStamp1);
        timeStamp1 = tickerData.getTimeStamp();

        tickerData = region.getEntry("async2key-" + i).getValue();
        assertThat(tickerData.getPrice()).isEqualTo(i);
        assertThat(tickerData.getTimeStamp()).isGreaterThanOrEqualTo(timeStamp2);
        timeStamp2 = tickerData.getTimeStamp();
      }
    });

    // verify cache server 2 for async keys
    server2.invoke(() -> {
      Region<String, TickerData> region = getCache().getRegion(regionName);
      assertThat(region.size()).isEqualTo(ONE_HUNDRED * 2);

      long timeStamp1 = 0;
      long timeStamp2 = 0;
      for (int i = 0; i < ONE_HUNDRED; i++) {
        TickerData tickerData = region.getEntry("async1key-" + i).getValue();
        assertThat(tickerData.getPrice()).isEqualTo(i);
        assertThat(tickerData.getTimeStamp()).isGreaterThanOrEqualTo(timeStamp1);
        timeStamp1 = tickerData.getTimeStamp();

        tickerData = region.getEntry("async2key-" + i).getValue();
        assertThat(tickerData.getPrice()).isEqualTo(i);
        assertThat(tickerData.getTimeStamp()).isGreaterThanOrEqualTo(timeStamp2);
        timeStamp2 = tickerData.getTimeStamp();
      }
    });

    // client2 verify async putAll
    client2.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      await().untilAsserted(() -> assertThat(region.size()).isEqualTo(ONE_HUNDRED * 2));

      for (int i = 0; i < ONE_HUNDRED; i++) {
        Entry<String, TickerData> regionEntry = region.getEntry("async1key-" + i);
        assertThat(regionEntry.getValue()).isNull();
      }

      for (int i = 0; i < ONE_HUNDRED; i++) {
        Entry<String, TickerData> regionEntry = region.getEntry("async2key-" + i);
        assertThat(regionEntry.getValue()).isNull();
      }
    });

    // async removeAll1 from client1
    AsyncInvocation<Void> removeAll1InClient1 = client1.invokeAsync(() -> {
      doRemoveAll(getClientCache().getRegion(regionName), "async1key-", ONE_HUNDRED);
    });

    // async removeAll2 from client1
    AsyncInvocation<Void> removeAll2InClient1 = client1.invokeAsync(() -> {
      doRemoveAll(getClientCache().getRegion(regionName), "async2key-", ONE_HUNDRED);
    });

    removeAll1InClient1.await();
    removeAll2InClient1.await();

    // client1 removeAll
    client1.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      doRemoveAll(region, "key-", ONE_HUNDRED);
      assertThat(region.size()).isZero();
    });

    // verify async removeAll cache server 1
    server1.invoke(() -> {
      assertThat(getCache().getRegion(regionName).size()).isZero();
    });

    // verify async removeAll cache server 2
    server2.invoke(() -> {
      assertThat(getCache().getRegion(regionName).size()).isZero();
    });

    // client2 verify async removeAll
    client2.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      await().untilAsserted(() -> assertThat(region.size()).isZero());
    });

    // server1 execute P2P putAll
    server1.invoke(() -> {
      Region<String, TickerData> region = getCache().getRegion(regionName);
      doPutAll(region, "p2pkey-", ONE_HUNDRED);

      long timeStamp = 0;
      for (int i = 0; i < ONE_HUNDRED; i++) {
        TickerData tickerData = region.getEntry("p2pkey-" + i).getValue();
        assertThat(tickerData.getPrice()).isEqualTo(i);
        assertThat(tickerData.getTimeStamp()).isGreaterThanOrEqualTo(timeStamp);
        timeStamp = tickerData.getTimeStamp();
      }
    });

    // verify cache server 2 for p2p keys
    server2.invoke(() -> {
      Region<String, TickerData> region = getCache().getRegion(regionName);

      long timeStamp = 0;
      for (int i = 0; i < ONE_HUNDRED; i++) {
        TickerData tickerData = region.getEntry("p2pkey-" + i).getValue();
        assertThat(tickerData.getPrice()).isEqualTo(i);
        assertThat(tickerData.getTimeStamp()).isGreaterThanOrEqualTo(timeStamp);
        timeStamp = tickerData.getTimeStamp();
      }
    });

    // client2 verify p2p putAll
    client2.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      await().untilAsserted(() -> assertThat(region.size()).isEqualTo(ONE_HUNDRED));

      for (int i = 0; i < ONE_HUNDRED; i++) {
        Entry<String, TickerData> regionEntry = region.getEntry("p2pkey-" + i);
        assertThat(regionEntry.getValue()).isNull();
      }
    });

    // client1 verify p2p putAll
    client1.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      await().untilAsserted(() -> assertThat(region.size()).isEqualTo(ONE_HUNDRED));

      for (int i = 0; i < ONE_HUNDRED; i++) {
        Entry<String, TickerData> regionEntry = region.getEntry("p2pkey-" + i);
        assertThat(regionEntry.getValue()).isNull();
      }
    });

    // server1 execute P2P removeAll
    server1.invoke(() -> {
      Region<String, TickerData> region = getCache().getRegion(regionName);
      doRemoveAll(region, "p2pkey-", ONE_HUNDRED);
      assertThat(region.size()).isZero();
    });

    // verify p2p removeAll cache server 2
    server2.invoke(() -> {
      assertThat(getCache().getRegion(regionName).size()).isZero();
    });

    // client2 verify p2p removeAll
    client2.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      await().untilAsserted(() -> assertThat(region.size()).isZero());
    });

    // client1 verify p2p removeAll
    client1.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      await().untilAsserted(() -> assertThat(region.size()).isZero());
    });

    // execute putAll on client2 for key 0-10
    client2.invoke(() -> doPutAll(getClientCache().getRegion(regionName), "key-", 10));

    // verify client1 for local invalidate
    client1.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      for (int i = 0; i < 10; i++) {
        String key = "key-" + i;

        await("entry with null value exists for " + key).until(() -> {
          Entry<String, TickerData> regionEntry = region.getEntry(key);
          return regionEntry != null && regionEntry.getValue() == null;
        });

        // local invalidate will set the value to null
        TickerData tickerData = region.getEntry("key-" + i).getValue();
        assertThat(tickerData).isNull();
      }
    });
  }

  @Test
  @Parameters({"true", "false"})
  @TestCaseName("{method}(concurrencyChecksEnabled={0})")
  public void test2NormalServer(boolean concurrencyChecksEnabled) {
    Properties config = getDistributedSystemProperties();
