[GitHub] [geode] kirklund commented on a change in pull request #6833: GEODE-9369: Command to copy region entries from a WAN site to another

[GitBox <gi...@apache.org>]

kirklund commented on a change in pull request #6833:
URL: https://github.com/apache/geode/pull/6833#discussion_r720551660



##########
File path: geode-wan/src/distributedTest/java/org/apache/geode/cache/wan/internal/cli/commands/WanCopyRegionCommandDUnitTest.java
##########
@@ -0,0 +1,1267 @@
+/*
+ * 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__MSG__ALREADY__RUNNING__COMMAND;
+import static org.apache.geode.cache.wan.internal.cli.commands.WanCopyRegionCommand.WAN_COPY_REGION__MSG__CANCELED__BEFORE__HAVING__COPIED;
+import static org.apache.geode.cache.wan.internal.cli.commands.WanCopyRegionCommand.WAN_COPY_REGION__MSG__COPIED__ENTRIES;
+import static org.apache.geode.cache.wan.internal.cli.commands.WanCopyRegionCommand.WAN_COPY_REGION__MSG__EXECUTION__CANCELED;
+import static org.apache.geode.cache.wan.internal.cli.commands.WanCopyRegionCommand.WAN_COPY_REGION__MSG__NO__RUNNING__COMMAND;
+import static org.apache.geode.cache.wan.internal.cli.commands.WanCopyRegionCommand.WAN_COPY_REGION__MSG__REGION__NOT__FOUND;
+import static org.apache.geode.cache.wan.internal.cli.commands.WanCopyRegionCommand.WAN_COPY_REGION__MSG__SENDER__NOT__FOUND;
+import static org.apache.geode.cache.wan.internal.cli.commands.WanCopyRegionCommand.WAN_COPY_REGION__MSG__SENDER__SERIAL__AND__NOT__PRIMARY;
+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.test.awaitility.GeodeAwaitility.await;
+import static org.apache.geode.test.dunit.IgnoredException.addIgnoredException;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.Assert.assertNotNull;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.Future;
+import java.util.stream.Collectors;
+import java.util.stream.LongStream;
+
+import com.google.common.collect.ImmutableList;
+import junitparams.JUnitParamsRunner;
+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;
+
+@RunWith(JUnitParamsRunner.class)
+@Category({WanTest.class})
+public class WanCopyRegionCommandDUnitTest extends WANTestBase {
+
+  protected static VM vm8;
+
+  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);
+  }
+
+  @Test
+  public void testUnsuccessfulExecution_RegionNotFound() throws Exception {
+    List<VM> serversInA = Arrays.asList(vm5, vm6, vm7);
+    VM serverInB = vm3;
+    VM serverInC = vm4;
+    VM client = vm8;
+    String senderIdInA = "B";
+    String senderIdInB = "C";
+
+    int senderLocatorPort = create3WanSitesAndClient(true, vm0,
+        vm1, vm2, serversInA, serverInB, serverInC, client,
+        senderIdInA, senderIdInB);
+
+    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));
+    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);
+    VM serverInB = vm3;
+    VM serverInC = vm4;
+    VM client = vm8;
+    String senderIdInA = "B";
+    String senderIdInB = "C";
+
+    int senderLocatorPort = create3WanSitesAndClient(true, vm0,
+        vm1, vm2, serversInA, serverInB, serverInC, client,
+        senderIdInA, senderIdInB);
+
+    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));
+    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);
+    VM serverInB = vm3;
+    VM serverInC = vm4;
+    VM client = vm8;
+    String senderIdInA = "B";
+    String senderIdInB = "C";
+
+    int senderLocatorPort = create3WanSitesAndClient(isPartitionedRegion, vm0,
+        vm1, vm2, serversInA, serverInB, serverInC, client,
+        senderIdInA, senderIdInB);
+
+    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
+    serverInB.invoke(() -> destroyRegion(regionName));
+
+    // Create senders and receivers with replication as follows: "A" -> "B" -> "C"
+    createSenders(isParallelGatewaySender, serversInA, serverInB,
+        senderIdInA, senderIdInB);
+    createReceivers(serverInB, 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));
+      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);
+    }
+  }
+
+  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);
+      createSenders(useParallel, serversInA, null, senderIdInA, null);
+    } else {
+      // Senders will connect to receiver in server1InB
+      server1InB.invoke(WANTestBase::createReceiver);
+      createSenders(useParallel, serversInA, null, senderIdInA, null);
+      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, wanCopyCommandFuture);
+    }
+  }
+
+  @Test
+  @Parameters({"false, false", "false, true", "true, true"})
+  public void testRegionDestroyedDuringExecution(boolean isParallelGatewaySender,
+      boolean isPartitionedRegion)
+      throws Exception {
+    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 = ImmutableList.of(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);
+    createSenders(isParallelGatewaySender, serversInA, null, senderIdInA, null);
+
+    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);
+    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;
+    if (isPartitionedRegion && !isParallelGatewaySender) {
+      regionDestroyedError = new Condition<>(
+          s -> (s.startsWith(
+              "Execution failed. Error: org.apache.geode.cache.RegionDestroyedException: ")),
+          "execution error");
+    } else {
+      regionDestroyedError = new Condition<>(
+          s -> (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);
+      createSenders(useParallel, serversInA, null, senderIdInA, null);
+    } else {
+      // Senders will connect to receiver in server1InB
+      server1InB.invoke(WANTestBase::createReceiver);
+      createSenders(useParallel, serversInA, null, senderIdInA, null);
+      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));
+      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, false", "false, false"})
+  public void testSuccessfulExecution(boolean isPartitionedRegion,
+      boolean isParallelGatewaySender) throws Exception {
+    List<VM> serversInA = Arrays.asList(vm5, vm6, vm7);
+    VM serverInB = vm3;
+    VM serverInC = vm4;
+    VM client = vm8;
+    String senderIdInA = "B";
+    String senderIdInB = "C";
+
+    int senderLocatorPort = create3WanSitesAndClient(isPartitionedRegion, vm0,
+        vm1, vm2, serversInA, serverInB, serverInC, client,
+        senderIdInA, senderIdInB);
+
+    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"
+    serverInB.invoke(() -> validateRegionSize(regionName, 0));
+    serverInC.invoke(() -> validateRegionSize(regionName, 0));
+
+    // Create senders and receivers with replication as follows: "A" -> "B" -> "C"
+    createSenders(isParallelGatewaySender, serversInA, serverInB,
+        senderIdInA, senderIdInB);
+    createReceivers(serverInB, serverInC);
+
+    // Check that entries are not copied to "B" nor "C"
+    serverInB.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));
+    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, 100);
+      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"
+    serverInB.invoke(() -> validateRegionSize(regionName, entries));
+
+    // Check that the region's data is the same in sites "A" and "B"
+    checkEqualRegionData(regionName, serversInA.get(0), serverInB);
+
+    // Check that wanCopyRegionBatchSize is correctly used by the command
+    long 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);
+    VM serverInB = vm3;
+    VM serverInC = vm4;
+    VM client = vm8;
+    String senderIdInA = "B";
+    String senderIdInB = "C";
+
+    int senderLocatorPort = create3WanSitesAndClient(isPartitionedRegion, vm0,
+        vm1, vm2, serversInA, serverInB, serverInC, client,
+        senderIdInA, senderIdInB);
+
+    int wanCopyRegionBatchSize = 20;
+    int entries = 1000;
+    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 with replication as follows: "A" -> "B" -> "C"
+    createSenders(isParallelGatewaySender, serversInA, serverInB,
+        senderIdInA, senderIdInB);
+    createReceivers(serverInB, serverInC);
+
+    // Execute wan-copy region command
+    gfsh.connectAndVerify(senderLocatorPort, GfshCommandRule.PortType.locator);
+    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))
+        .getCommandString();
+
+    // While the command is running, send some random operations over the same keys
+    AsyncInvocation<Object> asyncOps1 =
+        client.invokeAsync(() -> sendRandomOpsFromClient(regionName, keySet, 10));
+    AsyncInvocation<Object> asyncOps2 =
+        client.invokeAsync(() -> sendRandomOpsFromClient(regionName, keySet, 10));
+    AsyncInvocation<Object> asyncOps3 =
+        client.invokeAsync(() -> sendRandomOpsFromClient(regionName, keySet, 10));

Review comment:
       Since you don't actually need a return from these (you just call `await()` down below), you should just use the `Void` type: `AsyncInvocation<Void>`.

##########
File path: geode-wan/src/test/java/org/apache/geode/management/internal/cli/functions/WanCopyRegionFunctionDelegateTest.java
##########
@@ -0,0 +1,526 @@
+/*
+ * 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.management.internal.cli.functions;
+
+import static org.apache.geode.util.internal.UncheckedUtils.uncheckedCast;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.time.Clock;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+
+import org.apache.geode.cache.Region;
+import org.apache.geode.cache.RegionAttributes;
+import org.apache.geode.cache.client.NoAvailableServersException;
+import org.apache.geode.cache.client.ServerConnectivityException;
+import org.apache.geode.cache.client.internal.Connection;
+import org.apache.geode.cache.client.internal.PoolImpl;
+import org.apache.geode.cache.client.internal.pooling.ConnectionDestroyedException;
+import org.apache.geode.cache.client.internal.pooling.PooledConnection;
+import org.apache.geode.cache.wan.GatewayQueueEvent;
+import org.apache.geode.cache.wan.GatewaySender;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.cache.InternalRegion;
+import org.apache.geode.internal.cache.wan.AbstractGatewaySender;
+import org.apache.geode.internal.cache.wan.AbstractGatewaySenderEventProcessor;
+import org.apache.geode.internal.cache.wan.BatchException70;
+import org.apache.geode.internal.cache.wan.GatewaySenderEventDispatcher;
+import org.apache.geode.internal.serialization.KnownVersion;
+import org.apache.geode.management.internal.cli.functions.WanCopyRegionFunctionDelegate.EventCreator;
+import org.apache.geode.management.internal.functions.CliFunctionResult;
+
+public class WanCopyRegionFunctionDelegateTest {
+
+  private WanCopyRegionFunctionDelegate function;
+  private long startTime;
+  private final int entries = 25;
+  private Clock clockMock;
+  private WanCopyRegionFunctionDelegate.ThreadSleeper threadSleeperMock;
+  private InternalCache internalCacheMock;
+  private GatewaySender gatewaySenderMock;
+  private PoolImpl poolMock;
+  private Connection connectionMock;
+  private GatewaySenderEventDispatcher dispatcherMock;
+
+  private final Region<?, ?> regionMock =
+      uncheckedCast(mock(InternalRegion.class));
+
+  private EventCreator eventCreatorMock;
+
+  @Before
+  public void setUp() throws InterruptedException {

Review comment:
       [comment] Some of these big methods are easier to read if you group all of the mocks together at the top, and then all of the stubbings (when statements) down lower.

##########
File path: geode-wan/src/distributedTest/java/org/apache/geode/cache/wan/internal/cli/commands/WanCopyRegionCommandDUnitTest.java
##########
@@ -0,0 +1,1267 @@
+/*
+ * 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__MSG__ALREADY__RUNNING__COMMAND;
+import static org.apache.geode.cache.wan.internal.cli.commands.WanCopyRegionCommand.WAN_COPY_REGION__MSG__CANCELED__BEFORE__HAVING__COPIED;
+import static org.apache.geode.cache.wan.internal.cli.commands.WanCopyRegionCommand.WAN_COPY_REGION__MSG__COPIED__ENTRIES;
+import static org.apache.geode.cache.wan.internal.cli.commands.WanCopyRegionCommand.WAN_COPY_REGION__MSG__EXECUTION__CANCELED;
+import static org.apache.geode.cache.wan.internal.cli.commands.WanCopyRegionCommand.WAN_COPY_REGION__MSG__NO__RUNNING__COMMAND;
+import static org.apache.geode.cache.wan.internal.cli.commands.WanCopyRegionCommand.WAN_COPY_REGION__MSG__REGION__NOT__FOUND;
+import static org.apache.geode.cache.wan.internal.cli.commands.WanCopyRegionCommand.WAN_COPY_REGION__MSG__SENDER__NOT__FOUND;
+import static org.apache.geode.cache.wan.internal.cli.commands.WanCopyRegionCommand.WAN_COPY_REGION__MSG__SENDER__SERIAL__AND__NOT__PRIMARY;
+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.test.awaitility.GeodeAwaitility.await;
+import static org.apache.geode.test.dunit.IgnoredException.addIgnoredException;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.Assert.assertNotNull;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.Future;
+import java.util.stream.Collectors;
+import java.util.stream.LongStream;
+
+import com.google.common.collect.ImmutableList;
+import junitparams.JUnitParamsRunner;
+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;
+
+@RunWith(JUnitParamsRunner.class)
+@Category({WanTest.class})
+public class WanCopyRegionCommandDUnitTest extends WANTestBase {
+
+  protected static VM vm8;
+
+  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);
+  }
+
+  @Test
+  public void testUnsuccessfulExecution_RegionNotFound() throws Exception {
+    List<VM> serversInA = Arrays.asList(vm5, vm6, vm7);
+    VM serverInB = vm3;
+    VM serverInC = vm4;
+    VM client = vm8;
+    String senderIdInA = "B";
+    String senderIdInB = "C";
+
+    int senderLocatorPort = create3WanSitesAndClient(true, vm0,
+        vm1, vm2, serversInA, serverInB, serverInC, client,
+        senderIdInA, senderIdInB);
+
+    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));
+    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);
+    VM serverInB = vm3;
+    VM serverInC = vm4;
+    VM client = vm8;
+    String senderIdInA = "B";
+    String senderIdInB = "C";
+
+    int senderLocatorPort = create3WanSitesAndClient(true, vm0,
+        vm1, vm2, serversInA, serverInB, serverInC, client,
+        senderIdInA, senderIdInB);
+
+    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));
+    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);
+    VM serverInB = vm3;
+    VM serverInC = vm4;
+    VM client = vm8;
+    String senderIdInA = "B";
+    String senderIdInB = "C";
+
+    int senderLocatorPort = create3WanSitesAndClient(isPartitionedRegion, vm0,
+        vm1, vm2, serversInA, serverInB, serverInC, client,
+        senderIdInA, senderIdInB);
+
+    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
+    serverInB.invoke(() -> destroyRegion(regionName));
+
+    // Create senders and receivers with replication as follows: "A" -> "B" -> "C"
+    createSenders(isParallelGatewaySender, serversInA, serverInB,
+        senderIdInA, senderIdInB);
+    createReceivers(serverInB, 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));
+      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);
+    }

Review comment:
       [comment] Not a request for changes, but I generally try to format these chained APIs with linefeeds like:
   ```
   command
       .hasTableSection(ResultModel.MEMBER_STATUS_SECTION)
       .hasColumn("Message")
       .asList()
       .haveExactly(1, exceptionError)
       .haveExactly(2, senderNotPrimary);
   ```
   I'm not sure if it's easier to read for everyone, but it is for me.

##########
File path: geode-wan/src/test/java/org/apache/geode/cache/wan/internal/WanCopyRegionFunctionServiceTest.java
##########
@@ -0,0 +1,184 @@
+/*
+ * 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;
+
+import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy;
+import static org.mockito.Mockito.mock;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.management.internal.functions.CliFunctionResult;
+
+public class WanCopyRegionFunctionServiceTest {
+
+  private WanCopyRegionFunctionService service;
+  InternalCache cache = mock(InternalCache.class);
+
+  @Before
+  public void setUp() throws Exception {
+    service = new WanCopyRegionFunctionService();
+    service.init(cache);
+  }
+
+  @Test
+  public void severalExecuteWithSameRegionAndSenderNotAllowed()
+      throws InterruptedException, WanCopyRegionFunctionServiceAlreadyRunningException {
+    CountDownLatch latch = new CountDownLatch(1);
+    Callable<CliFunctionResult> firstExecution = () -> {
+      latch.await();
+      return null;
+    };
+
+    String regionName = "myRegion";
+    String senderId = "mySender";
+    CompletableFuture
+        .supplyAsync(() -> {
+          try {
+            return service.execute(firstExecution, regionName, senderId);
+          } catch (Exception e) {
+            return null;

Review comment:
       [comment] Returning null is probably ok. Or you could use an `ErrorCollector` JUnit rule here as well.

##########
File path: geode-wan/src/test/java/org/apache/geode/management/internal/cli/functions/WanCopyRegionFunctionDelegateTest.java
##########
@@ -0,0 +1,526 @@
+/*
+ * 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.management.internal.cli.functions;
+
+import static org.apache.geode.util.internal.UncheckedUtils.uncheckedCast;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.time.Clock;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+
+import org.apache.geode.cache.Region;
+import org.apache.geode.cache.RegionAttributes;
+import org.apache.geode.cache.client.NoAvailableServersException;
+import org.apache.geode.cache.client.ServerConnectivityException;
+import org.apache.geode.cache.client.internal.Connection;
+import org.apache.geode.cache.client.internal.PoolImpl;
+import org.apache.geode.cache.client.internal.pooling.ConnectionDestroyedException;
+import org.apache.geode.cache.client.internal.pooling.PooledConnection;
+import org.apache.geode.cache.wan.GatewayQueueEvent;
+import org.apache.geode.cache.wan.GatewaySender;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.cache.InternalRegion;
+import org.apache.geode.internal.cache.wan.AbstractGatewaySender;
+import org.apache.geode.internal.cache.wan.AbstractGatewaySenderEventProcessor;
+import org.apache.geode.internal.cache.wan.BatchException70;
+import org.apache.geode.internal.cache.wan.GatewaySenderEventDispatcher;
+import org.apache.geode.internal.serialization.KnownVersion;
+import org.apache.geode.management.internal.cli.functions.WanCopyRegionFunctionDelegate.EventCreator;
+import org.apache.geode.management.internal.functions.CliFunctionResult;
+
+public class WanCopyRegionFunctionDelegateTest {
+
+  private WanCopyRegionFunctionDelegate function;
+  private long startTime;
+  private final int entries = 25;
+  private Clock clockMock;
+  private WanCopyRegionFunctionDelegate.ThreadSleeper threadSleeperMock;
+  private InternalCache internalCacheMock;
+  private GatewaySender gatewaySenderMock;
+  private PoolImpl poolMock;
+  private Connection connectionMock;
+  private GatewaySenderEventDispatcher dispatcherMock;
+
+  private final Region<?, ?> regionMock =
+      uncheckedCast(mock(InternalRegion.class));
+
+  private EventCreator eventCreatorMock;
+
+  @Before
+  public void setUp() throws InterruptedException {
+    startTime = System.currentTimeMillis();
+
+    clockMock = mock(Clock.class);
+
+    threadSleeperMock = mock(WanCopyRegionFunctionDelegate.ThreadSleeper.class);
+    doNothing().when(threadSleeperMock).sleep(anyLong());
+
+    gatewaySenderMock = mock(AbstractGatewaySender.class);
+    when(gatewaySenderMock.getId()).thenReturn("mySender");
+
+    poolMock = mock(PoolImpl.class);
+
+    connectionMock = mock(PooledConnection.class);
+    when(connectionMock.getWanSiteVersion()).thenReturn(KnownVersion.GEODE_1_15_0.ordinal());
+
+    dispatcherMock = mock(GatewaySenderEventDispatcher.class);
+
+    eventCreatorMock = mock(EventCreator.class);
+    when(eventCreatorMock.createGatewaySenderEvent(any(), any(), any(), any()))
+        .thenReturn(uncheckedCast(mock(GatewayQueueEvent.class)));
+
+    function = new WanCopyRegionFunctionDelegate(clockMock, threadSleeperMock, eventCreatorMock);
+
+    AbstractGatewaySenderEventProcessor eventProcessorMock =
+        mock(AbstractGatewaySenderEventProcessor.class);
+    when(eventProcessorMock.getDispatcher()).thenReturn(dispatcherMock);
+    when(((AbstractGatewaySender) gatewaySenderMock).getEventProcessor())
+        .thenReturn(eventProcessorMock);
+
+    RegionAttributes<?, ?> attributesMock = mock(RegionAttributes.class);
+    Set<?> idsMock = mock(Set.class);
+    when(idsMock.contains(anyString())).thenReturn(true);
+    when(attributesMock.getGatewaySenderIds()).thenReturn(uncheckedCast(idsMock));
+    when(regionMock.getAttributes()).thenReturn(uncheckedCast(attributesMock));
+
+    internalCacheMock = mock(InternalCache.class);
+    when(internalCacheMock.getRegion(any())).thenReturn(uncheckedCast(regionMock));
+  }
+
+  @Test
+  public void doPostSendBatchActions_DoNotSleepIfMaxRateIsZero()
+      throws InterruptedException {
+    when(internalCacheMock.getRegion(any())).thenReturn(null);
+
+    function.wanCopyRegion(internalCacheMock, "member1", regionMock, gatewaySenderMock, 0L, 10);
+    verify(threadSleeperMock, never()).sleep(anyLong());
+  }
+
+  @Test
+  public void doPostSendBatchActions_SleepIfMaxRateIsNotZero()
+      throws InterruptedException, BatchException70 {
+    Object[] options = new Object[] {"myRegion", "mySender", false, 1L, 2};
+    WanCopyRegionFunctionDelegate.ThreadSleeper sleeperMock =
+        mock(WanCopyRegionFunctionDelegate.ThreadSleeper.class);
+    doNothing().when(sleeperMock).sleep(anyLong());
+
+    executeWanCopyRegionFunction(options, sleeperMock);
+    ArgumentCaptor<Long> captor = ArgumentCaptor.forClass(Long.class);
+    verify(sleeperMock, times(1)).sleep(captor.capture());
+    assertThat(captor.getValue()).isGreaterThan(0).isLessThanOrEqualTo(2000);

Review comment:
       [comment] Another way to organize these tests for (hopefully) easier reading is to group it in arrange/act/assert sections:
   ```
       // arrange
       Object[] options = new Object[] {"myRegion", "mySender", false, 1L, 2};
       WanCopyRegionFunctionDelegate.ThreadSleeper sleeperMock =
           mock(WanCopyRegionFunctionDelegate.ThreadSleeper.class);
       doNothing().when(sleeperMock).sleep(anyLong());
   
       // act
       executeWanCopyRegionFunction(options, sleeperMock);
   
       // assert
       ArgumentCaptor<Long> captor = ArgumentCaptor.forClass(Long.class);
       verify(sleeperMock, times(1)).sleep(captor.capture());
       assertThat(captor.getValue()).isGreaterThan(0).isLessThanOrEqualTo(2000);
   ```

##########
File path: geode-wan/src/test/java/org/apache/geode/management/internal/cli/functions/WanCopyRegionFunctionDelegateTest.java
##########
@@ -0,0 +1,526 @@
+/*
+ * 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.management.internal.cli.functions;
+
+import static org.apache.geode.util.internal.UncheckedUtils.uncheckedCast;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.time.Clock;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+
+import org.apache.geode.cache.Region;
+import org.apache.geode.cache.RegionAttributes;
+import org.apache.geode.cache.client.NoAvailableServersException;
+import org.apache.geode.cache.client.ServerConnectivityException;
+import org.apache.geode.cache.client.internal.Connection;
+import org.apache.geode.cache.client.internal.PoolImpl;
+import org.apache.geode.cache.client.internal.pooling.ConnectionDestroyedException;
+import org.apache.geode.cache.client.internal.pooling.PooledConnection;
+import org.apache.geode.cache.wan.GatewayQueueEvent;
+import org.apache.geode.cache.wan.GatewaySender;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.cache.InternalRegion;
+import org.apache.geode.internal.cache.wan.AbstractGatewaySender;
+import org.apache.geode.internal.cache.wan.AbstractGatewaySenderEventProcessor;
+import org.apache.geode.internal.cache.wan.BatchException70;
+import org.apache.geode.internal.cache.wan.GatewaySenderEventDispatcher;
+import org.apache.geode.internal.serialization.KnownVersion;
+import org.apache.geode.management.internal.cli.functions.WanCopyRegionFunctionDelegate.EventCreator;
+import org.apache.geode.management.internal.functions.CliFunctionResult;
+
+public class WanCopyRegionFunctionDelegateTest {
+
+  private WanCopyRegionFunctionDelegate function;
+  private long startTime;
+  private final int entries = 25;
+  private Clock clockMock;
+  private WanCopyRegionFunctionDelegate.ThreadSleeper threadSleeperMock;
+  private InternalCache internalCacheMock;
+  private GatewaySender gatewaySenderMock;
+  private PoolImpl poolMock;
+  private Connection connectionMock;
+  private GatewaySenderEventDispatcher dispatcherMock;
+
+  private final Region<?, ?> regionMock =
+      uncheckedCast(mock(InternalRegion.class));
+
+  private EventCreator eventCreatorMock;
+
+  @Before
+  public void setUp() throws InterruptedException {
+    startTime = System.currentTimeMillis();
+
+    clockMock = mock(Clock.class);
+
+    threadSleeperMock = mock(WanCopyRegionFunctionDelegate.ThreadSleeper.class);
+    doNothing().when(threadSleeperMock).sleep(anyLong());
+
+    gatewaySenderMock = mock(AbstractGatewaySender.class);
+    when(gatewaySenderMock.getId()).thenReturn("mySender");
+
+    poolMock = mock(PoolImpl.class);
+
+    connectionMock = mock(PooledConnection.class);
+    when(connectionMock.getWanSiteVersion()).thenReturn(KnownVersion.GEODE_1_15_0.ordinal());
+
+    dispatcherMock = mock(GatewaySenderEventDispatcher.class);
+
+    eventCreatorMock = mock(EventCreator.class);
+    when(eventCreatorMock.createGatewaySenderEvent(any(), any(), any(), any()))
+        .thenReturn(uncheckedCast(mock(GatewayQueueEvent.class)));
+
+    function = new WanCopyRegionFunctionDelegate(clockMock, threadSleeperMock, eventCreatorMock);
+
+    AbstractGatewaySenderEventProcessor eventProcessorMock =
+        mock(AbstractGatewaySenderEventProcessor.class);
+    when(eventProcessorMock.getDispatcher()).thenReturn(dispatcherMock);
+    when(((AbstractGatewaySender) gatewaySenderMock).getEventProcessor())
+        .thenReturn(eventProcessorMock);
+
+    RegionAttributes<?, ?> attributesMock = mock(RegionAttributes.class);
+    Set<?> idsMock = mock(Set.class);
+    when(idsMock.contains(anyString())).thenReturn(true);
+    when(attributesMock.getGatewaySenderIds()).thenReturn(uncheckedCast(idsMock));
+    when(regionMock.getAttributes()).thenReturn(uncheckedCast(attributesMock));
+
+    internalCacheMock = mock(InternalCache.class);
+    when(internalCacheMock.getRegion(any())).thenReturn(uncheckedCast(regionMock));
+  }
+
+  @Test
+  public void doPostSendBatchActions_DoNotSleepIfMaxRateIsZero()
+      throws InterruptedException {
+    when(internalCacheMock.getRegion(any())).thenReturn(null);
+
+    function.wanCopyRegion(internalCacheMock, "member1", regionMock, gatewaySenderMock, 0L, 10);
+    verify(threadSleeperMock, never()).sleep(anyLong());
+  }
+
+  @Test
+  public void doPostSendBatchActions_SleepIfMaxRateIsNotZero()
+      throws InterruptedException, BatchException70 {
+    Object[] options = new Object[] {"myRegion", "mySender", false, 1L, 2};
+    WanCopyRegionFunctionDelegate.ThreadSleeper sleeperMock =
+        mock(WanCopyRegionFunctionDelegate.ThreadSleeper.class);
+    doNothing().when(sleeperMock).sleep(anyLong());
+
+    executeWanCopyRegionFunction(options, sleeperMock);
+    ArgumentCaptor<Long> captor = ArgumentCaptor.forClass(Long.class);
+    verify(sleeperMock, times(1)).sleep(captor.capture());
+    assertThat(captor.getValue()).isGreaterThan(0).isLessThanOrEqualTo(2000);
+  }
+
+  @Test
+  public void doPostSendBatchActions_ThrowInterruptedIfInterruptedTimeToSleepNotZero()
+      throws InterruptedException {
+    long maxRate = 1;
+    long elapsedTime = 20L;
+    when(clockMock.millis()).thenReturn(startTime + elapsedTime);
+
+    // Sleep so that the sleep can be interrupted
+    doAnswer(invocation -> {
+      Thread.sleep(1);
+      return null;
+    }).when(threadSleeperMock).sleep(anyLong());
+
+    Thread.currentThread().interrupt();
+
+    assertThatThrownBy(
+        () -> function.doPostSendBatchActions(startTime, entries, maxRate))
+            .isInstanceOf(InterruptedException.class);
+  }
+
+  @Test
+  public void doPostSendBatchActions_ThrowInterruptedIfInterruptedTimeToSleepIsZero() {
+    long maxRate = 0;
+    Thread.currentThread().interrupt();
+    assertThatThrownBy(
+        () -> function.doPostSendBatchActions(startTime, entries, maxRate))
+            .isInstanceOf(InterruptedException.class);
+  }
+
+  @Test
+  public void wanCopyRegion_verifyErrorWhenRemoteSiteDoesNotSupportCommand()
+      throws InterruptedException {
+    when(((AbstractGatewaySender) gatewaySenderMock).getProxy()).thenReturn(poolMock);
+    PooledConnection oldWanSiteConn = mock(PooledConnection.class);
+    when(oldWanSiteConn.getWanSiteVersion()).thenReturn(KnownVersion.GEODE_1_14_0.ordinal());
+    when(poolMock.acquireConnection()).thenReturn(oldWanSiteConn);
+    Set<Region.Entry<Object, Object>> entries = new HashSet<>();
+    entries.add(uncheckedCast(mock(Region.Entry.class)));
+    when(regionMock.entrySet()).thenReturn(uncheckedCast(entries));
+
+    CliFunctionResult result =
+        function.wanCopyRegion(internalCacheMock, "member1", regionMock, gatewaySenderMock, 1, 10);
+    assertThat(result.getStatus()).isEqualTo(CliFunctionResult.StatusState.ERROR.toString());
+    assertThat(result.getStatusMessage())
+        .isEqualTo("Command not supported at remote site.");
+  }
+
+  @Test
+  public void wanCopyRegion_verifyErrorWhenNoPoolAvailableAndEntriesInRegion()
+      throws InterruptedException {
+    when(((AbstractGatewaySender) gatewaySenderMock).getProxy()).thenReturn(null);
+    when(poolMock.acquireConnection()).thenThrow(NoAvailableServersException.class)
+        .thenThrow(NoAvailableServersException.class);
+    Set<Region.Entry<Object, Object>> entries = new HashSet<>();
+    entries.add(uncheckedCast(mock(Region.Entry.class)));
+    when(regionMock.entrySet()).thenReturn(uncheckedCast(entries));
+
+    CliFunctionResult result =
+        function.wanCopyRegion(internalCacheMock, "member1", regionMock, gatewaySenderMock, 1, 10);
+    assertThat(result.getStatus()).isEqualTo(CliFunctionResult.StatusState.ERROR.toString());
+    assertThat(result.getStatusMessage())
+        .isEqualTo("No connection pool available to receiver");
+  }
+
+  @Test
+  public void wanCopyRegion_verifySuccessWhenNoPoolAvailableAndNoEntriesInRegion()
+      throws InterruptedException {
+    when(((AbstractGatewaySender) gatewaySenderMock).getProxy()).thenReturn(null);
+    when(poolMock.acquireConnection()).thenThrow(NoAvailableServersException.class)
+        .thenThrow(NoAvailableServersException.class);
+    when(regionMock.entrySet()).thenReturn(new HashSet<>());
+
+    CliFunctionResult result =
+        function.wanCopyRegion(internalCacheMock, "member1", regionMock, gatewaySenderMock, 1, 10);
+    assertThat(result.getStatus()).isEqualTo(CliFunctionResult.StatusState.OK.toString());
+    assertThat(result.getStatusMessage())
+        .isEqualTo("Entries copied: 0");
+  }
+
+  @Test
+  public void wanCopyRegion_verifyErrorWhenNoConnectionAvailableAtStartAndEntriesInRegion() {
+    when(((AbstractGatewaySender) gatewaySenderMock).getProxy()).thenReturn(poolMock);
+    when(poolMock.acquireConnection()).thenThrow(NoAvailableServersException.class)
+        .thenThrow(NoAvailableServersException.class);
+    Set<Region.Entry<Object, Object>> entries = new HashSet<>();
+    entries.add(uncheckedCast(mock(Region.Entry.class)));
+    when(regionMock.entrySet()).thenReturn(uncheckedCast(entries));
+
+    assertThatThrownBy(
+        () -> function.wanCopyRegion(internalCacheMock, "member1", regionMock, gatewaySenderMock, 1,
+            10))
+                .isInstanceOf(NoAvailableServersException.class);

Review comment:
       [comment] If you decide you like arrange/act/assert then you can use this syntax if you want:
   ```
   // arrange
   ...
   
   // act
   Throwable thrown = catchThrowable(() -> {
     function.wanCopyRegion(internalCacheMock, "member1", regionMock, gatewaySenderMock, 1, 10);
   });
   
   // assert
   assertThat(thrown).isInstanceOf(NoAvailableServersException.class);
   ```

##########
File path: geode-wan/src/test/java/org/apache/geode/cache/wan/internal/WanCopyRegionFunctionServiceTest.java
##########
@@ -0,0 +1,184 @@
+/*
+ * 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;
+
+import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy;
+import static org.mockito.Mockito.mock;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.management.internal.functions.CliFunctionResult;
+
+public class WanCopyRegionFunctionServiceTest {
+
+  private WanCopyRegionFunctionService service;
+  InternalCache cache = mock(InternalCache.class);

Review comment:
       `private InternalCache cache = mock(InternalCache.class);`

##########
File path: geode-wan/src/distributedTest/java/org/apache/geode/cache/wan/internal/cli/commands/WanCopyRegionCommandDUnitTest.java
##########
@@ -0,0 +1,1267 @@
+/*
+ * 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__MSG__ALREADY__RUNNING__COMMAND;
+import static org.apache.geode.cache.wan.internal.cli.commands.WanCopyRegionCommand.WAN_COPY_REGION__MSG__CANCELED__BEFORE__HAVING__COPIED;
+import static org.apache.geode.cache.wan.internal.cli.commands.WanCopyRegionCommand.WAN_COPY_REGION__MSG__COPIED__ENTRIES;
+import static org.apache.geode.cache.wan.internal.cli.commands.WanCopyRegionCommand.WAN_COPY_REGION__MSG__EXECUTION__CANCELED;
+import static org.apache.geode.cache.wan.internal.cli.commands.WanCopyRegionCommand.WAN_COPY_REGION__MSG__NO__RUNNING__COMMAND;
+import static org.apache.geode.cache.wan.internal.cli.commands.WanCopyRegionCommand.WAN_COPY_REGION__MSG__REGION__NOT__FOUND;
+import static org.apache.geode.cache.wan.internal.cli.commands.WanCopyRegionCommand.WAN_COPY_REGION__MSG__SENDER__NOT__FOUND;
+import static org.apache.geode.cache.wan.internal.cli.commands.WanCopyRegionCommand.WAN_COPY_REGION__MSG__SENDER__SERIAL__AND__NOT__PRIMARY;
+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.test.awaitility.GeodeAwaitility.await;
+import static org.apache.geode.test.dunit.IgnoredException.addIgnoredException;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.Assert.assertNotNull;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.Future;
+import java.util.stream.Collectors;
+import java.util.stream.LongStream;
+
+import com.google.common.collect.ImmutableList;
+import junitparams.JUnitParamsRunner;
+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;
+
+@RunWith(JUnitParamsRunner.class)
+@Category({WanTest.class})
+public class WanCopyRegionCommandDUnitTest extends WANTestBase {
+
+  protected static VM vm8;
+
+  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);
+  }
+
+  @Test
+  public void testUnsuccessfulExecution_RegionNotFound() throws Exception {
+    List<VM> serversInA = Arrays.asList(vm5, vm6, vm7);
+    VM serverInB = vm3;
+    VM serverInC = vm4;
+    VM client = vm8;
+    String senderIdInA = "B";
+    String senderIdInB = "C";
+
+    int senderLocatorPort = create3WanSitesAndClient(true, vm0,
+        vm1, vm2, serversInA, serverInB, serverInC, client,
+        senderIdInA, senderIdInB);
+
+    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));
+    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);
+    VM serverInB = vm3;
+    VM serverInC = vm4;
+    VM client = vm8;
+    String senderIdInA = "B";
+    String senderIdInB = "C";
+
+    int senderLocatorPort = create3WanSitesAndClient(true, vm0,
+        vm1, vm2, serversInA, serverInB, serverInC, client,
+        senderIdInA, senderIdInB);
+
+    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));
+    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);
+    VM serverInB = vm3;
+    VM serverInC = vm4;
+    VM client = vm8;
+    String senderIdInA = "B";
+    String senderIdInB = "C";
+
+    int senderLocatorPort = create3WanSitesAndClient(isPartitionedRegion, vm0,
+        vm1, vm2, serversInA, serverInB, serverInC, client,
+        senderIdInA, senderIdInB);
+
+    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
+    serverInB.invoke(() -> destroyRegion(regionName));
+
+    // Create senders and receivers with replication as follows: "A" -> "B" -> "C"
+    createSenders(isParallelGatewaySender, serversInA, serverInB,
+        senderIdInA, senderIdInB);
+    createReceivers(serverInB, 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));
+      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);
+    }
+  }
+
+  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);
+      createSenders(useParallel, serversInA, null, senderIdInA, null);
+    } else {
+      // Senders will connect to receiver in server1InB
+      server1InB.invoke(WANTestBase::createReceiver);
+      createSenders(useParallel, serversInA, null, senderIdInA, null);
+      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, wanCopyCommandFuture);
+    }
+  }
+
+  @Test
+  @Parameters({"false, false", "false, true", "true, true"})
+  public void testRegionDestroyedDuringExecution(boolean isParallelGatewaySender,
+      boolean isPartitionedRegion)
+      throws Exception {
+    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 = ImmutableList.of(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);
+    createSenders(isParallelGatewaySender, serversInA, null, senderIdInA, null);
+
+    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);
+    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;
+    if (isPartitionedRegion && !isParallelGatewaySender) {
+      regionDestroyedError = new Condition<>(
+          s -> (s.startsWith(
+              "Execution failed. Error: org.apache.geode.cache.RegionDestroyedException: ")),
+          "execution error");
+    } else {
+      regionDestroyedError = new Condition<>(
+          s -> (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);
+      createSenders(useParallel, serversInA, null, senderIdInA, null);
+    } else {
+      // Senders will connect to receiver in server1InB
+      server1InB.invoke(WANTestBase::createReceiver);
+      createSenders(useParallel, serversInA, null, senderIdInA, null);
+      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));
+      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, false", "false, false"})
+  public void testSuccessfulExecution(boolean isPartitionedRegion,
+      boolean isParallelGatewaySender) throws Exception {
+    List<VM> serversInA = Arrays.asList(vm5, vm6, vm7);
+    VM serverInB = vm3;
+    VM serverInC = vm4;
+    VM client = vm8;
+    String senderIdInA = "B";
+    String senderIdInB = "C";
+
+    int senderLocatorPort = create3WanSitesAndClient(isPartitionedRegion, vm0,
+        vm1, vm2, serversInA, serverInB, serverInC, client,
+        senderIdInA, senderIdInB);
+
+    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"
+    serverInB.invoke(() -> validateRegionSize(regionName, 0));
+    serverInC.invoke(() -> validateRegionSize(regionName, 0));
+
+    // Create senders and receivers with replication as follows: "A" -> "B" -> "C"
+    createSenders(isParallelGatewaySender, serversInA, serverInB,
+        senderIdInA, senderIdInB);
+    createReceivers(serverInB, serverInC);
+
+    // Check that entries are not copied to "B" nor "C"
+    serverInB.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));
+    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, 100);
+      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"
+    serverInB.invoke(() -> validateRegionSize(regionName, entries));
+
+    // Check that the region's data is the same in sites "A" and "B"
+    checkEqualRegionData(regionName, serversInA.get(0), serverInB);
+
+    // Check that wanCopyRegionBatchSize is correctly used by the command
+    long 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);
+    VM serverInB = vm3;
+    VM serverInC = vm4;
+    VM client = vm8;
+    String senderIdInA = "B";
+    String senderIdInB = "C";
+
+    int senderLocatorPort = create3WanSitesAndClient(isPartitionedRegion, vm0,
+        vm1, vm2, serversInA, serverInB, serverInC, client,
+        senderIdInA, senderIdInB);
+
+    int wanCopyRegionBatchSize = 20;
+    int entries = 1000;
+    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 with replication as follows: "A" -> "B" -> "C"
+    createSenders(isParallelGatewaySender, serversInA, serverInB,
+        senderIdInA, senderIdInB);
+    createReceivers(serverInB, serverInC);
+
+    // Execute wan-copy region command
+    gfsh.connectAndVerify(senderLocatorPort, GfshCommandRule.PortType.locator);
+    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))
+        .getCommandString();
+
+    // While the command is running, send some random operations over the same keys
+    AsyncInvocation<Object> asyncOps1 =
+        client.invokeAsync(() -> sendRandomOpsFromClient(regionName, keySet, 10));
+    AsyncInvocation<Object> asyncOps2 =
+        client.invokeAsync(() -> sendRandomOpsFromClient(regionName, keySet, 10));
+    AsyncInvocation<Object> asyncOps3 =
+        client.invokeAsync(() -> sendRandomOpsFromClient(regionName, keySet, 10));
+
+    // Check command status and output
+    verifyStatusIsOk(gfsh.executeAndAssertThat(command));
+
+    // Wait for random operations to finish
+    asyncOps1.await();
+    asyncOps2.await();
+    asyncOps3.await();
+
+    // Wait for entries to be replicated (replication queues empty)
+    for (VM server : serversInA) {
+      server.invoke(() -> getSenderStats(senderIdInA, 0));
+    }
+
+    // Check that the region's data is the same in sites "A" and "B"
+    checkEqualRegionData(regionName, serversInA.get(0), serverInB);
+  }
+
+  /**
+   * Cancel is executed when no wan-copy region command is running.
+   */
+  @Test
+  @Parameters({"true, true", "true, false", "false, false"})
+  public void testUnsuccessfulCancelExecution(boolean isPartitionedRegion,
+      boolean isParallelGatewaySender) throws Exception {
+    List<VM> serversInA = Arrays.asList(vm5, vm6, vm7);
+    VM serverInB = vm3;
+    VM serverInC = vm4;
+    VM client = vm8;
+    String senderIdInA = "B";
+    String senderIdInB = "C";
+
+    int senderLocatorPort = create3WanSitesAndClient(isPartitionedRegion, vm0,
+        vm1, vm2, serversInA, serverInB, serverInC, client,
+        senderIdInA, senderIdInB);
+
+    String regionName = getRegionName(isPartitionedRegion);
+
+    // Create senders and receivers with replication as follows: "A" -> "B" -> "C"
+    createSenders(isParallelGatewaySender, serversInA, serverInB,
+        senderIdInA, senderIdInB);
+    createReceivers(serverInB, serverInC);
+
+    // Execute cancel wan-copy region command
+    gfsh.connectAndVerify(senderLocatorPort, GfshCommandRule.PortType.locator);
+    String cancelCommand = new CommandStringBuilder(WAN_COPY_REGION)
+        .addOption(WAN_COPY_REGION__REGION, regionName)
+        .addOption(WAN_COPY_REGION__SENDERID, senderIdInA)
+        .addOption(WAN_COPY_REGION__CANCEL)
+        .getCommandString();
+    CommandResultAssert cancelCommandResult =
+        verifyStatusIsError(gfsh.executeAndAssertThat(cancelCommand));
+    String msg1 = CliStrings.format(WAN_COPY_REGION__MSG__NO__RUNNING__COMMAND,
+        regionName, senderIdInA);
+    cancelCommandResult.hasTableSection(ResultModel.MEMBER_STATUS_SECTION).hasColumn("Message")
+        .containsExactlyInAnyOrder(msg1, msg1, msg1);
+  }
+
+  /**
+   * 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"
+   * in a different thread. The maxRate is set to a very low value so that there is
+   * time to cancel it before it finishes.
+   *
+   * The "wan-copy region" command with the cancel option
+   * is run from "A" site passing sender "B".
+   *
+   * It must be verified that the command is canceled.
+   * Also, the output of the command must show
+   * the number of entries copied before the command was canceled.
+   */
+  @Test
+  @Parameters({"true, true", "true, false", "false, false"})
+  public void testSuccessfulCancelExecution(boolean isPartitionedRegion,
+      boolean isParallelGatewaySender) throws Exception {
+    List<VM> serversInA = Arrays.asList(vm5, vm6, vm7);
+    VM serverInB = vm3;
+    VM serverInC = vm4;
+    VM client = vm8;
+    String senderIdInA = "B";
+    String senderIdInB = "C";
+
+    int senderLocatorPort = create3WanSitesAndClient(isPartitionedRegion, vm0,
+        vm1, vm2, serversInA, serverInB, serverInC, client,
+        senderIdInA, senderIdInB);
+
+    int wanCopyRegionBatchSize = 20;
+    int entries = 100;
+    String regionName = getRegionName(isPartitionedRegion);
+    // Put entries
+    client.invoke(() -> doPutsFrom(regionName, 0, entries));
+
+    // Create senders and receivers with replication as follows: "A" -> "B" -> "C"
+    createSenders(isParallelGatewaySender, serversInA, serverInB,
+        senderIdInA, senderIdInB);
+    createReceivers(serverInB, serverInC);
+
+    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(1))
+          .getCommandString();
+      try {
+        gfsh.connectAndVerify(senderLocatorPort, GfshCommandRule.PortType.locator);
+      } catch (Exception e) {
+        errorCollector.addError(e);
+      }
+      return gfsh.executeAndAssertThat(command).statusIsError();
+    };
+
+    Future<CommandResultAssert> wanCopyCommandFuture =
+        executorServiceRule.submit(wanCopyCommandCallable);
+
+    // Wait for the wan-copy command to start
+    waitForWanCopyRegionCommandToStart(isParallelGatewaySender, isPartitionedRegion, serversInA);
+
+    // Cancel wan-copy region command
+    GfshCommandRule gfshCancelCommand = new GfshCommandRule();
+    gfshCancelCommand.connectAndVerify(senderLocatorPort, GfshCommandRule.PortType.locator);
+    String cancelCommand = new CommandStringBuilder(WAN_COPY_REGION)
+        .addOption(WAN_COPY_REGION__REGION, regionName)
+        .addOption(WAN_COPY_REGION__SENDERID, senderIdInA)
+        .addOption(WAN_COPY_REGION__CANCEL)
+        .getCommandString();
+    CommandResultAssert cancelCommandResult =
+        gfshCancelCommand.executeAndAssertThat(cancelCommand);
+
+    if (isPartitionedRegion && isParallelGatewaySender) {
+      verifyStatusIsOk(cancelCommandResult);
+      cancelCommandResult.hasTableSection(ResultModel.MEMBER_STATUS_SECTION).hasColumn("Message")
+          .containsExactlyInAnyOrder(WAN_COPY_REGION__MSG__EXECUTION__CANCELED,
+              WAN_COPY_REGION__MSG__EXECUTION__CANCELED,
+              WAN_COPY_REGION__MSG__EXECUTION__CANCELED);
+    } else {
+      verifyStatusIsOkInOneServer(cancelCommandResult);
+      String msg1 = CliStrings.format(WAN_COPY_REGION__MSG__NO__RUNNING__COMMAND,
+          regionName, senderIdInA);
+      cancelCommandResult.hasTableSection(ResultModel.MEMBER_STATUS_SECTION).hasColumn("Message")
+          .containsExactlyInAnyOrder(msg1, msg1, WAN_COPY_REGION__MSG__EXECUTION__CANCELED);
+    }
+
+    // Check wan-copy region command output
+    CommandResultAssert wanCopyCommandResult = wanCopyCommandFuture.get();
+    if (isPartitionedRegion && isParallelGatewaySender) {
+      verifyStatusIsError(wanCopyCommandResult);
+      String msg = WAN_COPY_REGION__MSG__CANCELED__BEFORE__HAVING__COPIED;
+      wanCopyCommandResult.hasTableSection(ResultModel.MEMBER_STATUS_SECTION).hasColumn("Message")
+          .containsExactly(msg, msg, msg);
+    } else {
+      verifyStatusIsErrorInOneServer(wanCopyCommandResult);
+      String msg1 = CliStrings
+          .format(WAN_COPY_REGION__MSG__SENDER__SERIAL__AND__NOT__PRIMARY, senderIdInA);
+      wanCopyCommandResult.hasTableSection(ResultModel.MEMBER_STATUS_SECTION).hasColumn("Message")
+          .containsExactlyInAnyOrder(WAN_COPY_REGION__MSG__CANCELED__BEFORE__HAVING__COPIED, msg1,
+              msg1);
+    }
+    addIgnoredExceptionsForClosingAfterCancelCommand();
+  }
+
+  private int create3WanSitesAndClient(boolean isPartitionedRegion, VM locatorSender,
+      VM locatorSenderReceiver, VM locatorReceiver, List<VM> serversInA, VM serverInB,
+      VM serverInC, VM client, String senderIdInA, String senderIdInB) {
+    // Create locators
+    int receiverLocatorPort =
+        locatorReceiver.invoke(() -> createFirstLocatorWithDSId(3));
+    int senderReceiverLocatorPort = locatorSenderReceiver
+        .invoke(() -> createFirstRemoteLocator(2, receiverLocatorPort));
+    int senderLocatorPort = locatorSender.invoke(() -> {
+      Properties props = getDistributedSystemProperties();
+      props.setProperty(DISTRIBUTED_SYSTEM_ID, "" + 1);
+      props.setProperty(REMOTE_LOCATORS, "localhost[" + senderReceiverLocatorPort + "]");
+      LocatorLauncherStartupRule launcherStartupRule =
+          new LocatorLauncherStartupRule().withProperties(props);
+      launcherStartupRule.start();
+      return launcherStartupRule.getLauncher().getPort();
+    });
+
+    // Create servers
+    serverInB.invoke(() -> createServer(senderReceiverLocatorPort));
+    serverInC.invoke(() -> createServer(receiverLocatorPort));
+    for (VM server : serversInA) {
+      server.invoke(() -> createServer(senderLocatorPort));
+    }
+
+    // Create region in servers
+    final String regionName = getRegionName(isPartitionedRegion);
+    if (isPartitionedRegion) {
+      for (VM server : serversInA) {
+        server
+            .invoke(() -> createPartitionedRegion(regionName, senderIdInA, 1, 100,
+                isOffHeap(), RegionShortcut.PARTITION, true));
+      }
+      serverInB.invoke(
+          () -> createPartitionedRegion(regionName, senderIdInB, 0, 100,
+              isOffHeap(), RegionShortcut.PARTITION, true));
+      serverInC.invoke(() -> createPartitionedRegion(regionName, null, 0, 100,
+          isOffHeap(), RegionShortcut.PARTITION, true));
+    } else {
+      for (VM server : serversInA) {
+        server.invoke(() -> createReplicatedRegion(regionName, senderIdInA,
+            Scope.GLOBAL, DataPolicy.REPLICATE,
+            isOffHeap(), true));
+      }
+      serverInB
+          .invoke(() -> createReplicatedRegion(regionName, senderIdInB,
+              Scope.GLOBAL, DataPolicy.REPLICATE,
+              isOffHeap(), true));
+      serverInC.invoke(() -> createReplicatedRegion(regionName, null,
+          Scope.GLOBAL, DataPolicy.REPLICATE, isOffHeap(), true));
+    }
+
+    // Create client
+    client.invoke(() -> createClientWithLocatorAndRegion(senderLocatorPort, "localhost",
+        regionName, ClientRegionShortcut.PROXY));
+
+    return senderLocatorPort;
+  }
+
+  private int create2WanSitesAndClient(VM locatorInA, List<VM> serversInA, String senderIdInA,
+      VM locatorInB, List<VM> serversInB, VM client, boolean usePartitionedRegion,
+      String regionName) {
+    // Create locators
+    int locatorBPort = locatorInB.invoke(() -> createFirstLocatorWithDSId(2));
+    int locatorAPort = locatorInA.invoke(() -> {
+      Properties props = getDistributedSystemProperties();
+      props.setProperty(DISTRIBUTED_SYSTEM_ID, "" + 1);
+      props.setProperty(REMOTE_LOCATORS, "localhost[" + locatorBPort + "]");
+      LocatorLauncherStartupRule launcherStartupRule =
+          new LocatorLauncherStartupRule().withProperties(props);
+      launcherStartupRule.start();
+      return launcherStartupRule.getLauncher().getPort();
+    });
+
+    // Create servers and regions
+    createServersAndRegions(locatorBPort, serversInB, usePartitionedRegion, regionName, null);
+    createServersAndRegions(locatorAPort, serversInA, usePartitionedRegion, regionName,
+        senderIdInA);
+
+    // Create client
+    client.invoke(() -> createClientWithLocatorAndRegion(locatorAPort, "localhost",
+        regionName, ClientRegionShortcut.PROXY));
+
+    return locatorAPort;
+  }
+
+  private void createSenders(boolean isParallelGatewaySender, List<VM> serversInA,
+      VM serverInB, String senderIdInA, String senderIdInB) {
+    if (serverInB != null && senderIdInB != null) {
+      serverInB.invoke(() -> createSender(senderIdInB, 3,
+          isParallelGatewaySender, 100, 10, false,
+          false, null, false));
+    }
+    for (VM server : serversInA) {
+      server.invoke(() -> createSender(senderIdInA, 2, isParallelGatewaySender,
+          100, 10, false,
+          false, null, true));
+    }
+    startSenderInVMsAsync(senderIdInA, serversInA.toArray(new VM[0]));
+  }
+
+  private void createReceivers(VM serverInB, VM serverInC) {
+    createReceiverInVMs(serverInB);
+    createReceiverInVMs(serverInC);
+  }
+
+  private void stopReceiverAndVerifyResult(boolean useParallel, boolean stopPrimarySender,
+      int entries, String regionName, VM server1InB, VM server2InB, VM server3InB,
+      Future<CommandResultAssert> commandFuture)
+      throws InterruptedException, java.util.concurrent.ExecutionException {
+    // if parallel sender: stop any receiver
+    // if serial sender: stop receiver connected to primary or secondary
+    if (useParallel) {
+      server2InB.invoke(() -> cache.close());
+    } else {
+      // Region type has no influence on which server should be stopped
+      if (stopPrimarySender) {
+        // Stop the first server which had an available receiver
+        server1InB.invoke(() -> cache.close());
+      } else {
+        server3InB.invoke(() -> cache.close());
+      }
+    }
+
+    CommandResultAssert result = commandFuture.get();
+    // Verify result
+    if (useParallel) {
+      verifyResultOfStoppingReceiverWhenUsingParallelSender(result);
+    } else {
+      verifyResultOfStoppingReceiverWhenUsingSerialSender(result);
+      server2InB.invoke(() -> validateRegionSize(regionName, entries));
+    }
+  }
+
+  private void stopSenderAndVerifyResult(boolean useParallel, boolean stopPrimarySender,
+      VM server2InA, List<VM> serversInA, String senderIdInA,
+      Future<CommandResultAssert> wanCopyCommandFuture)
+      throws InterruptedException, java.util.concurrent.ExecutionException {
+    // If parallel: stop any server
+    // If serial: stop primary or secondary
+    if (useParallel) {
+      server2InA.invoke(() -> killSender(senderIdInA));
+    } else {
+      for (VM server : serversInA) {
+        boolean senderWasStopped = server.invoke(() -> {
+          GatewaySender sender = cache.getGatewaySender(senderIdInA);
+          if (((InternalGatewaySender) sender).isPrimary() == stopPrimarySender) {
+            killSender();
+            return true;
+          }
+          return false;
+        });
+        if (senderWasStopped) {
+          break;
+        }
+      }
+    }
+
+    CommandResultAssert result = wanCopyCommandFuture.get();
+    // Verify result
+    if (useParallel) {
+      verifyResultOfStoppingParallelSender(result);
+    } else {
+      if (stopPrimarySender) {
+        verifyResultOfStoppingPrimarySerialSender(result);
+      } else {
+        verifyResultStoppingSecondarySerialSender(result);
+      }
+    }
+  }
+
+  public void verifyResultOfStoppingReceiverWhenUsingSerialSender(
+      CommandResultAssert command) {
+    verifyStatusIsOk(command);
+    Condition<String> haveEntriesCopied =
+        new Condition<>(s -> s.startsWith("Entries copied: "), "Entries copied");
+    Condition<String> senderNotPrimary = new Condition<>(
+        s -> s.equals("Sender B is serial and not primary. 0 entries copied."),
+        "sender not primary");
+
+    command.hasTableSection(ResultModel.MEMBER_STATUS_SECTION).hasColumn("Message")
+        .asList().haveExactly(1, haveEntriesCopied).haveExactly(2, senderNotPrimary);
+  }
+
+  public void verifyResultOfStoppingReceiverWhenUsingParallelSender(
+      CommandResultAssert command) {
+    verifyStatusIsOk(command);
+    Condition<String> haveEntriesCopied =
+        new Condition<>(s -> s.startsWith("Entries copied: "), "Entries copied");
+
+    command.hasTableSection(ResultModel.MEMBER_STATUS_SECTION).hasColumn("Message")
+        .asList().haveExactly(3, haveEntriesCopied);
+  }
+
+  public void verifyResultOfStoppingParallelSender(CommandResultAssert command) {
+    verifyStatusIsErrorInOneServer(command);
+    Condition<String> startsWithError = new Condition<>(
+        s -> (s.startsWith("Execution failed. Error:")
+            || s.startsWith("Error (Unknown error sending batch)")
+            || s.startsWith("Error (Region destroyed)")
+            || s.startsWith("MemberResponse got memberDeparted event for")
+            || s.equals(WAN_COPY_REGION__MSG__CANCELED__BEFORE__HAVING__COPIED)),
+        "execution error");
+    Condition<String> haveEntriesCopied =
+        new Condition<>(s -> s.startsWith("Entries copied:"), "Entries copied");
+
+    command.hasTableSection(ResultModel.MEMBER_STATUS_SECTION).hasColumn("Message")
+        .asList().haveExactly(1, startsWithError).haveExactly(2, haveEntriesCopied);
+  }
+
+  public void verifyResultOfStoppingPrimarySerialSender(
+      CommandResultAssert command) {
+    verifyStatusIsErrorInOneServer(command);
+
+    Condition<String> startsWithError = new Condition<>(
+        s -> (s.startsWith("Execution failed. Error:")
+            || s.startsWith("Error (Unknown error sending batch)")
+            || s.startsWith("No connection available towards receiver after having copied")
+            || s.startsWith("Error (Region destroyed)")
+            || s.startsWith("MemberResponse got memberDeparted event for")
+            || s.equals(WAN_COPY_REGION__MSG__CANCELED__BEFORE__HAVING__COPIED)),
+        "execution error");
+
+    Condition<String> senderNotPrimary = new Condition<>(
+        s -> s.equals("Sender B is serial and not primary. 0 entries copied."),
+        "sender not primary");
+
+    command.hasTableSection(ResultModel.MEMBER_STATUS_SECTION).hasColumn("Message")
+        .asList().haveExactly(1, startsWithError).haveExactly(2, senderNotPrimary);
+  }
+
+  public void verifyResultStoppingSecondarySerialSender(
+      CommandResultAssert command) {
+    command.statusIsSuccess();
+    command.hasTableSection().hasColumns().hasSize(3);
+    command.hasTableSection(ResultModel.MEMBER_STATUS_SECTION).hasColumn("Member")
+        .hasSize(3);
+    command.hasTableSection(ResultModel.MEMBER_STATUS_SECTION).hasColumn("Message")
+        .hasSize(3);
+    command.hasTableSection(ResultModel.MEMBER_STATUS_SECTION).hasColumn("Status")
+        .containsExactlyInAnyOrder("OK", "OK", "OK");
+
+    Condition<String> haveEntriesCopied =
+        new Condition<>(s -> s.startsWith("Entries copied:"), "Entries copied");
+    Condition<String> senderNotPrimary = new Condition<>(
+        s -> s.equals("Sender B is serial and not primary. 0 entries copied."),
+        "sender not primary");
+
+    command.hasTableSection(ResultModel.MEMBER_STATUS_SECTION).hasColumn("Message")
+        .asList().haveExactly(1, haveEntriesCopied).haveExactly(2, senderNotPrimary);
+  }
+
+  private String getRegionName(boolean isPartitionedRegion) {
+    return getTestMethodName() + (isPartitionedRegion ? "_PR" : "RR");
+  }
+
+  public static void removeEntry(String regionName, long key) {
+    Region<?, ?> r = cache.getRegion(SEPARATOR + regionName);

Review comment:
       You should go ahead and name any variables with a full word like `region`.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org