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

import static org.apache.geode.cache.Region.SEPARATOR;
import static org.apache.geode.cache.wan.internal.cli.commands.WanCopyRegionCommand.WAN_COPY_REGION;
import static org.apache.geode.cache.wan.internal.cli.commands.WanCopyRegionCommand.WAN_COPY_REGION__BATCHSIZE;
import static org.apache.geode.cache.wan.internal.cli.commands.WanCopyRegionCommand.WAN_COPY_REGION__CANCEL;
import static org.apache.geode.cache.wan.internal.cli.commands.WanCopyRegionCommand.WAN_COPY_REGION__MAXRATE;
import static org.apache.geode.cache.wan.internal.cli.commands.WanCopyRegionCommand.WAN_COPY_REGION__REGION;
import static org.apache.geode.cache.wan.internal.cli.commands.WanCopyRegionCommand.WAN_COPY_REGION__SENDERID;
import static org.apache.geode.distributed.ConfigurationProperties.DISTRIBUTED_SYSTEM_ID;
import static org.apache.geode.distributed.ConfigurationProperties.REMOTE_LOCATORS;
import static org.apache.geode.management.internal.cli.functions.WanCopyRegionFunction.WAN_COPY_REGION__MSG__ALREADY__RUNNING__COMMAND;
import static org.apache.geode.management.internal.cli.functions.WanCopyRegionFunction.WAN_COPY_REGION__MSG__CANCELED__BEFORE__HAVING__COPIED;
import static org.apache.geode.management.internal.cli.functions.WanCopyRegionFunction.WAN_COPY_REGION__MSG__EXECUTION__CANCELED;
import static org.apache.geode.management.internal.cli.functions.WanCopyRegionFunction.WAN_COPY_REGION__MSG__NO__RUNNING__COMMAND;
import static org.apache.geode.management.internal.cli.functions.WanCopyRegionFunction.WAN_COPY_REGION__MSG__REGION__NOT__FOUND;
import static org.apache.geode.management.internal.cli.functions.WanCopyRegionFunction.WAN_COPY_REGION__MSG__SENDER__NOT__FOUND;
import static org.apache.geode.management.internal.cli.functions.WanCopyRegionFunction.WAN_COPY_REGION__MSG__SENDER__SERIAL__AND__NOT__PRIMARY;
import static org.apache.geode.management.internal.cli.functions.WanCopyRegionFunctionDelegate.WAN_COPY_REGION__MSG__COPIED__ENTRIES;
import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
import static org.apache.geode.test.dunit.IgnoredException.addIgnoredException;
import static org.assertj.core.api.Assertions.assertThat;

import java.text.NumberFormat;
import java.text.ParseException;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Locale;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
import java.util.stream.LongStream;

import com.google.common.collect.ImmutableList;
import junitparams.Parameters;
import org.assertj.core.api.Condition;
import org.junit.BeforeClass;
import org.junit.Rule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.junit.runner.RunWith;

import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionShortcut;
import org.apache.geode.cache.Scope;
import org.apache.geode.cache.client.ClientRegionShortcut;
import org.apache.geode.cache.wan.GatewaySender;
import org.apache.geode.cache.wan.internal.WanCopyRegionFunctionService;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.wan.InternalGatewaySender;
import org.apache.geode.internal.cache.wan.WANTestBase;
import org.apache.geode.management.internal.cli.result.model.ResultModel;
import org.apache.geode.management.internal.cli.util.CommandStringBuilder;
import org.apache.geode.management.internal.i18n.CliStrings;
import org.apache.geode.test.dunit.AsyncInvocation;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.rules.DistributedErrorCollector;
import org.apache.geode.test.dunit.rules.DistributedExecutorServiceRule;
import org.apache.geode.test.junit.assertions.CommandResultAssert;
import org.apache.geode.test.junit.categories.WanTest;
import org.apache.geode.test.junit.rules.GfshCommandRule;
import org.apache.geode.test.junit.rules.LocatorLauncherStartupRule;
import org.apache.geode.test.junit.runners.GeodeParamsRunner;

@RunWith(GeodeParamsRunner.class)
@Category({WanTest.class})
public class WanCopyRegionCommandDistributedTest extends WANTestBase {

  protected static VM vm8;
  protected static VM vm9;

  private static final long serialVersionUID = 1L;

  private enum Gateway {
    SENDER, RECEIVER
  }

  @Rule
  public transient GfshCommandRule gfsh = new GfshCommandRule();

  @Rule
  public DistributedErrorCollector errorCollector = new DistributedErrorCollector();

  @Rule
  public DistributedExecutorServiceRule executorServiceRule = new DistributedExecutorServiceRule();

  @BeforeClass
  public static void beforeClassWanCopyRegionCommandDUnitTest() {
    vm8 = VM.getVM(8);
    vm9 = VM.getVM(9);
  }

  @Test
  public void testUnsuccessfulExecution_RegionNotFound() throws Exception {
    List<VM> serversInA = Arrays.asList(vm5, vm6, vm7);
    List<VM> serversInB = Collections.singletonList(vm3);
    VM serverInC = vm4;
    VM client = vm8;
    String senderIdInA = "B";
    String senderIdInB = "C";

    int senderLocatorPort = create3WanSitesAndClient(true, vm0,
        vm1, vm2, serversInA, serversInB, serverInC, client,
        senderIdInA, senderIdInB, true);

    int wanCopyRegionBatchSize = 20;
    String regionName = "foo";

    // Execute wan-copy region command
    gfsh.connectAndVerify(senderLocatorPort, GfshCommandRule.PortType.locator);
    String commandString = new CommandStringBuilder(WAN_COPY_REGION)
        .addOption(WAN_COPY_REGION__REGION, regionName)
        .addOption(WAN_COPY_REGION__SENDERID, senderIdInA)
        .addOption(WAN_COPY_REGION__BATCHSIZE, String.valueOf(wanCopyRegionBatchSize))
        .getCommandString();

    // Check command status and output
    CommandResultAssert command =
        verifyStatusIsError(gfsh.executeAndAssertThat(commandString), serversInA.size());
    String message =
        CliStrings.format(WAN_COPY_REGION__MSG__REGION__NOT__FOUND,
            regionName);
    command
        .hasTableSection(ResultModel.MEMBER_STATUS_SECTION)
        .hasColumn("Message")
        .containsExactly(message, message, message);
  }

  @Test
  public void testUnsuccessfulExecution_SenderNotFound() throws Exception {
    List<VM> serversInA = Arrays.asList(vm5, vm6, vm7);
    List<VM> serversInB = Collections.singletonList(vm3);
    VM serverInC = vm4;
    VM client = vm8;
    String senderIdInA = "B";
    String senderIdInB = "C";

    int senderLocatorPort = create3WanSitesAndClient(true, vm0,
        vm1, vm2, serversInA, serversInB, serverInC, client,
        senderIdInA, senderIdInB, true);

    int wanCopyRegionBatchSize = 20;
    String regionName = getRegionName(true);

    // Execute wan-copy region command
    gfsh.connectAndVerify(senderLocatorPort, GfshCommandRule.PortType.locator);
    String commandString = new CommandStringBuilder(WAN_COPY_REGION)
        .addOption(WAN_COPY_REGION__REGION, regionName)
        .addOption(WAN_COPY_REGION__SENDERID, senderIdInA)
        .addOption(WAN_COPY_REGION__BATCHSIZE, String.valueOf(wanCopyRegionBatchSize))
        .getCommandString();

    // Check command status and output
    CommandResultAssert command =
        verifyStatusIsError(gfsh.executeAndAssertThat(commandString), serversInA.size());
    String message =
        CliStrings.format(WAN_COPY_REGION__MSG__SENDER__NOT__FOUND, senderIdInA);
    command
        .hasTableSection(ResultModel.MEMBER_STATUS_SECTION)
        .hasColumn("Message")
        .containsExactly(message, message, message);
  }

  @Test
  @Parameters({"true, true", "true, false", "false, false"})
  public void testUnsuccessfulExecution_ExceptionAtReceiver(
      boolean isPartitionedRegion, boolean isParallelGatewaySender) throws Exception {
    List<VM> serversInA = Arrays.asList(vm5, vm6, vm7);
    List<VM> serversInB = Collections.singletonList(vm3);
    VM serverInC = vm4;
    VM client = vm8;
    String senderIdInA = "B";
    String senderIdInB = "C";

    int senderLocatorPort = create3WanSitesAndClient(isPartitionedRegion, vm0,
        vm1, vm2, serversInA, serversInB, serverInC, client,
        senderIdInA, senderIdInB, true);

    String regionName = getRegionName(isPartitionedRegion);
    int wanCopyRegionBatchSize = 20;

    int entries = 20;
    // Put entries
    client.invoke(() -> doPutsFrom(regionName, 0, entries));

    // Check that entries are put in the region
    for (VM member : serversInA) {
      member.invoke(() -> validateRegionSize(regionName, entries));
    }

    // destroy region to provoke the exception
    for (VM server : serversInB) {
      server.invoke(() -> destroyRegion(regionName));
    }

    // Create senders and receivers
    createSender(isParallelGatewaySender, serversInA, senderIdInA, 2);
    createSender(isParallelGatewaySender, serversInB, senderIdInB, 3);
    createReceivers(serversInB, serverInC);

    // Execute wan-copy region command
    gfsh.connectAndVerify(senderLocatorPort, GfshCommandRule.PortType.locator);
    String commandString = new CommandStringBuilder(WAN_COPY_REGION)
        .addOption(WAN_COPY_REGION__REGION, regionName)
        .addOption(WAN_COPY_REGION__SENDERID, senderIdInA)
        .addOption(WAN_COPY_REGION__BATCHSIZE, String.valueOf(wanCopyRegionBatchSize))
        .getCommandString();

    // Check command status and output
    if (isParallelGatewaySender) {
      CommandResultAssert command =
          verifyStatusIsError(gfsh.executeAndAssertThat(commandString), serversInA.size());
      Condition<String> exceptionError =
          new Condition<>(s -> s.startsWith("Error ("), "Error");
      command
          .hasTableSection(ResultModel.MEMBER_STATUS_SECTION)
          .hasColumn("Message")
          .asList()
          .haveExactly(3, exceptionError);
    } else {
      CommandResultAssert command =
          verifyStatusIsErrorInOneServer(gfsh.executeAndAssertThat(commandString));
      Condition<String> exceptionError =
          new Condition<>(s -> s.startsWith("Error ("), "Error");
      Condition<String> senderNotPrimary = new Condition<>(
          s -> s.equals(CliStrings
              .format(WAN_COPY_REGION__MSG__SENDER__SERIAL__AND__NOT__PRIMARY,
                  senderIdInA)),
          "sender not primary");
      command
          .hasTableSection(ResultModel.MEMBER_STATUS_SECTION)
          .hasColumn("Message")
          .asList()
          .haveExactly(1, exceptionError)
          .haveExactly(2, senderNotPrimary);
    }
  }

  /**
   * Used by {@link #testSenderOrReceiverGoesDownDuringExecution} in annotation.
   */
  @SuppressWarnings("unused")
  private Object[] parametersToTestSenderOrReceiverGoesDownDuringExecution() {
    return new Object[] {
        new Object[] {true, true, Gateway.SENDER, false},
        new Object[] {false, true, Gateway.SENDER, true},
        new Object[] {false, true, Gateway.SENDER, false},
        new Object[] {false, false, Gateway.SENDER, true},
        new Object[] {false, false, Gateway.SENDER, false},
        new Object[] {true, true, Gateway.RECEIVER, false},
        new Object[] {false, true, Gateway.RECEIVER, true},
        new Object[] {false, true, Gateway.RECEIVER, false},
        new Object[] {false, false, Gateway.RECEIVER, true},
        new Object[] {false, false, Gateway.RECEIVER, false}
    };
  }

  /**
   * This test creates two sites A & B, each one containing 3 servers.
   * A region is created in both sites, and populated in site A.
   * After that replication is configured from site A to site B.
   * WanCopyRegionFunction is called, and while it is running, a sender in site A
   * or a receiver in site B are killed.
   */
  @Test
  @Parameters(method = "parametersToTestSenderOrReceiverGoesDownDuringExecution")
  public void testSenderOrReceiverGoesDownDuringExecution(boolean useParallel,
      boolean usePartitionedRegion, Gateway gwToBeStopped, boolean stopPrimarySender)
      throws Exception {

    if (gwToBeStopped == Gateway.SENDER && (useParallel || stopPrimarySender)) {
      addIgnoredExceptionsForSenderInUseWentDown();
    }
    if (gwToBeStopped == Gateway.RECEIVER &&
        (usePartitionedRegion || !useParallel)) {
      addIgnoredExceptionsForReceiverConnectedToSenderInUseWentDown();
    }
    final int wanCopyRegionBatchSize = 10;
    final int entries;
    if (!useParallel && !usePartitionedRegion && stopPrimarySender) {
      entries = 2500;
    } else {
      entries = 1000;
    }

    final String regionName = getRegionName(usePartitionedRegion);

    // Site A
    VM locatorInA = vm0;
    VM server1InA = vm1;
    VM server2InA = vm2;
    VM server3InA = vm3;
    List<VM> serversInA = Arrays.asList(server1InA, server2InA, server3InA);
    final String senderIdInA = "B";

    // Site B
    VM locatorInB = vm4;
    VM server1InB = vm5;
    VM server2InB = vm6;
    VM server3InB = vm7;
    List<VM> serversInB = Arrays.asList(server1InB, server2InB, server3InB);
    VM client = vm8;

    int locatorAPort = create2WanSitesAndClient(locatorInA, serversInA, senderIdInA, locatorInB,
        serversInB, client, usePartitionedRegion, regionName);

    // Put entries
    client.invoke(() -> doPutsFrom(regionName, 0, entries));
    for (VM member : serversInA) {
      member.invoke(() -> validateRegionSize(regionName, entries));
    }

    // Create senders and receivers with replication as follows: "A" -> "B"
    if (useParallel) {
      createReceiverInVMs(server1InB, server2InB, server3InB);
      createSender(useParallel, serversInA, senderIdInA, 2);
    } else {
      // Senders will connect to receiver in server1InB
      server1InB.invoke(WANTestBase::createReceiver);
      createSender(useParallel, serversInA, senderIdInA, 2);
      createReceiverInVMs(server2InB, server3InB);
    }

    Callable<CommandResultAssert> wanCopyCommandCallable = () -> {
      String command = new CommandStringBuilder(WAN_COPY_REGION)
          .addOption(WAN_COPY_REGION__REGION, regionName)
          .addOption(WAN_COPY_REGION__SENDERID, senderIdInA)
          .addOption(WAN_COPY_REGION__BATCHSIZE, String.valueOf(wanCopyRegionBatchSize))
          .addOption(WAN_COPY_REGION__MAXRATE, "50")
          .getCommandString();
      try {
        gfsh.connectAndVerify(locatorAPort, GfshCommandRule.PortType.locator);
      } catch (Exception e) {
        errorCollector.addError(e);
      }
      return gfsh.executeAndAssertThat(command);
    };

    Future<CommandResultAssert> wanCopyCommandFuture =
        executorServiceRule.submit(wanCopyCommandCallable);

    // Wait for the wan-copy command to start
    waitForWanCopyRegionCommandToStart(useParallel, usePartitionedRegion, serversInA);

    // Stop sender or receiver and verify result
    if (gwToBeStopped == Gateway.SENDER) {
      stopSenderAndVerifyResult(useParallel, stopPrimarySender, server2InA, serversInA, senderIdInA,
          wanCopyCommandFuture);
    } else if (gwToBeStopped == Gateway.RECEIVER) {
      stopReceiverAndVerifyResult(useParallel, stopPrimarySender, entries, regionName, server1InB,
          server2InB, server3InB, serversInA.size(), wanCopyCommandFuture);
    }
  }

  @Test
  @Parameters({"false, false", "false, true", "true, true"})
  public void testRegionDestroyedDuringExecution(boolean isParallelGatewaySender,
      boolean isPartitionedRegion)
      throws Exception {
    addIgnoredException("org.apache.geode.cache.RegionDestroyedException");
    final int wanCopyRegionBatchSize = 10;
    final int entries = 1000;

    final String regionName = getRegionName(isPartitionedRegion);

    // Site A
    VM locatorInA = vm0;
    VM server1InA = vm1;
    List<VM> serversInA = ImmutableList.of(server1InA);
    final String senderIdInA = "B";

    // Site B
    VM locatorInB = vm4;
    VM server1InB = vm5;
    List<VM> serversInB = Collections.singletonList(server1InB);
    VM client = vm8;

    int locatorAPort = create2WanSitesAndClient(locatorInA, serversInA, senderIdInA, locatorInB,
        serversInB, client, isPartitionedRegion, regionName);

    // Put entries
    client.invoke(() -> doPutsFrom(regionName, 0, entries));
    for (VM member : serversInA) {
      member.invoke(() -> validateRegionSize(regionName, entries));
    }

    // Create senders and receivers with replication as follows: "A" -> "B"
    createReceiverInVMs(server1InB);
    createSender(isParallelGatewaySender, serversInA, senderIdInA, 2);

    Callable<CommandResultAssert> wanCopyCommandCallable = () -> {
      String command = new CommandStringBuilder(WAN_COPY_REGION)
          .addOption(WAN_COPY_REGION__REGION, regionName)
          .addOption(WAN_COPY_REGION__SENDERID, senderIdInA)
          .addOption(WAN_COPY_REGION__BATCHSIZE, String.valueOf(wanCopyRegionBatchSize))
          .addOption(WAN_COPY_REGION__MAXRATE, "5")
          .getCommandString();
      try {
        gfsh.connectAndVerify(locatorAPort, GfshCommandRule.PortType.locator);
      } catch (Exception e) {
        errorCollector.addError(e);
      }
      return gfsh.executeAndAssertThat(command);
    };

    Future<CommandResultAssert> wanCopyCommandFuture =
        executorServiceRule.submit(wanCopyCommandCallable);

    // Wait for the wan-copy command to start
    waitForWanCopyRegionCommandToStart(isParallelGatewaySender, isPartitionedRegion, serversInA);

    // Destroy region in a server in A
    server1InA.invoke(() -> cache.getRegion(regionName).destroyRegion());

    CommandResultAssert result = wanCopyCommandFuture.get();
    result.statusIsError();
    result
        .hasTableSection()
        .hasColumns()
        .hasSize(3);
    result
        .hasTableSection(ResultModel.MEMBER_STATUS_SECTION)
        .hasColumn("Member")
        .hasSize(1);
    result
        .hasTableSection(ResultModel.MEMBER_STATUS_SECTION)
        .hasColumn("Message")
        .hasSize(1);
    result
        .hasTableSection(ResultModel.MEMBER_STATUS_SECTION)
        .hasColumn("Status")
        .containsExactly("ERROR");

    Condition<String> regionDestroyedError = new Condition<>(
        s -> (s.startsWith(
            "Execution failed. Error: org.apache.geode.cache.RegionDestroyedException: ")
            || s.startsWith("Error (Region destroyed) in operation after having copied")),
        "execution error");
    result
        .hasTableSection(ResultModel.MEMBER_STATUS_SECTION)
        .hasColumn("Message")
        .asList()
        .haveExactly(1, regionDestroyedError);
  }

  @Test
  @Parameters({"false, false", "false, true", "true, true"})
  public void testDetectOngoingExecution(boolean useParallel,
      boolean usePartitionedRegion)
      throws Exception {
    final int wanCopyRegionBatchSize = 10;
    final int entries = 1000;

    final String regionName = getRegionName(usePartitionedRegion);

    // Site A
    VM locatorInA = vm0;
    VM server1InA = vm1;
    VM server2InA = vm2;
    VM server3InA = vm3;
    List<VM> serversInA = Arrays.asList(server1InA, server2InA, server3InA);
    final String senderIdInA = "B";

    // Site B
    VM locatorInB = vm4;
    VM server1InB = vm5;
    VM server2InB = vm6;
    VM server3InB = vm7;
    List<VM> serversInB = Arrays.asList(server1InB, server2InB, server3InB);
    VM client = vm8;

    int locatorAPort = create2WanSitesAndClient(locatorInA, serversInA, senderIdInA, locatorInB,
        serversInB, client, usePartitionedRegion, regionName);

    // Put entries
    client.invoke(() -> doPutsFrom(regionName, 0, entries));
    for (VM member : serversInA) {
      member.invoke(() -> validateRegionSize(regionName, entries));
    }

    // Create senders and receivers with replication as follows: "A" -> "B"
    if (useParallel) {
      createReceiverInVMs(server1InB, server2InB, server3InB);
      createSender(useParallel, serversInA, senderIdInA, 2);
    } else {
      // Senders will connect to receiver in server1InB
      server1InB.invoke(WANTestBase::createReceiver);
      createSender(useParallel, serversInA, senderIdInA, 2);
      createReceiverInVMs(server2InB, server3InB);
    }

    Callable<CommandResultAssert> wanCopyCommandCallable = () -> {
      String command = new CommandStringBuilder(WAN_COPY_REGION)
          .addOption(WAN_COPY_REGION__REGION, regionName)
          .addOption(WAN_COPY_REGION__SENDERID, senderIdInA)
          .addOption(WAN_COPY_REGION__BATCHSIZE, String.valueOf(wanCopyRegionBatchSize))
          .addOption(WAN_COPY_REGION__MAXRATE, "5")
          .getCommandString();
      try {
        gfsh.connectAndVerify(locatorAPort, GfshCommandRule.PortType.locator);
      } catch (Exception e) {
        errorCollector.addError(e);
      }
      return gfsh.executeAndAssertThat(command);
    };

    Future<CommandResultAssert> wanCopyCommandFuture =
        executorServiceRule.submit(wanCopyCommandCallable);

    // Wait for the wan-copy command to start
    waitForWanCopyRegionCommandToStart(useParallel, usePartitionedRegion, serversInA);

    // Execute again the same wan-copy region command
    String commandString = new CommandStringBuilder(WAN_COPY_REGION)
        .addOption(WAN_COPY_REGION__REGION, regionName)
        .addOption(WAN_COPY_REGION__SENDERID, senderIdInA)
        .addOption(WAN_COPY_REGION__BATCHSIZE, String.valueOf(wanCopyRegionBatchSize))
        .getCommandString();

    // Check command status and output
    Condition<String> exceptionError = new Condition<>(
        s -> s.equals(CliStrings.format(WAN_COPY_REGION__MSG__ALREADY__RUNNING__COMMAND,
            regionName, senderIdInA)),
        "already running");
    if (useParallel) {
      CommandResultAssert command =
          verifyStatusIsError(gfsh.executeAndAssertThat(commandString), serversInA.size());
      command
          .hasTableSection(ResultModel.MEMBER_STATUS_SECTION)
          .hasColumn("Message")
          .asList()
          .haveExactly(3, exceptionError);
    } else {
      CommandResultAssert command =
          verifyStatusIsErrorInOneServer(gfsh.executeAndAssertThat(commandString));
      Condition<String> senderNotPrimary = new Condition<>(
          s -> s.equals(CliStrings
              .format(WAN_COPY_REGION__MSG__SENDER__SERIAL__AND__NOT__PRIMARY,
                  senderIdInA)),
          "sender not primary");
      command
          .hasTableSection(ResultModel.MEMBER_STATUS_SECTION)
          .hasColumn("Message")
          .asList()
          .haveExactly(1, exceptionError)
          .haveExactly(2, senderNotPrimary);
    }

    // cancel command
    String commandString1 = new CommandStringBuilder(WAN_COPY_REGION)
        .addOption(WAN_COPY_REGION__REGION, regionName)
        .addOption(WAN_COPY_REGION__SENDERID, senderIdInA)
        .addOption(WAN_COPY_REGION__BATCHSIZE, String.valueOf(wanCopyRegionBatchSize))
        .addOption(WAN_COPY_REGION__CANCEL)
        .getCommandString();
    gfsh.executeAndAssertThat(commandString1);
    wanCopyCommandFuture.get();
    addIgnoredExceptionsForClosingAfterCancelCommand();
  }

  /**
   * Scenario with 3 WAN sites: "A", "B" and "C".
   * Initially, no replication is configured between sites.
   * Several entries are put in WAN site "A".
   *
   * The following gateway senders are created and started:
   * - In "A" site: to replicate region entries to "B" site. Sender called "B".
   * - In "B" site: to replicate region entries to "C" site. Sender called "C".
   * (Replication is as follows: A -> B -> C)
   *
   * The "wan-copy region" command is run from "A" site passing sender "B".
   *
   * It must be verified that the entries are copied to site "B".
   * It must also be verified that the entries are not transitively
   * copied to "C" even though replication is configured from "B" to "C"
   * because with this command generateCallbacks is set to false in the
   * events generated.
   *
   */
  @Test
  @Parameters({"true, true, true", "true, false, true", "false, false, true", "true, true, false",
      "true, false, false", "false, false, false"})
  public void testSuccessfulExecution(boolean isPartitionedRegion,
      boolean isParallelGatewaySender, boolean concurrencyChecksEnabled) throws Exception {
    List<VM> serversInA = Arrays.asList(vm5, vm6, vm7);
    List<VM> serversInB = Arrays.asList(vm9, vm3);
    VM serverInC = vm4;
    VM client = vm8;
    String senderIdInA = "B";
    String senderIdInB = "C";

    int senderLocatorPort = create3WanSitesAndClient(isPartitionedRegion, vm0,
        vm1, vm2, serversInA, serversInB, serverInC, client,
        senderIdInA, senderIdInB, concurrencyChecksEnabled);

    int wanCopyRegionBatchSize = 20;
    int entries = 100;
    String regionName = getRegionName(isPartitionedRegion);
    // Put entries
    client.invoke(() -> doPutsFrom(regionName, 0, entries + 1));
    // remove an entry to make sure that replication works well even when removes.
    client.invoke(() -> removeEntry(regionName, entries));

    // Check that entries are put in the region
    for (VM member : serversInA) {
      member.invoke(() -> validateRegionSize(regionName, entries));
    }

    // Check that entries are not copied to "B" nor "C"
    for (VM serverInB : serversInB) {
      serverInB.invoke(() -> validateRegionSize(regionName, 0));
    }
    serverInC.invoke(() -> validateRegionSize(regionName, 0));

    // Create senders and receivers with replication as follows: "A" -> "B" -> "C"
    createSender(isParallelGatewaySender, serversInA, senderIdInA, 2);
    createSender(isParallelGatewaySender, serversInB, senderIdInB, 3);
    createReceivers(serversInB, serverInC);

    // Check that entries are not copied to "B" nor "C"
    serversInB.get(0).invoke(() -> validateRegionSize(regionName, 0));
    serverInC.invoke(() -> validateRegionSize(regionName, 0));

    // Execute wan-copy region command
    gfsh.connectAndVerify(senderLocatorPort, GfshCommandRule.PortType.locator);
    String commandString = new CommandStringBuilder(WAN_COPY_REGION)
        .addOption(WAN_COPY_REGION__REGION, regionName)
        .addOption(WAN_COPY_REGION__SENDERID, senderIdInA)
        .addOption(WAN_COPY_REGION__BATCHSIZE, String.valueOf(wanCopyRegionBatchSize))
        .getCommandString();

    // Check command status and output
    CommandResultAssert command =
        verifyStatusIsOk(gfsh.executeAndAssertThat(commandString), serversInA.size());
    if (isPartitionedRegion && isParallelGatewaySender) {
      String msg1 = CliStrings.format(WAN_COPY_REGION__MSG__COPIED__ENTRIES, 33);
      String msg2 = CliStrings.format(WAN_COPY_REGION__MSG__COPIED__ENTRIES, 34);
      command
          .hasTableSection(ResultModel.MEMBER_STATUS_SECTION)
          .hasColumn("Message")
          .containsExactlyInAnyOrder(msg1, msg1, msg2);
    } else {
      String msg1 = CliStrings.format(WAN_COPY_REGION__MSG__COPIED__ENTRIES, entries);
      String msg2 = CliStrings
          .format(WAN_COPY_REGION__MSG__SENDER__SERIAL__AND__NOT__PRIMARY, senderIdInA);
      command
          .hasTableSection(ResultModel.MEMBER_STATUS_SECTION)
          .hasColumn("Message")
          .containsExactlyInAnyOrder(msg1, msg2, msg2);
    }

    // Check that entries are copied in "B"
    for (VM serverInB : serversInB) {
      serverInB.invoke(() -> validateRegionSize(regionName, entries));
    }

    // Check that no entries are left in the queue from B to C
    for (VM serverInB : serversInB) {
      serverInB.invoke(() -> getSenderStats(senderIdInB, 0));
    }

    // Check that the region's data is the same in sites "A" and "B"
    checkEqualRegionData(regionName, serversInA.get(0), serversInB.get(0),
        concurrencyChecksEnabled);

    // Check that wanCopyRegionBatchSize is correctly used by the command
    long receivedBatches = 0;
    for (VM serverInB : serversInB) {
      receivedBatches += serverInB.invoke(() -> getReceiverStats().get(2));
    }
    if (isPartitionedRegion && isParallelGatewaySender) {
      assertThat(receivedBatches).isEqualTo(6);
    } else {
      assertThat(receivedBatches).isEqualTo(5);
    }

    // Check that entries are not copied in "C" (generateCallbacks is false)
    serverInC.invoke(() -> validateRegionSize(regionName, 0));
  }

  /**
   * Scenario with 2 WAN sites: "A" and "B".
   * Initially, no replication is configured between sites.
   * Several entries are put in WAN site "A".
   *
   * The following gateway senders are created and started:
   * - In "A" site: to replicate region entries to "B" site. Sender called "B".
   * (Replication is as follows: A -> B)
   *
   * The "wan-copy region" command is run from "A" site passing sender "B".
   * Simultaneously, random operations for entries with the same
   * keys as the one previously put are run.
   *
   * When the command finishes and the puts finish it
   * must be verified that the entries in the region in site "A"
   * are the same as the ones in region in site "B".
   */
  @Test
  @Parameters({"true, true", "true, false", "false, false"})
  public void testSuccessfulExecutionWhileRunningOpsOnRegion(
      boolean isPartitionedRegion,
      boolean isParallelGatewaySender) throws Exception {
    List<VM> serversInA = Arrays.asList(vm5, vm6, vm7);
    List<VM> serversInB = Collections.singletonList(vm3);
    VM serverInC = vm4;
    VM client = vm8;
    String senderIdInA = "B";
    String senderIdInB = "C";

    int senderLocatorPort = create3WanSitesAndClient(isPartitionedRegion, vm0,
        vm1, vm2, serversInA, serversInB, serverInC, client,
        senderIdInA, senderIdInB, true);

    int wanCopyRegionBatchSize = 20;
    int entries = 10000;
    Set<Long> keySet = LongStream.range(0L, entries).boxed().collect(Collectors.toSet());
    String regionName = getRegionName(isPartitionedRegion);
    // Put entries
    client.invoke(() -> doPutsFrom(regionName, 0, entries));

    // Check that entries are put in the region
    for (VM member : serversInA) {
      member.invoke(() -> validateRegionSize(regionName, entries));
    }

    // Create senders and receivers
    createSender(isParallelGatewaySender, serversInA, senderIdInA, 2);
    createSender(isParallelGatewaySender, serversInB, senderIdInB, 3);
    createReceivers(serversInB, serverInC);

    // Let maxRate be a 5th of the number of entries in order to have the
    // command running for about 5 seconds so that there is time
    // for the random operations to be done at the same time
    // as the command is running.
    // The rate is divided by the number of servers in case the sender is parallel
    int maxRate = (entries / 5) / (isParallelGatewaySender ? serversInA.size() : 1);

    // Execute wan-copy region command
    Callable<CommandResultAssert> wanCopyCommandCallable = () -> {
      String command = new CommandStringBuilder(WAN_COPY_REGION)
          .addOption(WAN_COPY_REGION__REGION, regionName)
          .addOption(WAN_COPY_REGION__SENDERID, senderIdInA)
          .addOption(WAN_COPY_REGION__BATCHSIZE, String.valueOf(wanCopyRegionBatchSize))
          .addOption(WAN_COPY_REGION__MAXRATE, String.valueOf(maxRate))
          .getCommandString();
      try {
        gfsh.connectAndVerify(senderLocatorPort, GfshCommandRule.PortType.locator);
      } catch (Exception e) {
        errorCollector.addError(e);
      }
      return gfsh.executeAndAssertThat(command);
    };

    Future<CommandResultAssert> wanCopyCommandFuture =
        executorServiceRule.submit(wanCopyCommandCallable);

    // Wait for the wan-copy command to start
    waitForWanCopyRegionCommandToStart(isParallelGatewaySender, isPartitionedRegion, serversInA);

    // While the command is running, send some random operations over the same keys
    AsyncInvocation<Void> asyncOps1 =
        client.invokeAsync(() -> sendRandomOpsFromClient(regionName, keySet, 3));
    AsyncInvocation<Void> asyncOps2 =
