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

import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.MINUTES;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.geode.distributed.ConfigurationProperties.DISABLE_AUTO_RECONNECT;
import static org.apache.geode.distributed.ConfigurationProperties.ENABLE_CLUSTER_CONFIGURATION;
import static org.apache.geode.distributed.ConfigurationProperties.ENABLE_NETWORK_PARTITION_DETECTION;
import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
import static org.apache.geode.distributed.ConfigurationProperties.LOCATOR_WAIT_TIME;
import static org.apache.geode.distributed.ConfigurationProperties.LOG_LEVEL;
import static org.apache.geode.distributed.ConfigurationProperties.MAX_WAIT_TIME_RECONNECT;
import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
import static org.apache.geode.distributed.ConfigurationProperties.MEMBER_TIMEOUT;
import static org.apache.geode.distributed.ConfigurationProperties.NAME;
import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_PEER_AUTHENTICATOR;
import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_PEER_AUTH_INIT;
import static org.apache.geode.distributed.ConfigurationProperties.SSL_CIPHERS;
import static org.apache.geode.distributed.ConfigurationProperties.SSL_ENABLED_COMPONENTS;
import static org.apache.geode.distributed.ConfigurationProperties.SSL_KEYSTORE;
import static org.apache.geode.distributed.ConfigurationProperties.SSL_KEYSTORE_PASSWORD;
import static org.apache.geode.distributed.ConfigurationProperties.SSL_KEYSTORE_TYPE;
import static org.apache.geode.distributed.ConfigurationProperties.SSL_LOCATOR_ALIAS;
import static org.apache.geode.distributed.ConfigurationProperties.SSL_PROTOCOLS;
import static org.apache.geode.distributed.ConfigurationProperties.SSL_TRUSTSTORE;
import static org.apache.geode.distributed.ConfigurationProperties.SSL_TRUSTSTORE_PASSWORD;
import static org.apache.geode.distributed.ConfigurationProperties.START_LOCATOR;
import static org.apache.geode.distributed.ConfigurationProperties.USE_CLUSTER_CONFIGURATION;
import static org.apache.geode.internal.AvailablePortHelper.getRandomAvailableTCPPort;
import static org.apache.geode.internal.security.SecurableCommunicationChannel.LOCATOR;
import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
import static org.apache.geode.test.dunit.Disconnect.disconnectFromDS;
import static org.apache.geode.test.dunit.DistributedTestUtils.deleteLocatorStateFile;
import static org.apache.geode.test.dunit.IgnoredException.addIgnoredException;
import static org.apache.geode.test.dunit.VM.getController;
import static org.apache.geode.test.dunit.VM.getHostName;
import static org.apache.geode.test.dunit.VM.getVM;
import static org.apache.geode.test.dunit.VM.toArray;
import static org.apache.geode.test.util.ResourceUtils.createTempFileFromResource;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import java.io.File;
import java.io.IOException;
import java.io.Serializable;
import java.net.ConnectException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Properties;
import java.util.Set;

import org.apache.logging.log4j.Logger;
import org.junit.After;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.junit.experimental.categories.Category;

import org.apache.geode.ForcedDisconnectException;
import org.apache.geode.GemFireConfigException;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheFactory;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionShortcut;
import org.apache.geode.distributed.internal.ClusterDistributionManager;
import org.apache.geode.distributed.internal.Distribution;
import org.apache.geode.distributed.internal.DistributionException;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.HighPriorityAckedMessage;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.InternalLocator;
import org.apache.geode.distributed.internal.MembershipListener;
import org.apache.geode.distributed.internal.MembershipTestHook;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.distributed.internal.membership.api.MemberDisconnectedException;
import org.apache.geode.distributed.internal.membership.api.MemberIdentifier;
import org.apache.geode.distributed.internal.membership.api.MembershipConfigurationException;
import org.apache.geode.distributed.internal.membership.api.MembershipManagerHelper;
import org.apache.geode.distributed.internal.membership.api.MembershipView;
import org.apache.geode.internal.AvailablePortHelper;
import org.apache.geode.internal.membership.utils.AvailablePort;
import org.apache.geode.internal.tcp.Connection;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.test.awaitility.GeodeAwaitility;
import org.apache.geode.test.dunit.AsyncInvocation;
import org.apache.geode.test.dunit.DUnitBlackboard;
import org.apache.geode.test.dunit.DistributedTestUtils;
import org.apache.geode.test.dunit.Host;
import org.apache.geode.test.dunit.IgnoredException;
import org.apache.geode.test.dunit.Invoke;
import org.apache.geode.test.dunit.NetworkUtils;
import org.apache.geode.test.dunit.SerializableRunnable;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.Wait;
import org.apache.geode.test.dunit.internal.DUnitLauncher;
import org.apache.geode.test.dunit.rules.DistributedErrorCollector;
import org.apache.geode.test.dunit.rules.DistributedRule;
import org.apache.geode.test.junit.categories.MembershipTest;

/**
 * Tests the ability of the {@link Locator} API to start and stop locators running in remote VMs.
 *
 * @since GemFire 4.0
 */
@Category(MembershipTest.class)
public class LocatorDUnitTest implements Serializable {
  private static final Logger logger = LogService.getLogger();

  protected static volatile InternalDistributedSystem system;
  private static volatile DUnitBlackboard blackboard;
  private static volatile TestHook hook;

  protected int port1;
  private int port2;
  private int port3;
  private int port4;

  protected String hostName;

  protected VM vm0;
  private VM vm1;
  private VM vm2;
  private VM vm3;
  private VM vm4;

  @Rule
  public DistributedRule distributedRule = new DistributedRule(6);

  @Rule
  public DistributedErrorCollector errorCollector = new DistributedErrorCollector();

  @Before
  public void setUp() {
    addIgnoredException("Removing shunned member");

    vm0 = getVM(0);
    vm1 = getVM(1);
    vm2 = getVM(2);
    vm3 = getVM(3);
    vm4 = getVM(4);

    hostName = NetworkUtils.getServerHostName();

    int[] ports = AvailablePortHelper.getRandomAvailableTCPPorts(4);
    port1 = ports[0];
    port2 = ports[1];
    port3 = ports[2];
    port4 = ports[3];
    Invoke.invokeInEveryVM(() -> deleteLocatorStateFile(port1, port2, port3, port4));
    addIgnoredException(MemberDisconnectedException.class); // ignore this suspect string
    addIgnoredException(MembershipConfigurationException.class); // ignore this suspect string
  }

  @After
  public void tearDown() {
    for (VM vm : toArray(getController(), vm0, vm1, vm2, vm3, vm4)) {
      vm.invoke(() -> {
        stopLocator();
        disconnectFromDS();
        system = null;
        blackboard = null;
        hook = null;
      });
    }

    // delete locator state files so they don't accidentally get used by other tests
    Invoke.invokeInEveryVM(() -> {
      for (int port : new int[] {port1, port2, port3, port4}) {
        if (port > 0) {
          deleteLocatorStateFile(port);
        }
      }
    });
  }

  private static DUnitBlackboard getBlackboard() {
    if (blackboard == null) {
      blackboard = new DUnitBlackboard();
    }
    return blackboard;
  }

  private static void expectSystemToContainThisManyMembers(final int expectedMembers) {
    assertThat(system).isNotNull();
    await()
        .untilAsserted(() -> assertEquals(expectedMembers, system.getDM().getViewMembers().size()));
  }

  private static boolean isSystemConnected() {
    return system != null && system.isConnected();
  }

  private static void disconnectDistributedSystem() {
    if (system != null && system.isConnected()) {
      system.disconnect();
    }
    MembershipManagerHelper.inhibitForcedDisconnectLogging(false);
  }

  /**
   * return the distributed member id for the ds on this vm
   */
  private static DistributedMember getDistributedMember(Properties props) {
    props.setProperty("name", "vm_" + VM.getCurrentVMNum());
    DistributedSystem sys = getConnectedDistributedSystem(props);
    addIgnoredException("service failure");
    addIgnoredException(ForcedDisconnectException.class);
    return sys.getDistributedMember();
  }

  /**
   * find a running locator and return its distributed member id
   */
  private static DistributedMember getLocatorDistributedMember() {
    return Locator.getLocator().getDistributedSystem().getDistributedMember();
  }

  /**
   * find the lead member and return its id
   */
  private static DistributedMember getLeadMember() {
    return MembershipManagerHelper.getLeadMember(system);
  }

  protected static void stopLocator() {
    MembershipManagerHelper.inhibitForcedDisconnectLogging(false);
    Locator loc = Locator.getLocator();
    if (loc != null) {
      loc.stop();
      assertThat(Locator.hasLocator()).isFalse();
    }
  }

  @Test
  @Ignore("GEODE-7760 - test sometimes hangs due to product issue")
  public void testCrashLocatorMultipleTimes() throws Exception {
    port1 = getRandomAvailableTCPPort();
    DistributedTestUtils.deleteLocatorStateFile(port1);
    File logFile = new File("");
    File stateFile = new File("locator" + port1 + "state.dat");
    VM vm = VM.getVM(0);
    final Properties properties =
        getBasicProperties(Host.getHost(0).getHostName() + "[" + port1 + "]");
    int memberTimeoutMS = 3000;
    properties.put(MEMBER_TIMEOUT, "" + memberTimeoutMS);
    properties.put(MAX_WAIT_TIME_RECONNECT, "" + (3 * memberTimeoutMS));
    // since we're restarting location services let's be a little forgiving about that service
    // starting up so that stress-tests can pass
    properties.put(LOCATOR_WAIT_TIME, "" + 3);
    addDSProps(properties);
    if (stateFile.exists()) {
      assertThat(stateFile.delete()).isTrue();
    }

    IgnoredException
        .addIgnoredException("Possible loss of quorum due to the loss of 1 cache processes");

    Locator locator = Locator.startLocatorAndDS(port1, logFile, properties);
    system = (InternalDistributedSystem) locator.getDistributedSystem();

    vm.invoke(() -> {
      getConnectedDistributedSystem(properties);
      return null;
    });

    try {
      for (int i = 0; i < 4; i++) {
        forceDisconnect();
        system.waitUntilReconnected(GeodeAwaitility.getTimeout().toMillis(), MILLISECONDS);
        assertThat(system.getReconnectedSystem()).isNotNull();
        system = (InternalDistributedSystem) system.getReconnectedSystem();
      }
      assertEquals(2, ((InternalDistributedSystem) locator.getDistributedSystem()).getDM()
          .getViewMembers().size());
    } finally {
      vm.invoke("disconnect", () -> {
        getConnectedDistributedSystem(properties).disconnect();
        return null;
      });
      locator.stop();
    }

  }

  /**
   * This tests that the locator can resume control as coordinator after all locators have been shut
   * down and one is restarted. It's necessary to have a lock service start so elder failover is
   * forced to happen. Prior to fixing how this worked it hung with the restarted locator trying to
   * become elder again because it put its address at the beginning of the new view it sent out.
   */
  @Test
  public void testCollocatedLocatorWithSecurity() {
    String locators = hostName + "[" + port1 + "]";

    Properties properties = new Properties();
    properties.setProperty(SECURITY_PEER_AUTH_INIT,
        "org.apache.geode.distributed.AuthInitializer.create");
    properties.setProperty(SECURITY_PEER_AUTHENTICATOR,
        "org.apache.geode.distributed.MyAuthenticator.create");
    properties.setProperty(START_LOCATOR, locators);
    addDSProps(properties);

    system = getConnectedDistributedSystem(properties);
    assertThat(system.getDistributedMember().getVmKind())
        .describedAs("expected the VM to have NORMAL vmKind")
        .isEqualTo(ClusterDistributionManager.NORMAL_DM_TYPE);

    properties.remove(START_LOCATOR);
    properties.setProperty(LOCATORS, locators);

    SerializableRunnable startSystem = new SerializableRunnable("start system") {
      @Override
      public void run() {
        system = getConnectedDistributedSystem(properties);
      }
    };
    vm1.invoke(startSystem);
    vm2.invoke(startSystem);

    // ensure that I, as a collocated locator owner, can create a cache region
    Cache cache = CacheFactory.create(system);
    Region<?, ?> r = cache.createRegionFactory(RegionShortcut.REPLICATE).create("test-region");
    assertThat(r).describedAs("expected to create a region").isNotNull();

    // create a lock service and have every vm get a lock
    DistributedLockService service = DistributedLockService.create("test service", system);
    service.becomeLockGrantor();
    service.lock("foo0", 0, 0);

    vm1.invoke("get the lock service and lock something",
        () -> DistributedLockService.create("test service", system).lock("foo1", 0, 0));

    vm2.invoke("get the lock service and lock something",
        () -> DistributedLockService.create("test service", system).lock("foo2", 0, 0));

    // cause elder failover. vm1 will become the lock grantor
    system.disconnect();

    vm1.invoke("ensure grantor failover", () -> {
      DistributedLockService serviceNamed = DistributedLockService.getServiceNamed("test service");
      serviceNamed.lock("foo3", 0, 0);
      await().until(serviceNamed::isLockGrantor);
      assertThat(serviceNamed.isLockGrantor()).isTrue();
    });

    properties.setProperty(START_LOCATOR, locators);

    system = getConnectedDistributedSystem(properties);
    System.out.println("done connecting distributed system.  Membership view is "
        + MembershipManagerHelper.getDistribution(system).getView());

    assertThat(MembershipManagerHelper.getCoordinator(system))
        .describedAs("should be the coordinator").isEqualTo(system.getDistributedMember());
    MembershipView<InternalDistributedMember> view =
        MembershipManagerHelper.getDistribution(system).getView();
    logger.info("view after becoming coordinator is " + view);
    assertThat(system.getDistributedMember())
        .describedAs("should not be the first member in the view (" + view + ")")
        .isNotSameAs(view.get(0));

    service = DistributedLockService.create("test service", system);

    // now force a non-elder VM to get a lock. This will hang if the bug is not fixed
    vm2.invoke("get the lock service and lock something", () -> {
      DistributedLockService.getServiceNamed("test service").lock("foo4", 0, 0);
    });

    assertThat(service.isLockGrantor()).describedAs("should not have become lock grantor")
        .isFalse();

    // Now demonstrate that a new member can join and use the lock service
    properties.remove(START_LOCATOR);
    vm3.invoke(startSystem);
    vm3.invoke("get the lock service and lock something(2)",
        () -> DistributedLockService.create("test service", system).lock("foo5", 0, 0));
  }

  /**
   * Bug 30341 concerns race conditions in JGroups that allow two locators to start up in a
   * split-brain configuration. To work around this we have always told customers that they need to
   * stagger the starting of locators. This test configures two locators to start up simultaneously
   * and shows that they find each other and form a single system.
   */
  @Test
  public void testStartTwoLocators() throws Exception {
    String locators = hostName + "[" + port1 + "]," + hostName + "[" + port2 + "]";

    Properties properties = getClusterProperties(locators, "false");
    addDSProps(properties);

    startVerifyAndStopLocator(vm1, vm2, port1, port2, properties);
    startVerifyAndStopLocator(vm1, vm2, port1, port2, properties);
    startVerifyAndStopLocator(vm1, vm2, port1, port2, properties);
  }

  @Test
  public void testStartTwoLocatorsWithSingleKeystoreSSL() throws Exception {
    String locators = hostName + "[" + port1 + "]," + hostName + "[" + port2 + "]";

    Properties properties = getClusterProperties(locators, "false");
    properties.setProperty(SSL_CIPHERS, "any");
    properties.setProperty(SSL_ENABLED_COMPONENTS, LOCATOR.getConstant());
    properties.setProperty(SSL_KEYSTORE, getSingleKeyKeystore());
    properties.setProperty(SSL_KEYSTORE_PASSWORD, "password");
    properties.setProperty(SSL_KEYSTORE_TYPE, "JKS");
    properties.setProperty(SSL_PROTOCOLS, "TLSv1.2");
    properties.setProperty(SSL_TRUSTSTORE, getSingleKeyKeystore());
    properties.setProperty(SSL_TRUSTSTORE_PASSWORD, "password");

    startVerifyAndStopLocator(vm1, vm2, port1, port2, properties);
  }

  @Test
  public void testStartTwoLocatorsWithMultiKeystoreSSL() throws Exception {
    String locators = hostName + "[" + port1 + "]," + hostName + "[" + port2 + "]";

    Properties properties = getClusterProperties(locators, "false");
    properties.setProperty(SSL_CIPHERS, "any");
    properties.setProperty(SSL_ENABLED_COMPONENTS, LOCATOR.getConstant());
    properties.setProperty(SSL_KEYSTORE, getMultiKeyKeystore());
    properties.setProperty(SSL_KEYSTORE_PASSWORD, "password");
    properties.setProperty(SSL_KEYSTORE_TYPE, "JKS");
    properties.setProperty(SSL_TRUSTSTORE, getMultiKeyTruststore());
    properties.setProperty(SSL_TRUSTSTORE_PASSWORD, "password");
    properties.setProperty(SSL_LOCATOR_ALIAS, "locatorkey");
    properties.setProperty(SSL_PROTOCOLS, "any");

    startVerifyAndStopLocator(vm1, vm2, port1, port2, properties);
  }

  /**
   * test lead member selection
   */
  @Test
  public void testLeadMemberSelection() throws Exception {
    String locators = hostName + "[" + port1 + "]";

    Properties properties = getBasicProperties(locators);
    properties.setProperty(DISABLE_AUTO_RECONNECT, "true");
    properties.setProperty(ENABLE_NETWORK_PARTITION_DETECTION, "true");
    addDSProps(properties);

    Locator locator = Locator.startLocatorAndDS(port1, null, properties);

    system = (InternalDistributedSystem) locator.getDistributedSystem();

    assertThat(MembershipManagerHelper.getLeadMember(system)).isNull();

    // connect three vms and then watch the lead member selection as they
    // are disconnected/reconnected
    properties.setProperty("name", "vm1");

    DistributedMember mem1 = vm1.invoke(() -> getDistributedMember(properties));

    waitForMemberToBecomeLeadMemberOfDistributedSystem(mem1, system);

    properties.setProperty("name", "vm2");
    DistributedMember mem2 = vm2.invoke(() -> getDistributedMember(properties));
    waitForMemberToBecomeLeadMemberOfDistributedSystem(mem1, system);

    properties.setProperty("name", "vm3");
    DistributedMember mem3 = vm3.invoke(() -> getDistributedMember(properties));
    waitForMemberToBecomeLeadMemberOfDistributedSystem(mem1, system);

    // after disconnecting the first vm, the second one should become the leader
    vm1.invoke(LocatorDUnitTest::disconnectDistributedSystem);
    MembershipManagerHelper.getDistribution(system).waitForDeparture(
        (InternalDistributedMember) mem1);
    waitForMemberToBecomeLeadMemberOfDistributedSystem(mem2, system);

    properties.setProperty("name", "vm1");
    mem1 = vm1.invoke(() -> getDistributedMember(properties));
    waitForMemberToBecomeLeadMemberOfDistributedSystem(mem2, system);

    vm2.invoke(LocatorDUnitTest::disconnectDistributedSystem);
    MembershipManagerHelper.getDistribution(system).waitForDeparture(
        (InternalDistributedMember) mem2);
    waitForMemberToBecomeLeadMemberOfDistributedSystem(mem3, system);

    vm1.invoke(LocatorDUnitTest::disconnectDistributedSystem);
    MembershipManagerHelper.getDistribution(system).waitForDeparture(
        (InternalDistributedMember) mem1);
    waitForMemberToBecomeLeadMemberOfDistributedSystem(mem3, system);

    vm3.invoke(LocatorDUnitTest::disconnectDistributedSystem);
    MembershipManagerHelper.getDistribution(system).waitForDeparture(
        (InternalDistributedMember) mem3);
    waitForMemberToBecomeLeadMemberOfDistributedSystem(null, system);
  }

  /**
   * test lead member and coordinator failure with network partition detection enabled. It would be
   * nice for this test to have more than two "server" vms, to demonstrate that they all exit when
   * the leader and potential- coordinator both disappear in the loss-correlation-window, but there
   * are only four vms available for DUnit testing.
   * <p>
   * So, we start two locators with admin distributed systems, then start two regular distributed
   * members.
   * <p>
   * We kill the second locator (which is not the view coordinator) and then kill the non-lead
   * member. That should be okay - the lead and remaining locator continue to run.
   * <p>
   * We then kill the lead member and demonstrate that the original locator (which is now the sole
   * remaining member) shuts itself down.
   */
  @Test
  public void testLeadAndCoordFailure() throws Exception {
    addIgnoredException("Possible loss of quorum due");

    String locators = hostName + "[" + port1 + "]," + hostName + "[" + port2 + "]";

    Properties properties = getClusterProperties(locators, "true");
    addDSProps(properties);

    Locator locator = Locator.startLocatorAndDS(port1, null, properties);
    system = (InternalDistributedSystem) locator.getDistributedSystem();

    addIgnoredException(ConnectException.class);
    MembershipManagerHelper.inhibitForcedDisconnectLogging(true);
    startLocator(vm3, properties, port2);

    assertThat(MembershipManagerHelper.getLeadMember(system)).isNull();

    DistributedMember mem1 = vm1.invoke(() -> getDistributedMember(properties));
    vm2.invoke(() -> getDistributedMember(properties));
    waitForMemberToBecomeLeadMemberOfDistributedSystem(mem1, system);

    assertThat(system.getDistributedMember())
        .isEqualTo(MembershipManagerHelper.getCoordinator(system));

    // crash the second vm and the locator. Should be okay
    DistributedTestUtils.crashDistributedSystem(vm2);
    vm3.invoke(() -> {
      Locator loc = Locator.getLocator();
      MembershipManagerHelper.crashDistributedSystem(loc.getDistributedSystem());
      loc.stop();
    });

    assertThat(vm1.invoke(LocatorDUnitTest::isSystemConnected))
        .describedAs("Distributed system should not have disconnected").isTrue();

    // ensure quorumLost is properly invoked
    DistributionManager dm = system.getDistributionManager();
    MyMembershipListener listener = new MyMembershipListener();
    dm.addMembershipListener(listener);
    // ensure there is an unordered reader thread for the member
    new HighPriorityAckedMessage().send(Collections.singleton(mem1), false);

    // disconnect the first vm and demonstrate that the third vm and the
    // locator notice the failure and exit
    DistributedTestUtils.crashDistributedSystem(vm1);

    /*
     * This vm is watching vm1, which is watching vm2 which is watching locatorVM. It will take 3
     * (3 member-timeout) milliseconds to detect the full failure and eject the lost members from
     * the view.
     */

    logger.info("waiting for my distributed system to disconnect due to partition detection");

    await().until(() -> !system.isConnected());

    if (system.isConnected()) {
      fail(
          "Distributed system did not disconnect as expected - network partition detection is broken");
    }
    // quorumLost should be invoked if we get a ForcedDisconnect in this situation
    assertThat(listener.quorumLostInvoked).describedAs("expected quorumLost to be invoked")
        .isTrue();
    assertThat(listener.suspectReasons.contains(Connection.INITIATING_SUSPECT_PROCESSING))
        .describedAs("expected suspect processing initiated by TCPConduit").isTrue();
  }

  /**
   * test lead member failure and normal coordinator shutdown with network partition detection
   * enabled.
   * <p>
   * Start two locators with admin distributed systems, then start two regular distributed members.
   * <p>
   * We kill the lead member and demonstrate that the other members continue to operate normally.
   * <p>
   * We then shut down the group coordinator and observe the second locator pick up the job and the
   * remaining member continues to operate normally.
   */
  @Test
  public void testLeadFailureAndCoordShutdown() throws Exception {
    String locators = hostName + "[" + port1 + "]," + hostName + "[" + port2 + "]";

    Properties properties = getClusterProperties(locators, "true");
    addDSProps(properties);

    Locator locator = Locator.startLocatorAndDS(port1, null, properties);
    system = (InternalDistributedSystem) locator.getDistributedSystem();

    vm3.invoke(() -> {
      Locator loc = Locator.startLocatorAndDS(port2, null, properties);
      system = (InternalDistributedSystem) loc.getDistributedSystem();
      MembershipManagerHelper.inhibitForcedDisconnectLogging(true);
    });

    assertThat(MembershipManagerHelper.getLeadMember(system)).isNull();

    DistributedMember mem1 = vm1.invoke(() -> getDistributedMember(properties));
    DistributedMember mem2 = vm2.invoke(() -> getDistributedMember(properties));

    assertThat(mem1).isEqualTo(MembershipManagerHelper.getLeadMember(system));

    assertThat(system.getDistributedMember())
        .isEqualTo(MembershipManagerHelper.getCoordinator(system));

    MembershipManagerHelper.inhibitForcedDisconnectLogging(true);

    // crash the lead vm. Should be okay
    vm1.invoke(() -> {
      addIgnoredException("service failure");
      addIgnoredException(ForcedDisconnectException.class);
      MembershipManagerHelper.crashDistributedSystem(system);
    });

    waitUntilTheSystemIsConnected(vm2, vm3);
    // stop the locator normally. This should also be okay
    locator.stop();

    await()
        .until(() -> {
          assertThat(Locator.getLocator()).describedAs("locator is not stopped").isNull();
          return true;
        });

    checkSystemConnectedInVMs(vm2, vm3);

    // the remaining non-locator member should now be the lead member
    assertEquals(
        "This test sometimes fails.  If the log contains "
            + "'failed to collect all ACKs' it is a false failure.",
        mem2, vm2.invoke(LocatorDUnitTest::getLeadMember));

    // disconnect the first vm and demonstrate that the third vm and the
    // locator notice the failure and exit
    vm2.invoke(LocatorDUnitTest::disconnectDistributedSystem);
    vm3.invoke(LocatorDUnitTest::stopLocator);
  }

  /**
   * test lead member failure and normal coordinator shutdown with network partition detection
   * enabled.
   * <p>
   * Start one locators with admin distributed systems, then start two regular distributed members.
   * <p>
   * We kill the lead member and demonstrate that the other members continue to operate normally.
   * <p>
   * We then shut down the group coordinator and observe the second locator pick up the job and the
   * remaining member continues to operate normally.
   */
  @Test
  public void testForceDisconnectAndPeerShutdownCause() throws Exception {
    String locators = hostName + "[" + port1 + "]," + hostName + "[" + port2 + "]";

    Properties properties = new Properties();
    properties.setProperty(MCAST_PORT, "0");
    properties.setProperty(LOCATORS, locators);
    properties.setProperty(ENABLE_NETWORK_PARTITION_DETECTION, "true");
    properties.setProperty(DISABLE_AUTO_RECONNECT, "true");
    properties.setProperty(MEMBER_TIMEOUT, "2000");
    addDSProps(properties);

    Locator locator = Locator.startLocatorAndDS(port1, null, properties);
    system = (InternalDistributedSystem) locator.getDistributedSystem();

    vm3.invoke(() -> {
      Locator loc = Locator.startLocatorAndDS(port2, null, properties);
      system = (InternalDistributedSystem) loc.getDistributedSystem();
    });

    SerializableRunnable crashSystem = new SerializableRunnable("Crash system") {
      @Override
      public void run() {
        addIgnoredException("service failure");
        addIgnoredException("Possible loss of quorum");
        addIgnoredException(ForcedDisconnectException.class);

        hook = new TestHook();
        MembershipManagerHelper.addTestHook(system, hook);
        try {
          MembershipManagerHelper.crashDistributedSystem(system);
        } finally {
          hook.reset();
        }
      }
    };

    assertNull(MembershipManagerHelper.getLeadMember(system));

    DistributedMember mem1 = vm1.invoke(() -> getDistributedMember(properties));
    DistributedMember mem2 = vm2.invoke(() -> getDistributedMember(properties));

    assertEquals(mem1, MembershipManagerHelper.getLeadMember(system));

    assertEquals(system.getDistributedMember(), MembershipManagerHelper.getCoordinator(system));

    assertTrue("Distributed system should not have disconnected", isSystemConnected());

    assertTrue("Distributed system should not have disconnected",
        vm2.invoke(LocatorDUnitTest::isSystemConnected));

    assertTrue("Distributed system should not have disconnected",
        vm3.invoke(LocatorDUnitTest::isSystemConnected));

    vm2.invokeAsync(crashSystem);

    // request member removal for first peer from second peer.
    vm2.invoke(new SerializableRunnable("Request Member Removal") {

      @Override
      public void run() {
        Distribution mmgr = MembershipManagerHelper.getDistribution(system);

        // check for shutdown cause in Distribution. Following call should
        // throw DistributedSystemDisconnectedException which should have cause as
        // ForceDisconnectException.
        await().until(() -> !mmgr.getMembership().isConnected());
        await().untilAsserted(() -> {
          Throwable cause = mmgr.getShutdownCause();
          assertThat(cause).isInstanceOf(ForcedDisconnectException.class);
        });
        try (IgnoredException i = addIgnoredException("Membership: requesting removal of")) {
          mmgr.requestMemberRemoval((InternalDistributedMember) mem1, "test reasons");
          fail("It should have thrown exception in requestMemberRemoval");
        } catch (DistributedSystemDisconnectedException e) {
          // expected
          assertThat(e.getCause()).isInstanceOf(ForcedDisconnectException.class);
        } finally {
          hook.reset();
        }
      }
    });
  }

  /**
   * test lead member shutdown and coordinator crashing with network partition detection enabled.
   * <p>
   * Start two locators with admin distributed systems, then start two regular distributed members.
   * <p>
   * We kill the coordinator and shut down the lead member and observe the second locator pick up
   * the job and the remaining member continue to operate normally.
   */
  @Test
  public void testLeadShutdownAndCoordFailure() throws Exception {
    VM memberThatWillBeShutdownVM = vm1;
    VM memberVM = vm2;
    VM locatorThatWillBeShutdownVM = vm3;

