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

import static java.lang.System.currentTimeMillis;
import static java.lang.System.lineSeparator;
import static java.util.Arrays.asList;
import static java.util.Arrays.stream;
import static java.util.stream.Collectors.toList;
import static org.apache.geode.cache.Region.SEPARATOR;
import static org.apache.geode.cache.RegionShortcut.REPLICATE;
import static org.apache.geode.cache.wan.GatewaySender.DEFAULT_ORDER_POLICY;
import static org.apache.geode.cache.wan.GatewaySender.DEFAULT_SOCKET_BUFFER_SIZE;
import static org.apache.geode.distributed.ConfigurationProperties.DISTRIBUTED_SYSTEM_ID;
import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
import static org.apache.geode.distributed.ConfigurationProperties.REMOTE_LOCATORS;
import static org.apache.geode.distributed.ConfigurationProperties.START_LOCATOR;
import static org.apache.geode.internal.AvailablePortHelper.getRandomAvailableTCPPort;
import static org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID.getNewProxyMembership;
import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
import static org.apache.geode.test.dunit.IgnoredException.addIgnoredException;
import static org.apache.geode.test.dunit.VM.getCurrentVMNum;
import static org.apache.geode.test.dunit.VM.getVM;
import static org.apache.geode.test.dunit.VM.toArray;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.catchThrowable;
import static org.hamcrest.Matchers.greaterThan;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.File;
import java.io.IOException;
import java.io.Serializable;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Properties;
import java.util.Set;
import java.util.StringTokenizer;

import org.apache.logging.log4j.Logger;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.junit.runners.Parameterized.Parameter;
import org.junit.runners.Parameterized.Parameters;
import org.junit.runners.Parameterized.UseParametersRunnerFactory;

import org.apache.geode.DataSerializable;
import org.apache.geode.DataSerializer;
import org.apache.geode.Instantiator;
import org.apache.geode.cache.CacheClosedException;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.DiskStore;
import org.apache.geode.cache.DiskStoreFactory;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.RegionFactory;
import org.apache.geode.cache.Scope;
import org.apache.geode.cache.client.PoolManager;
import org.apache.geode.cache.client.internal.locator.QueueConnectionRequest;
import org.apache.geode.cache.client.internal.locator.QueueConnectionResponse;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.cache.wan.GatewayReceiver;
import org.apache.geode.cache.wan.GatewayReceiverFactory;
import org.apache.geode.cache.wan.GatewaySender;
import org.apache.geode.cache.wan.GatewaySender.OrderPolicy;
import org.apache.geode.distributed.Locator;
import org.apache.geode.distributed.internal.InternalLocator;
import org.apache.geode.distributed.internal.ServerLocator;
import org.apache.geode.internal.cache.ForceReattemptException;
import org.apache.geode.internal.cache.InternalRegion;
import org.apache.geode.internal.cache.RegionQueue;
import org.apache.geode.internal.cache.tier.sockets.CacheServerTestUtil;
import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
import org.apache.geode.internal.cache.wan.AbstractGatewaySenderEventProcessor;
import org.apache.geode.internal.cache.wan.GatewaySenderException;
import org.apache.geode.internal.cache.wan.InternalGatewaySender;
import org.apache.geode.internal.cache.wan.InternalGatewaySenderFactory;
import org.apache.geode.internal.cache.wan.parallel.ParallelGatewaySenderQueue;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.test.dunit.AsyncInvocation;
import org.apache.geode.test.dunit.IgnoredException;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.cache.CacheTestCase;
import org.apache.geode.test.junit.categories.WanTest;
import org.apache.geode.test.junit.rules.serializable.SerializableTemporaryFolder;
import org.apache.geode.test.junit.runners.CategoryWithParameterizedRunnerFactory;

@Category(WanTest.class)
@RunWith(Parameterized.class)
@UseParametersRunnerFactory(CategoryWithParameterizedRunnerFactory.class)
@SuppressWarnings("serial")
public class SerialGatewaySenderOperationsDistributedTest extends CacheTestCase {
  private static final Logger logger = LogService.getLogger();

  @Parameters(name = "{index}: numDispatchers={0}")
  public static Collection<Integer> data() {
    return asList(1, 3, 5);
  }

  @Parameter
  public int numDispatchers;

  private VM vm0;
  private VM vm1;
  private VM vm2;
  private VM vm3;
  private VM vm4;
  private VM vm5;
  private VM vm6;
  private VM vm7;

  private String className;

  @Rule
  public SerializableTemporaryFolder temporaryFolder = new SerializableTemporaryFolder();

  @Before
  public void setUp() {
    className = getClass().getSimpleName();

    vm0 = getVM(0);
    vm1 = getVM(1);
    vm2 = getVM(2);
    vm3 = getVM(3);
    vm4 = getVM(4);
    vm5 = getVM(5);
    vm6 = getVM(6);
    vm7 = getVM(7);

    addIgnoredException("Broken pipe");
    addIgnoredException("Connection refused");
    addIgnoredException("Connection reset");
    addIgnoredException("could not get remote locator information");
    addIgnoredException("Software caused connection abort");
    addIgnoredException("Unexpected IOException");

    // Stopping the gateway closed the region, which causes this exception to get logged
    addIgnoredException(RegionDestroyedException.class);
  }

  @Test
  public void testSerialGatewaySenderOperationsWithoutStarting() {
    int lnPort = vm0.invoke(() -> createFirstLocatorWithDSId(1));
    int nyPort = vm1.invoke(() -> createFirstRemoteLocator(2, lnPort));

    for (VM vm : toArray(vm2, vm3)) {
      vm.invoke(() -> {
        createCache(nyPort);
        createReceiver();
      });
    }

    vm4.invoke(() -> createCache(lnPort));
    vm5.invoke(() -> createCache(lnPort));
    vm6.invoke(() -> createCache(lnPort));
    vm7.invoke(() -> createCache(lnPort));

    vm4.invoke(this::createSenderInVm4);
    vm5.invoke(this::createSenderInVm5);

    vm2.invoke(() -> createReplicatedRegion(className + "_RR", null));
    vm3.invoke(() -> createReplicatedRegion(className + "_RR", null));

    vm4.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
    vm5.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
    vm6.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
    vm7.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));

    vm4.invoke(() -> doPuts(className + "_RR", 10));
    vm4.invoke(() -> doPuts(className + "_RR", 100));

    vm4.invoke(() -> validateSenderOperations("ln"));
    vm5.invoke(() -> validateSenderOperations("ln"));
  }

  @Test
  public void testStartPauseResumeSerialGatewaySender() throws Exception {
    int lnPort = vm0.invoke(() -> createFirstLocatorWithDSId(1));
    int nyPort = vm1.invoke(() -> createFirstRemoteLocator(2, lnPort));

    for (VM vm : toArray(vm2, vm3)) {
      vm.invoke(() -> {
        createCache(nyPort);
        createReceiver();
      });
    }

    vm4.invoke(() -> createCache(lnPort));
    vm5.invoke(() -> createCache(lnPort));
    vm6.invoke(() -> createCache(lnPort));
    vm7.invoke(() -> createCache(lnPort));

    vm4.invoke(this::createSenderInVm4);
    vm5.invoke(this::createSenderInVm5);

    vm2.invoke(() -> createReplicatedRegion(className + "_RR", null));
    vm3.invoke(() -> createReplicatedRegion(className + "_RR", null));

    vm4.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
    vm5.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
    vm6.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
    vm7.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));

    vm4.invoke(() -> doPuts(className + "_RR", 10));

    for (VM vm : toArray(vm4, vm5)) {
      vm.invoke(() -> startSender("ln"));
    }

    vm4.invoke(() -> doPuts(className + "_RR", 100));

    vm4.invoke(() -> pauseSender("ln"));
    vm5.invoke(() -> pauseSender("ln"));

    vm4.invoke(() -> validateSenderPausedState("ln"));
    vm5.invoke(() -> validateSenderPausedState("ln"));

    AsyncInvocation<Void> doPutsInVm4 =
        vm4.invokeAsync(() -> doPuts(className + "_RR", 10));

    vm4.invoke(() -> resumeSender("ln"));
    vm5.invoke(() -> resumeSender("ln"));

    vm4.invoke(() -> validateSenderResumedState("ln"));
    vm5.invoke(() -> validateSenderResumedState("ln"));

    doPutsInVm4.await();

    vm4.invoke(() -> validateQueueContents("ln", 0));
    vm5.invoke(() -> validateQueueContents("ln", 0));

    vm2.invoke(() -> validateRegionSize(className + "_RR", 100));
    vm3.invoke(() -> validateRegionSize(className + "_RR", 100));
  }

  @Test
  public void testStopSerialGatewaySender() throws Exception {
    int lnPort = vm0.invoke(() -> createFirstLocatorWithDSId(1));
    int nyPort = vm1.invoke(() -> createFirstRemoteLocator(2, lnPort));

    for (VM vm : toArray(vm2, vm3)) {
      vm.invoke(() -> {
        createCache(nyPort);
        createReceiver();
      });
    }

    vm4.invoke(() -> createCache(lnPort));
    vm5.invoke(() -> createCache(lnPort));
    vm6.invoke(() -> createCache(lnPort));
    vm7.invoke(() -> createCache(lnPort));

    vm4.invoke(this::createSenderInVm4);
    vm5.invoke(this::createSenderInVm5);

    vm2.invoke(() -> createReplicatedRegion(className + "_RR", null));
    vm3.invoke(() -> createReplicatedRegion(className + "_RR", null));

    vm4.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
    vm5.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
    vm6.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
    vm7.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));

    vm4.invoke(() -> doPuts(className + "_RR", 20));

    for (VM vm : toArray(vm4, vm5)) {
      vm.invoke(() -> startSender("ln"));
    }

    vm4.invoke(() -> doPuts(className + "_RR", 20));

    vm2.invoke(() -> validateRegionSize(className + "_RR", 20));
    vm3.invoke(() -> validateRegionSize(className + "_RR", 20));

    vm2.invoke(this::stopReceivers);
    vm3.invoke(this::stopReceivers);

    vm4.invoke(() -> doPuts(className + "_RR", 20));

    vm4.invoke(() -> validateQueueSizeStat("ln", 20));
    vm5.invoke(() -> validateQueueSizeStat("ln", 20));

    vm4.invoke(() -> stopSender("ln"));
    vm5.invoke(() -> stopSender("ln"));

    vm4.invoke(() -> validateSenderStoppedState("ln"));
    vm5.invoke(() -> validateSenderStoppedState("ln"));

    vm4.invoke(() -> validateQueueSizeStat("ln", 0));
    vm5.invoke(() -> validateQueueSizeStat("ln", 0));
    /*
     * Should have no effect on GatewaySenderState
     */
    vm4.invoke(() -> resumeSender("ln"));
    vm5.invoke(() -> resumeSender("ln"));

    vm4.invoke(() -> validateSenderStoppedState("ln"));
    vm5.invoke(() -> validateSenderStoppedState("ln"));

    AsyncInvocation<Void> startSenderInVm4 = vm4.invokeAsync(() -> startSender("ln"));
    AsyncInvocation<Void> startSenderInVm5 = vm5.invokeAsync(() -> startSender("ln"));

    startSenderInVm4.await();
    startSenderInVm5.await();

    vm4.invoke(() -> validateQueueSizeStat("ln", 20));
    vm5.invoke(() -> validateQueueSizeStat("ln", 20));

    vm5.invoke(() -> doPuts(className + "_RR", 110));

    vm4.invoke(() -> validateQueueSizeStat("ln", 130));
    vm5.invoke(() -> validateQueueSizeStat("ln", 130));

    vm2.invoke(this::startReceivers);
    vm3.invoke(this::startReceivers);

    vm4.invoke(() -> validateSenderResumedState("ln"));
    vm5.invoke(() -> validateSenderResumedState("ln"));

    vm2.invoke(() -> validateRegionSize(className + "_RR", 110));
    vm3.invoke(() -> validateRegionSize(className + "_RR", 110));

    vm4.invoke(() -> validateQueueSizeStat("ln", 0));
    vm5.invoke(() -> validateQueueSizeStat("ln", 0));
  }

  @Test
  public void testRestartSerialGatewaySendersWhilePutting() throws Exception {
    int lnPort = vm0.invoke(() -> createFirstLocatorWithDSId(1));
    int nyPort = vm1.invoke(() -> createFirstRemoteLocator(2, lnPort));

    for (VM vm : toArray(vm2, vm3)) {
      vm.invoke(() -> {
        createCache(nyPort);
        createReceiver();
      });
    }

    vm4.invoke(() -> createCache(lnPort));
    vm5.invoke(() -> createCache(lnPort));
    vm6.invoke(() -> createCache(lnPort));
    vm7.invoke(() -> createCache(lnPort));

    vm4.invoke(this::createSenderInVm4);
    vm5.invoke(this::createSenderInVm5);

    vm2.invoke(() -> createReplicatedRegion(className + "_RR", null));
    vm3.invoke(() -> createReplicatedRegion(className + "_RR", null));

    vm4.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
    vm5.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
    vm6.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
    vm7.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));

    vm7.invoke(() -> doPuts(className + "_RR", 20));

    for (VM vm : toArray(vm4, vm5)) {
      vm.invoke(() -> startSender("ln"));
    }

    vm7.invoke(() -> doPuts(className + "_RR", 20));

    vm2.invoke(() -> validateRegionSize(className + "_RR", 20));
    vm3.invoke(() -> validateRegionSize(className + "_RR", 20));

    vm4.invoke(() -> stopSender("ln"));
    vm5.invoke(() -> stopSender("ln"));

    vm4.invoke(() -> validateSenderStoppedState("ln"));
    vm5.invoke(() -> validateSenderStoppedState("ln"));

    vm4.invoke(() -> validateQueueSizeStat("ln", 0));
    vm5.invoke(() -> validateQueueSizeStat("ln", 0));


    // do a lot of puts while senders are restarting
    AsyncInvocation<Void> doPutsInVm7 = vm7.invokeAsync(() -> doPuts(className + "_RR", 5000));

    AsyncInvocation<Void> startSenderInVm4 = vm4.invokeAsync(() -> startSender("ln"));
    AsyncInvocation<Void> startSenderInVm5 = vm5.invokeAsync(() -> startSender("ln"));

    startSenderInVm4.await();
    startSenderInVm5.await();

    doPutsInVm7.await();

    vm4.invoke(() -> validateQueueSizeStat("ln", 0));
    vm5.invoke(() -> validateQueueSizeStat("ln", 0));
    vm4.invoke(() -> validateSecondaryQueueSizeStat("ln", 0));
    vm5.invoke(() -> validateSecondaryQueueSizeStat("ln", 0));
  }

  @Test
  public void testSerialGatewaySendersPrintQueueContents() throws Exception {
    int lnPort = vm0.invoke(() -> createFirstLocatorWithDSId(1));
    int nyPort = vm1.invoke(() -> createFirstRemoteLocator(2, lnPort));

    for (VM vm : toArray(vm2, vm3)) {
      vm.invoke(() -> {
        createCache(nyPort);
        createReceiver();
      });
    }

    vm4.invoke(() -> createCache(lnPort));
    vm5.invoke(() -> createCache(lnPort));
    vm6.invoke(() -> createCache(lnPort));
    vm7.invoke(() -> createCache(lnPort));

    vm4.invoke(this::createSenderInVm4);
    vm5.invoke(this::createSenderInVm5);

    vm2.invoke(() -> createReplicatedRegion(className + "_RR", null));
    vm3.invoke(() -> createReplicatedRegion(className + "_RR", null));

    vm4.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
    vm5.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
    vm6.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
    vm7.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));

    for (VM vm : toArray(vm4, vm5)) {
      vm.invoke(() -> startSender("ln"));
    }
    vm4.invoke(() -> pauseSender("ln"));
    vm5.invoke(() -> pauseSender("ln"));

    vm7.invoke(() -> doPuts(className + "_RR", 20));

    await()
        .until(() -> stream(toArray(vm4, vm5))
            .map(vm -> vm.invoke(() -> {
              InternalGatewaySender sender = findInternalGatewaySender("ln");
              logger.info(displaySerialQueueContent(sender));
              return sender.getEventQueueSize();
            }))
            .mapToInt(i -> i)
            .sum(),
            greaterThan(0));

    AsyncInvocation<Void> resumeSenderInVm4 = vm4.invokeAsync(() -> resumeSender("ln"));
    AsyncInvocation<Void> resumeSenderInVm5 = vm5.invokeAsync(() -> resumeSender("ln"));

    resumeSenderInVm4.await();
    resumeSenderInVm5.await();

    vm4.invoke(() -> validateQueueSizeStat("ln", 0));
    vm5.invoke(() -> validateQueueSizeStat("ln", 0));
    vm4.invoke(() -> validateSecondaryQueueSizeStat("ln", 0));
    vm5.invoke(() -> validateSecondaryQueueSizeStat("ln", 0));
  }

  @Test
  public void testStopOneSerialGatewaySenderBothPrimary() throws Exception {
    int lnPort = vm0.invoke(() -> createFirstLocatorWithDSId(1));
    int nyPort = vm1.invoke(() -> createFirstRemoteLocator(2, lnPort));

    for (VM vm : toArray(vm2, vm3)) {
      vm.invoke(() -> {
        createCache(nyPort);
        createReceiver();
      });
    }

    vm4.invoke(() -> createCache(lnPort));
    vm5.invoke(() -> createCache(lnPort));
    vm6.invoke(() -> createCache(lnPort));
    vm7.invoke(() -> createCache(lnPort));

    vm4.invoke(this::createSenderInVm4);
    vm5.invoke(this::createSenderInVm5);

    vm2.invoke(() -> createReplicatedRegion(className + "_RR", null));
    vm3.invoke(() -> createReplicatedRegion(className + "_RR", null));

    vm4.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
    vm5.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
    vm6.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
    vm7.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));

    for (VM vm : toArray(vm4, vm5)) {
      vm.invoke(() -> startSender("ln"));
    }

    vm4.invoke(() -> doPuts(className + "_RR", 100));

    vm4.invoke(() -> stopSender("ln"));

    vm4.invoke(() -> validateSenderStoppedState("ln"));

    vm4.invoke(() -> doPuts(className + "_RR", 200));

    vm2.invoke(() -> validateRegionSize(className + "_RR", 200));
    vm3.invoke(() -> validateRegionSize(className + "_RR", 200));

    // Do some puts from both vm4 and vm5 while restarting a sender
    AsyncInvocation<Void> doPutsInVm4 =
        vm4.invokeAsync(() -> doPuts(className + "_RR", 300));

    Thread.sleep(10);
    vm4.invoke(() -> startSender("ln"));

    doPutsInVm4.await();

    vm2.invoke(() -> validateRegionSize(className + "_RR", 300));
    vm3.invoke(() -> validateRegionSize(className + "_RR", 300));

    vm4.invoke(() -> validateQueueSizeStat("ln", 0));
  }

  @Test
  public void testStopOneSerialGatewaySender_PrimarySecondary() {
    int lnPort = vm0.invoke(() -> createFirstLocatorWithDSId(1));
    int nyPort = vm1.invoke(() -> createFirstRemoteLocator(2, lnPort));

    for (VM vm : toArray(vm2, vm3)) {
      vm.invoke(() -> {
        createCache(nyPort);
        createReceiver();
      });
    }

    vm4.invoke(() -> createCache(lnPort));
    vm5.invoke(() -> createCache(lnPort));
    vm6.invoke(() -> createCache(lnPort));
    vm7.invoke(() -> createCache(lnPort));

    vm4.invoke(this::createSenderInVm4);
    vm5.invoke(this::createSenderInVm5);

    vm2.invoke(() -> createReplicatedRegion(className + "_RR", null));
    vm3.invoke(() -> createReplicatedRegion(className + "_RR", null));

    vm4.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
    vm5.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
    vm6.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
    vm7.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));

    for (VM vm : toArray(vm4, vm5)) {
      vm.invoke(() -> startSender("ln"));
    }

    vm4.invoke(() -> doPuts(className + "_RR", 10));

    vm4.invoke(() -> stopSender("ln"));

    vm4.invoke(() -> validateSenderStoppedState("ln"));

    vm4.invoke(() -> doPuts(className + "_RR", 100));

    vm2.invoke(() -> validateRegionSize(className + "_RR", 100));
    vm3.invoke(() -> validateRegionSize(className + "_RR", 100));
  }

  @Test
  public void testStopOneSender_StartAnotherSender() {
    int lnPort = vm0.invoke(() -> createFirstLocatorWithDSId(1));
    int nyPort = vm1.invoke(() -> createFirstRemoteLocator(2, lnPort));

    vm2.invoke(() -> createCache(nyPort));
    vm2.invoke(() -> createReplicatedRegion(className + "_RR", null));
    vm2.invoke(this::createReceiver);

    vm4.invoke(() -> createCache(lnPort));
    vm4.invoke(this::createSenderInVm4);
    vm4.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
    vm4.invoke(() -> startSender("ln"));

    vm4.invoke(() -> doPuts(className + "_RR", 10));
    vm4.invoke(() -> stopSender("ln"));
    vm4.invoke(() -> validateSenderStoppedState("ln"));

    vm5.invoke(() -> createCache(lnPort));
    vm5.invoke(this::createSenderInVm5);
    vm5.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
    vm5.invoke(() -> startSender("ln"));

    vm5.invoke(() -> doPuts(className + "_RR", 100));

    vm2.invoke(() -> validateRegionSize(className + "_RR", 100));
  }

  @Test
  public void test_Bug44153_StopOneSender_StartAnotherSender_CheckQueueSize() {
    int lnPort = vm0.invoke(() -> createFirstLocatorWithDSId(1));
    int nyPort = vm1.invoke(() -> createFirstRemoteLocator(2, lnPort));

    vm4.invoke(() -> createCache(lnPort));
    vm4.invoke(this::createSenderInVm4);
    vm4.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
    vm4.invoke(() -> startSender("ln"));

    vm4.invoke(() -> doPuts(className + "_RR", 10));
    vm4.invoke(() -> validateQueueContents("ln", 10));
    vm4.invoke(() -> stopSender("ln"));

    vm4.invoke(() -> validateSenderStoppedState("ln"));

    vm5.invoke(() -> createCache(lnPort));
    vm5.invoke(this::createSenderInVm5);
    vm5.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
    vm5.invoke(() -> startSender("ln"));

    vm5.invoke(() -> doPuts(className + "_RR", 10, 110));

    vm5.invoke(() -> validateQueueContents("ln", 100));
    vm5.invoke(() -> stopSender("ln"));
    vm5.invoke(() -> validateSenderStoppedState("ln"));

    vm4.invoke(() -> startSender("ln"));
    vm4.invoke(() -> validateQueueContents("ln", 10));
    vm4.invoke(() -> stopSender("ln"));

    vm5.invoke(() -> startSender("ln"));
    vm2.invoke(() -> createCache(nyPort));
    vm2.invoke(() -> createReplicatedRegion(className + "_RR", null));
    vm2.invoke(this::createReceiver);

    vm2.invoke(() -> validateRegionSize(className + "_RR", 100));
    vm5.invoke(() -> stopSender("ln"));

    vm4.invoke(() -> startSender("ln"));
    vm2.invoke(() -> validateRegionSize(className + "_RR", 110));
    vm4.invoke(() -> stopSender("ln"));
  }

  /**
   * Destroy SerialGatewaySender on all the nodes.
   */
  @Test
  public void testDestroySerialGatewaySenderOnAllNodes() {
    int lnPort = vm0.invoke(() -> createFirstLocatorWithDSId(1));
    int nyPort = vm1.invoke(() -> createFirstRemoteLocator(2, lnPort));

    for (VM vm : toArray(vm2, vm3)) {
      vm.invoke(() -> {
        createCache(nyPort);
        createReceiver();
      });
    }

    vm4.invoke(() -> createCache(lnPort));
    vm5.invoke(() -> createCache(lnPort));
    vm6.invoke(() -> createCache(lnPort));
    vm7.invoke(() -> createCache(lnPort));

    vm4.invoke(this::createSenderInVm4);
    vm5.invoke(this::createSenderInVm5);

    vm2.invoke(() -> createReplicatedRegion(className + "_RR", null));
    vm3.invoke(() -> createReplicatedRegion(className + "_RR", null));

    vm4.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
    vm5.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
    vm6.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
    vm7.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));

    for (VM vm : toArray(vm4, vm5)) {
      vm.invoke(() -> startSender("ln"));
    }

    vm4.invoke(() -> doPuts(className + "_RR", 10));

    // before destroying, stop the sender
    vm4.invoke(() -> stopSender("ln"));
    vm5.invoke(() -> stopSender("ln"));

    vm4.invoke(() -> removeSenderFromTheRegion("ln", className + "_RR"));
    vm5.invoke(() -> removeSenderFromTheRegion("ln", className + "_RR"));

    vm4.invoke(() -> destroySender("ln"));
    vm5.invoke(() -> destroySender("ln"));

    vm4.invoke(() -> validateSenderDestroyed("ln", false));
    vm5.invoke(() -> validateSenderDestroyed("ln", false));
  }

  /**
   * Destroy SerialGatewaySender on a single node.
   */
  @Test
  public void testDestroySerialGatewaySenderOnSingleNode() {
    int lnPort = vm0.invoke(() -> createFirstLocatorWithDSId(1));
    int nyPort = vm1.invoke(() -> createFirstRemoteLocator(2, lnPort));

    for (VM vm : toArray(vm2, vm3)) {
      vm.invoke(() -> {
        createCache(nyPort);
        createReceiver();
      });
    }

    vm4.invoke(() -> createCache(lnPort));
    vm5.invoke(() -> createCache(lnPort));
    vm6.invoke(() -> createCache(lnPort));
    vm7.invoke(() -> createCache(lnPort));

    vm4.invoke(this::createSenderInVm4);
    vm5.invoke(this::createSenderInVm5);

    vm2.invoke(() -> createReplicatedRegion(className + "_RR", null));
    vm3.invoke(() -> createReplicatedRegion(className + "_RR", null));

    vm4.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
    vm5.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
    vm6.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
    vm7.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));

    for (VM vm : toArray(vm4, vm5)) {
      vm.invoke(() -> startSender("ln"));
    }

    vm4.invoke(() -> doPuts(className + "_RR", 10));

    // before destroying, stop the sender
    vm4.invoke(() -> stopSender("ln"));

    vm4.invoke(() -> removeSenderFromTheRegion("ln", className + "_RR"));

    vm4.invoke(() -> destroySender("ln"));

    vm4.invoke(() -> validateSenderDestroyed("ln", false));
    vm5.invoke(() -> verifySenderRunningState("ln"));
  }

  /**
   * Since the sender is attached to a region and in use, it can not be destroyed. Hence, exception
   * is thrown by the sender API.
   */
  @Test
  public void testDestroySerialGatewaySenderExceptionScenario() {
    int lnPort = vm0.invoke(() -> createFirstLocatorWithDSId(1));
    int nyPort = vm1.invoke(() -> createFirstRemoteLocator(2, lnPort));

    for (VM vm : toArray(vm2, vm3)) {
      vm.invoke(() -> {
        createCache(nyPort);
        createReceiver();
      });
    }

    vm4.invoke(() -> createCache(lnPort));
    vm5.invoke(() -> createCache(lnPort));
    vm6.invoke(() -> createCache(lnPort));
    vm7.invoke(() -> createCache(lnPort));

    vm4.invoke(this::createSenderInVm4);
    vm5.invoke(this::createSenderInVm5);

    vm2.invoke(() -> createReplicatedRegion(className + "_RR", null));
    vm3.invoke(() -> createReplicatedRegion(className + "_RR", null));

    vm4.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
    vm5.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
    vm6.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
    vm7.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));

    for (VM vm : toArray(vm4, vm5)) {
      vm.invoke(() -> startSender("ln"));
    }

    vm4.invoke(() -> doPuts(className + "_RR", 10));

    vm4.invoke(() -> {
      Throwable thrown = catchThrowable(() -> destroySender("ln"));
      assertThat(thrown).isInstanceOf(GatewaySenderException.class);
    });

    vm2.invoke(() -> validateRegionSize(className + "_RR", 10));
  }

  @Test
  public void testGatewaySenderNotRegisteredAsCacheServer() {
    int lnPort = vm0.invoke(() -> createFirstLocatorWithDSId(1));
    int nyPort = vm1.invoke(() -> createFirstRemoteLocator(2, lnPort));

    for (VM vm : toArray(vm2, vm3)) {
      vm.invoke(() -> {
        createCache(nyPort);
        createReceiver();
      });
    }

    for (VM vm : toArray(vm4, vm5)) {
      vm.invoke(() -> createCache(lnPort));
    }

    vm4.invoke(() -> createSender("ln", 2, true, true, numDispatchers, DEFAULT_ORDER_POLICY));
    vm5.invoke(() -> createSender("ln", 2, true, true, numDispatchers, DEFAULT_ORDER_POLICY));

    for (VM vm : toArray(vm4, vm5)) {
      vm.invoke(() -> startSender("ln"));
    }

    for (VM vm : toArray(vm0, vm1)) {
      vm.invoke(() -> {
        InternalLocator internalLocator = (InternalLocator) Locator.getLocator();
        ServerLocator serverLocator = internalLocator.getServerLocatorAdvisee();

        assertThat(serverLocator.getLoadMap())
            .as("Empty server load map")
            .isEmpty();

        ClientProxyMembershipID clientProxyMembership = getNewProxyMembership(getSystem());
        QueueConnectionRequest request =
            new QueueConnectionRequest(clientProxyMembership, 1, new HashSet<>(), "", false);
        QueueConnectionResponse response =
            (QueueConnectionResponse) serverLocator.processRequest(request);

        assertThat(response.getServers())
            .as("Empty response servers")
            .isEmpty();
      });
    }
  }

  @Test
  public void registeringInstantiatorsInGatewayShouldNotCauseDeadlock() {
    int lnPort = vm0.invoke(() -> createFirstLocatorWithDSId(1));
    int nyPort = vm1.invoke(() -> createFirstRemoteLocator(2, lnPort));

    vm2.invoke(() -> createReceiverAndServer(nyPort));
    vm3.invoke(() -> createReceiverAndServer(lnPort));

    vm2.invoke(() -> createSender("ln", 1, false, false, numDispatchers, DEFAULT_ORDER_POLICY));
    vm3.invoke(() -> createSender("ny", 2, false, false, numDispatchers, DEFAULT_ORDER_POLICY));

    vm4.invoke(() -> createClientWithLocator(nyPort, "localhost"));

    // Register instantiator
    vm4.invoke(() -> Instantiator.register(new TestObjectInstantiator()));
  }

  @Test
  public void registeringDataSerializableInGatewayShouldNotCauseDeadlock() {
    int lnPort = vm0.invoke(() -> createFirstLocatorWithDSId(1));
    int nyPort = vm1.invoke(() -> createFirstRemoteLocator(2, lnPort));

    vm2.invoke(() -> createReceiverAndServer(nyPort));
    vm3.invoke(() -> createReceiverAndServer(lnPort));

    vm2.invoke(() -> createSender("ln", 1, false, false, numDispatchers, DEFAULT_ORDER_POLICY));
    vm3.invoke(() -> createSender("ny", 2, false, false, numDispatchers, DEFAULT_ORDER_POLICY));

    vm4.invoke(() -> createClientWithLocator(nyPort, "localhost"));

    // Register instantiator
    vm4.invoke(() -> DataSerializer.register(TestObjectDataSerializer.class));
  }

  protected boolean isOffHeap() {
    return false;
  }

  protected void createSenderInVm4() throws IOException {
    createSender("ln", 2, true, true, numDispatchers, DEFAULT_ORDER_POLICY);
  }

  protected void createSenderInVm5() throws IOException {
    createSender("ln", 2, true, true, numDispatchers, DEFAULT_ORDER_POLICY);
  }

  protected final void createSender(String id,
      int remoteDsId,
      boolean isPersistent,
      boolean isManualStart,
      int numDispatchers,
      OrderPolicy policy) throws IOException {
    try (IgnoredException ie = addIgnoredException("Could not connect")) {
      File persistentDirectory =
          temporaryFolder.newFolder(id + "_disk_" + currentTimeMillis() + "_" + getCurrentVMNum());
      DiskStoreFactory diskStoreFactory = getCache().createDiskStoreFactory();
      File[] dirs = new File[] {persistentDirectory};

      InternalGatewaySenderFactory gatewaySenderFactory =
          (InternalGatewaySenderFactory) getCache().createGatewaySenderFactory();

      gatewaySenderFactory.setParallel(false);
      gatewaySenderFactory.setMaximumQueueMemory(100);
      gatewaySenderFactory.setBatchSize(10);
      gatewaySenderFactory.setBatchConflationEnabled(false);
