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

import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatNoException;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.hamcrest.core.IsEqual.equalTo;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.contains;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.isA;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.timeout;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

import java.io.IOException;
import java.net.InetAddress;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Properties;
import java.util.Set;
import java.util.Timer;

import org.junit.After;
import org.junit.Assert;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.mockito.internal.verification.Times;
import org.mockito.verification.Timeout;

import org.apache.geode.distributed.internal.membership.api.Authenticator;
import org.apache.geode.distributed.internal.membership.api.LocatorConfigurationParser;
import org.apache.geode.distributed.internal.membership.api.MemberDataBuilder;
import org.apache.geode.distributed.internal.membership.api.MemberIdentifier;
import org.apache.geode.distributed.internal.membership.api.MemberIdentifierFactoryImpl;
import org.apache.geode.distributed.internal.membership.api.MemberStartupException;
import org.apache.geode.distributed.internal.membership.api.MembershipConfig;
import org.apache.geode.distributed.internal.membership.api.MembershipConfigurationException;
import org.apache.geode.distributed.internal.membership.gms.GMSMembershipView;
import org.apache.geode.distributed.internal.membership.gms.MemberIdentifierImpl;
import org.apache.geode.distributed.internal.membership.gms.Services;
import org.apache.geode.distributed.internal.membership.gms.Services.Stopper;
import org.apache.geode.distributed.internal.membership.gms.interfaces.HealthMonitor;
import org.apache.geode.distributed.internal.membership.gms.interfaces.Locator;
import org.apache.geode.distributed.internal.membership.gms.interfaces.Manager;
import org.apache.geode.distributed.internal.membership.gms.interfaces.Messenger;
import org.apache.geode.distributed.internal.membership.gms.locator.FindCoordinatorRequest;
import org.apache.geode.distributed.internal.membership.gms.locator.FindCoordinatorResponse;
import org.apache.geode.distributed.internal.membership.gms.membership.GMSJoinLeave.SearchState;
import org.apache.geode.distributed.internal.membership.gms.membership.GMSJoinLeave.ViewCreator;
import org.apache.geode.distributed.internal.membership.gms.membership.GMSJoinLeave.ViewReplyProcessor;
import org.apache.geode.distributed.internal.membership.gms.messages.InstallViewMessage;
import org.apache.geode.distributed.internal.membership.gms.messages.JoinRequestMessage;
import org.apache.geode.distributed.internal.membership.gms.messages.JoinResponseMessage;
import org.apache.geode.distributed.internal.membership.gms.messages.LeaveRequestMessage;
import org.apache.geode.distributed.internal.membership.gms.messages.NetworkPartitionMessage;
import org.apache.geode.distributed.internal.membership.gms.messages.RemoveMemberMessage;
import org.apache.geode.distributed.internal.membership.gms.messages.ViewAckMessage;
import org.apache.geode.distributed.internal.membership.gms.util.MemberIdentifierUtil;
import org.apache.geode.distributed.internal.tcpserver.HostAndPort;
import org.apache.geode.distributed.internal.tcpserver.TcpClient;
import org.apache.geode.internal.serialization.KnownVersion;
import org.apache.geode.test.junit.categories.MembershipTest;

@Category({MembershipTest.class})
public class GMSJoinLeaveJUnitTest {
  private Services services;
  private MembershipConfig mockConfig;
  private Authenticator authenticator;
  private HealthMonitor healthMonitor;
  private MemberIdentifier gmsJoinLeaveMemberId;
  private MemberIdentifier[] mockMembers;
  private MemberIdentifier mockOldMember;
  private final Properties credentials = new Properties();
  private Messenger messenger;
  private GMSJoinLeave gmsJoinLeave;
  private Manager manager;
  private Stopper stopper;
  private TestLocator testLocator;
  private MemberIdentifier removeMember = null;
  private MemberIdentifier leaveMember = null;
  private TcpClient locatorClient;

  public void initMocks() throws Exception {
    initMocks(false);
  }

  public void initMocks(boolean enableNetworkPartition) throws Exception {
    initMocks(enableNetworkPartition, false);
  }

  public void initMocks(boolean enableNetworkPartition, boolean useTestGMSJoinLeave)
      throws Exception {
    String locator = "localhost[12345]";
    initMocks(enableNetworkPartition, useTestGMSJoinLeave, locator, locator);
  }

  public void initMocks(boolean enableNetworkPartition, boolean useTestGMSJoinLeave,
      String locators, String startLocator)
      throws Exception {
    mockConfig = mock(MembershipConfig.class);
    when(mockConfig.isNetworkPartitionDetectionEnabled()).thenReturn(enableNetworkPartition);
    when(mockConfig.getSecurityUDPDHAlgo()).thenReturn("");
    when(mockConfig.getStartLocator()).thenReturn(startLocator);
    when(mockConfig.getLocators()).thenReturn(locators);
    when(mockConfig.getMcastPort()).thenReturn(0);
    when(mockConfig.getMemberTimeout()).thenReturn(2000L);

    authenticator = mock(Authenticator.class);

    stopper = mock(Stopper.class);
    when(stopper.isCancelInProgress()).thenReturn(false);

    manager = mock(Manager.class);

    healthMonitor = mock(HealthMonitor.class);
    when(healthMonitor.getFailureDetectionPort()).thenReturn(-1);

    services = mock(Services.class);
    when(services.getAuthenticator()).thenReturn(authenticator);
    when(services.getConfig()).thenReturn(mockConfig);
    when(services.getCancelCriterion()).thenReturn(stopper);
    when(services.getManager()).thenReturn(manager);
    when(services.getHealthMonitor()).thenReturn(healthMonitor);
    when(services.getMemberFactory())
        .thenReturn(new MemberIdentifierFactoryImpl());

    gmsJoinLeaveMemberId = services.getMemberFactory().create(
        MemberDataBuilder.newBuilderForLocalHost("localhost")
            .setMembershipPort(8887).build());


    messenger = mock(Messenger.class);
    when(messenger.getMemberID()).thenReturn(gmsJoinLeaveMemberId);
    when(services.getMessenger()).thenReturn(messenger);

    testLocator = new TestLocator();
    when(services.getLocator()).thenReturn(testLocator);

    Timer t = new Timer(true);
    when(services.getTimer()).thenReturn(t);

    mockMembers = new MemberIdentifier[4];
    for (int i = 0; i < mockMembers.length; i++) {
      mockMembers[i] = services.getMemberFactory().create(
          MemberDataBuilder.newBuilderForLocalHost("localhost")
              .setMembershipPort(8888 + i).build());
    }
    mockOldMember = services.getMemberFactory().create(
        MemberDataBuilder.newBuilderForLocalHost("localhost")
            .setMembershipPort(8700).build());
    mockOldMember.setVersionForTest(KnownVersion.OLDEST);
    locatorClient = mock(TcpClient.class);

    if (useTestGMSJoinLeave) {
      gmsJoinLeave = new GMSJoinLeaveTest(locatorClient);
    } else {
      gmsJoinLeave = new GMSJoinLeave(locatorClient);
    }
    gmsJoinLeave.init(services);
    gmsJoinLeave.start();
    gmsJoinLeave.started();
    gmsJoinLeave.setLocalAddress(gmsJoinLeaveMemberId);
  }

  @After
  public void tearDown() throws Exception {
    if (gmsJoinLeave != null) {
      gmsJoinLeave.stop();
      gmsJoinLeave.stopped();
    }
  }

  static class TestLocator implements Locator {

    boolean isCoordinator;

    @Override
    public void installView(GMSMembershipView v) {}

    @Override
    public void setIsCoordinator(boolean isCoordinator) {
      this.isCoordinator = isCoordinator;
    }

    public boolean isCoordinator() {
      return isCoordinator;
    }

  }

  @Test
  public void testFindCoordinatorPausesWhenLocatorWaitTimeIsSet() throws Exception {
    initMocks(false);
    when(mockConfig.getLocatorWaitTime()).thenReturn(15000);

    when(locatorClient.requestToServer(isA(HostAndPort.class),
        isA(FindCoordinatorRequest.class), anyInt(), anyBoolean()))
            .thenThrow(new IOException("Connection refused"));

    // interrupt this thread so that findCoordinator() won't keep looping
    // and will throw an exception when going to pause
    Thread.currentThread().interrupt();
    assertThatThrownBy(() -> gmsJoinLeave.findCoordinator())
        .isInstanceOf(MemberStartupException.class)
        .hasMessageContaining("Interrupted while trying to contact locators");
    assertThat(Thread.interrupted()).isTrue();
    verify(locatorClient, times(1)).requestToServer(isA(HostAndPort.class),
        isA(FindCoordinatorRequest.class), anyInt(), anyBoolean());
  }

  @Test
  public void testFindCoordinatorInView() throws Exception {
    initMocks();

    int viewId = 1;
    List<MemberIdentifier> mbrs = new LinkedList<>();
    mbrs.add(mockMembers[0]);
    mbrs.add(mockMembers[1]);
    mbrs.add(mockMembers[2]);

    // prepare the view
    GMSMembershipView netView = new GMSMembershipView(mockMembers[0], viewId, mbrs);
    SearchState state = gmsJoinLeave.searchState;
    state.view = netView;
    state.viewId = netView.getViewId();

    MemberIdentifier coordinator = mockMembers[2];
    coordinator.setVmViewId(viewId);

    // already tried joining using members 0 and 1
    Set<MemberIdentifier> set = new HashSet<>();
    mockMembers[0].setVmViewId(viewId - 1);
    set.add(mockMembers[0]);
    mockMembers[1].setVmViewId(viewId - 1);
    set.add(mockMembers[1]);
    state.alreadyTried = set;
    state.hasContactedAJoinedLocator = true;

    // simulate a response being received
    MemberIdentifier sender = mockMembers[2];
    FindCoordinatorResponse resp = new FindCoordinatorResponse(coordinator, sender, null, 0);
    gmsJoinLeave.processFindCoordinatorResponseMessage(resp);
    // tell GMSJoinLeave that a unit test is running so it won't clear the
    // responses collection
    gmsJoinLeave.unitTesting.add("findCoordinatorFromView");

    // now for the test
    boolean result = gmsJoinLeave.findCoordinatorFromView();
    assertTrue("should have found coordinator " + mockMembers[2], result);
    assertTrue("should have found " + coordinator + " but found " + state.possibleCoordinator,
        state.possibleCoordinator == coordinator);
  }

  @Test
  public void testProcessJoinMessageRejectOldMemberVersion() throws Exception {
    initMocks();

    gmsJoinLeave.processJoinRequestMessage(
        new JoinRequestMessage(mockOldMember, mockOldMember, null, -1, 0));
    assertTrue("JoinRequest should not have been added to view request",
        gmsJoinLeave.getViewRequests().size() == 0);
    verify(messenger).send(isA(JoinResponseMessage.class));
  }

  @Test
  public void testViewWithoutMemberInitiatesForcedDisconnect() throws Exception {
    initMocks();
    becomeCoordinatorForTest(gmsJoinLeave);
    List<MemberIdentifier> members = Arrays.asList(mockMembers);
    GMSMembershipView v = new GMSMembershipView(mockMembers[0], 2, members);
    InstallViewMessage message = getInstallViewMessage(v, null, false);
    gmsJoinLeave.processInstallViewMessage(message);
    verify(manager).forceDisconnect(isA(String.class));
  }


  @Test
  public void testProcessJoinMessageWithBadAuthentication() throws Exception {
    initMocks();
    when(authenticator.authenticate(mockMembers[0], credentials))
        .thenThrow(new SecurityException("we want to fail auth here"));

    gmsJoinLeave
        .processJoinRequestMessage(
            new JoinRequestMessage(mockMembers[0], mockMembers[0], credentials, -1, 0));
    assertTrue("JoinRequest should not have been added to view request",
        gmsJoinLeave.getViewRequests().size() == 0);
    verify(messenger).send(isA(JoinResponseMessage.class));
  }

  @Test
  public void testProcessJoinMessageWithAuthenticationButNullCredentials() throws Exception {
    initMocks();
    when(authenticator.authenticate(mockMembers[0], null))
        .thenThrow(new SecurityException("we want to fail auth here"));

    gmsJoinLeave
        .processJoinRequestMessage(
            new JoinRequestMessage(mockMembers[0], mockMembers[0], null, -1, 0));
    assertTrue("JoinRequest should not have been added to view request",
        gmsJoinLeave.getViewRequests().size() == 0);
    verify(messenger).send(isA(JoinResponseMessage.class));
  }

  // This test does not test the actual join process but rather that the join response gets logged
  @Test
  public void testProcessJoinResponseIsRecorded() throws Exception {
    initMocks();
    when(authenticator.authenticate(mockMembers[0], null))
        .thenThrow(new SecurityException("we want to fail auth here"));

    JoinResponseMessage[] joinResponse = gmsJoinLeave.getJoinResponseMessage();

    JoinResponseMessage jrm = new JoinResponseMessage(mockMembers[0], new byte[9], 233);
    gmsJoinLeave.processJoinResponseMessage(jrm);
    // this should NOT logs, this is just to inform member succesful joining
    Assert.assertEquals(null, joinResponse[0]);

    jrm = new JoinResponseMessage("rejected...", 0);
    gmsJoinLeave.processJoinResponseMessage(jrm);
    // this should log..
    Assert.assertEquals(jrm, joinResponse[0]);
  }

  /**
   * prepares and install a view
   *
   */
  private void prepareAndInstallView(MemberIdentifier coordinator,
      List<MemberIdentifier> members) throws IOException {
    int viewId = 1;

    // prepare the view
    GMSMembershipView<MemberIdentifier> netView =
        new GMSMembershipView(coordinator, viewId, members);
    for (MemberIdentifier member : netView.getMembers()) {
      netView.setPublicKey(member, member.toString());
    }
    InstallViewMessage installViewMessage = getInstallViewMessage(netView, credentials, true);
    gmsJoinLeave.processInstallViewMessage(installViewMessage);
    verify(messenger).send(isA(ViewAckMessage.class));

    // install the view
    installViewMessage = getInstallViewMessage(netView, credentials, false);
    gmsJoinLeave.processInstallViewMessage(installViewMessage);
    Assert.assertEquals(netView, gmsJoinLeave.getView());
  }

  private List<MemberIdentifier> createMemberList(MemberIdentifier... members) {
    List<MemberIdentifier> memberList =
        new ArrayList<>(members.length);
    for (MemberIdentifier member : members) {
      memberList.add(member);
    }
    return memberList;
  }

  @Test
  public void testRemoveMember() throws Exception {
    initMocks();
    prepareAndInstallView(mockMembers[0], createMemberList(mockMembers[0], gmsJoinLeaveMemberId));
    gmsJoinLeave.remove(gmsJoinLeaveMemberId, "removing for test");
    verify(messenger, timeout(2000).atLeastOnce()).send(isA(RemoveMemberMessage.class));
  }

  @Test
  public void testIsMemberLeaving() throws Exception {
    initMocks();
    prepareAndInstallView(mockMembers[0],
        createMemberList(mockMembers[0], mockMembers[1], gmsJoinLeaveMemberId));
    assertFalse(gmsJoinLeave.isMemberLeaving(mockMembers[0]));
    assertFalse(gmsJoinLeave.isMemberLeaving(mockMembers[1]));
    gmsJoinLeave.remove(mockMembers[0], "removing for test");
    verify(messenger, timeout(2000).atLeastOnce()).send(isA(RemoveMemberMessage.class));
    assertTrue(gmsJoinLeave.isMemberLeaving(mockMembers[0]));
    LeaveRequestMessage msg =
        new LeaveRequestMessage(gmsJoinLeave.getMemberID(), mockMembers[1], "leaving for test");
    msg.setSender(mockMembers[1]);
    gmsJoinLeave.processLeaveRequestMessage(msg);
    assertTrue(gmsJoinLeave.isMemberLeaving(mockMembers[1]));
  }

  @Test
  public void testRemoveAndLeaveIsNotACrash() throws Exception {
    // simultaneous leave & remove requests for a member
    // should not result in it's being seen as a crashed member
    initMocks();
    final int viewInstallationTime = 15000;

    when(healthMonitor.checkIfAvailable(isA(MemberIdentifier.class), isA(String.class),
        isA(Boolean.class))).thenReturn(true);

    gmsJoinLeave.delayViewCreationForTest(5000); // ensures multiple requests are queued for a view
                                                 // change
    becomeCoordinatorForTest(gmsJoinLeave);

    await()
        .until(() -> gmsJoinLeave.getView() != null);

    GMSMembershipView oldView = gmsJoinLeave.getView();

    GMSMembershipView newView = new GMSMembershipView(oldView, oldView.getViewId() + 1);
    newView.add(mockMembers[1]);
    newView.add(mockMembers[2]);
    gmsJoinLeave.installView(newView);

    gmsJoinLeave.memberShutdown(mockMembers[1], "shutting down for test");
    gmsJoinLeave.remove(mockMembers[1], "removing for test");

    await()
        .until(() -> gmsJoinLeave.getView().getViewId() > newView.getViewId());
    assertFalse(gmsJoinLeave.getView().getCrashedMembers().contains(mockMembers[1]));
  }

  @Test
  public void multipleLocatorsWithSameAddressAreCanonicalized() throws Exception {
    List<HostAndPort> locators = LocatorConfigurationParser.parseLocators(
        "localhost[1234],localhost[1234],localhost[1234]", (InetAddress) null);
    assertThat(locators.size()).isEqualTo(1);
  }


  @Test
  public void testRejectOlderView() throws Exception {
    initMocks();
    prepareAndInstallView(mockMembers[0], createMemberList(mockMembers[0], gmsJoinLeaveMemberId));

    List<MemberIdentifier> mbrs = new LinkedList<>();
    mbrs.add(mockMembers[0]);
    mbrs.add(mockMembers[1]);

    // try to install an older view where viewId < currentView.viewId
    GMSMembershipView olderNetView = new GMSMembershipView(mockMembers[0], 0, mbrs);
    InstallViewMessage installViewMessage = getInstallViewMessage(olderNetView, credentials, false);
    gmsJoinLeave.processInstallViewMessage(installViewMessage);
    Assert.assertNotEquals(gmsJoinLeave.getView(), olderNetView);
  }

  @Test
  public void testForceDisconnectedFromNewView() throws Exception {
    initMocks(true);// enabledNetworkPartition;
    Manager mockManager = mock(Manager.class);
    when(services.getManager()).thenReturn(mockManager);
    prepareAndInstallView(mockMembers[0], createMemberList(mockMembers[0], gmsJoinLeaveMemberId));

    int viewId = 2;
    List<MemberIdentifier> mbrs = new LinkedList<>();
    mbrs.add(mockMembers[1]);
    mbrs.add(mockMembers[2]);
    mbrs.add(mockMembers[3]);

    // install the view
    GMSMembershipView netView = new GMSMembershipView(mockMembers[0], viewId, mbrs);
    InstallViewMessage installViewMessage = getInstallViewMessage(netView, credentials, false);
    gmsJoinLeave.processInstallViewMessage(installViewMessage);

    Assert.assertNotEquals(netView, gmsJoinLeave.getView());
    verify(mockManager).forceDisconnect(isA(String.class));
  }

  @Test
  public void testNonMemberCantRemoveMember() throws Exception {
    String reason = "testing";
    initMocks();
    prepareAndInstallView(mockMembers[0], createMemberList(mockMembers[0], gmsJoinLeaveMemberId));
    // test that a non-member can't remove another member
    RemoveMemberMessage msg = new RemoveMemberMessage(mockMembers[0], mockMembers[1], reason);
    msg.setSender(MemberIdentifierUtil.createMemberID(9000));
    gmsJoinLeave.processRemoveMemberMessage(msg);
    assertTrue("RemoveMemberMessage should not have been added to view requests",
        gmsJoinLeave.getViewRequests().size() == 0);
  }

  @Test
  public void testDuplicateLeaveRequestDoesNotCauseNewView() throws Exception {
    String reason = "testing";
    initMocks();
    gmsJoinLeave.unitTesting.add("noRandomViewChange");
    prepareAndInstallView(gmsJoinLeaveMemberId,
        createMemberList(gmsJoinLeaveMemberId, mockMembers[0]));
    becomeCoordinatorForTest(gmsJoinLeave);

    LeaveRequestMessage msg =
        new LeaveRequestMessage(gmsJoinLeave.getMemberID(), mockMembers[0], reason);
    msg.setSender(mockMembers[0]);
    gmsJoinLeave.processLeaveRequestMessage(msg);
    msg = new LeaveRequestMessage(gmsJoinLeave.getMemberID(), mockMembers[0], reason);
    msg.setSender(mockMembers[0]);
    gmsJoinLeave.processLeaveRequestMessage(msg);

    waitForViewAndNoRequestsInProgress(7);

    GMSMembershipView view = gmsJoinLeave.getView();
    assertTrue("expected member to be removed: " + mockMembers[0] + "; view: " + view,
        !view.contains(mockMembers[0]));
    assertTrue("expected member to be in shutdownMembers collection: " + mockMembers[0] + "; view: "
        + view, view.getShutdownMembers().contains(mockMembers[0]));
  }

  @Test
  public void testDuplicateRemoveRequestDoesNotCauseNewView() throws Exception {
    String reason = "testing";
    initMocks();
    prepareAndInstallView(gmsJoinLeaveMemberId,
        createMemberList(gmsJoinLeaveMemberId, mockMembers[0]));
    gmsJoinLeave.getView().add(mockMembers[1]);
    gmsJoinLeave.unitTesting.add("noRandomViewChange");
    becomeCoordinatorForTest(gmsJoinLeave);
    RemoveMemberMessage msg =
        new RemoveMemberMessage(gmsJoinLeave.getMemberID(), mockMembers[0], reason);
    msg.setSender(mockMembers[0]);
    gmsJoinLeave.processRemoveMemberMessage(msg);
    msg = new RemoveMemberMessage(gmsJoinLeave.getMemberID(), mockMembers[0], reason);
    msg.setSender(mockMembers[0]);
    gmsJoinLeave.processRemoveMemberMessage(msg);

    waitForViewAndNoRequestsInProgress(7);

    GMSMembershipView view = gmsJoinLeave.getView();
    assertTrue("expected member to be removed: " + mockMembers[0] + "; view: " + view,
        !view.contains(mockMembers[0]));
    assertTrue(
        "expected member to be in crashedMembers collection: " + mockMembers[0] + "; view: " + view,
        view.getCrashedMembers().contains(mockMembers[0]));
  }

  @Test
  public void testDuplicateJoinRequestDoesNotCauseNewView() throws Exception {
    initMocks();
    when(healthMonitor.checkIfAvailable(isA(MemberIdentifier.class), isA(String.class),
        isA(Boolean.class), isA(Boolean.class))).thenReturn(true);

    gmsJoinLeave.unitTesting.add("noRandomViewChange");
    prepareAndInstallView(gmsJoinLeaveMemberId,
        createMemberList(gmsJoinLeaveMemberId, mockMembers[0]));
    gmsJoinLeave.getView().add(mockMembers[1]);
    becomeCoordinatorForTest(gmsJoinLeave);
    JoinRequestMessage msg =
        new JoinRequestMessage(gmsJoinLeaveMemberId, mockMembers[2], null, -1, 0);
    msg.setSender(mockMembers[2]);
    gmsJoinLeave.processJoinRequestMessage(msg);
    msg = new JoinRequestMessage(gmsJoinLeaveMemberId, mockMembers[2], null, -1, 0);
    msg.setSender(mockMembers[2]);
    gmsJoinLeave.processJoinRequestMessage(msg);

    waitForViewAndNoRequestsInProgress(7);

    GMSMembershipView view = gmsJoinLeave.getView();
    assertTrue("expected member to be added: " + mockMembers[2] + "; view: " + view,
        view.contains(mockMembers[2]));
    List<MemberIdentifier> members = view.getMembers();
    int occurrences = 0;
    for (MemberIdentifier mbr : members) {
      if (mbr.equals(mockMembers[2])) {
        occurrences += 1;
      }
    }
    assertTrue("expected member to only be in the view once: " + mockMembers[2] + "; view: " + view,
        occurrences == 1);
    verify(healthMonitor, times(5)).checkIfAvailable(isA(MemberIdentifier.class),
        isA(String.class), isA(Boolean.class), isA(Boolean.class));
  }


  private void waitForViewAndNoRequestsInProgress(int viewId) throws InterruptedException {
    // wait for the view processing thread to collect and process the requests
    int sleeps = 0;
    while (!gmsJoinLeave.isStopping() && !gmsJoinLeave.getViewCreator().isWaiting()
        && (!gmsJoinLeave.getViewRequests().isEmpty()
            || gmsJoinLeave.getView().getViewId() != viewId)) {
      if (sleeps++ > 20) {
        throw new RuntimeException("timeout waiting for view #" + viewId + " current view: "
            + gmsJoinLeave.getView() + "; view requests: " + gmsJoinLeave.getViewRequests());
      }
      Thread.sleep(1000);
    }
  }

  @Test
  public void testRemoveMessageForRogueCausesImmediateRemovalMessageToRogue() throws Exception {
    initMocks();
    gmsJoinLeave.currentView = createView();
    gmsJoinLeave.isCoordinator = true;
    RemoveMemberMessage msg = new RemoveMemberMessage(gmsJoinLeaveMemberId,
        MemberIdentifierUtil.createMemberID(10000), "removing for test");
    msg.setSender(mockMembers[0]);
    gmsJoinLeave.processRemoveMemberMessage(msg);
    verify(messenger).send(isA(RemoveMemberMessage.class));
  }

  @Test
  public void testRemoveRequestCausesForcedDisconnectInRogue() throws Exception {
    initMocks();
    // gmsJoinLeave mistakenly uses an old viewID when joining, making it a rogue member
    gmsJoinLeaveMemberId.setVmViewId(-1);
    MemberIdentifier previousMemberId = services.getMemberFactory().create(
        MemberDataBuilder.newBuilder(gmsJoinLeaveMemberId.getInetAddress(),
            gmsJoinLeaveMemberId.getHostName())
            .setMembershipPort(gmsJoinLeaveMemberId.getMembershipPort())
            .build());
    previousMemberId.setVmViewId(0);
    previousMemberId.setUUID(gmsJoinLeaveMemberId.getUUID());
    GMSMembershipView view = new GMSMembershipView(mockMembers[0], 1,
        createMemberList(mockMembers[0], previousMemberId, mockMembers[1]));
    InstallViewMessage viewMessage = new InstallViewMessage(view, 0, false);
    viewMessage.setSender(mockMembers[0]);
    gmsJoinLeave.processInstallViewMessage(viewMessage);
    assertEquals(0, gmsJoinLeaveMemberId.getVmViewId());
    // a RemoveMember message should cause it to force-disconnect
    RemoveMemberMessage msg =
        new RemoveMemberMessage(gmsJoinLeaveMemberId, gmsJoinLeaveMemberId, "removing for test");
    msg.setSender(mockMembers[0]);
    gmsJoinLeave.processRemoveMemberMessage(msg);
    verify(manager).forceDisconnect("removing for test");
  }

  @Test
  public void testRemoveCausesForcedDisconnect() throws Exception {
    String reason = "testing";
    initMocks();
    prepareAndInstallView(mockMembers[0], createMemberList(mockMembers[0], gmsJoinLeaveMemberId));
    gmsJoinLeave.getView().add(mockMembers[1]);
    RemoveMemberMessage msg =
        new RemoveMemberMessage(mockMembers[0], gmsJoinLeave.getMemberID(), reason);
    msg.setSender(mockMembers[1]);
    gmsJoinLeave.processRemoveMemberMessage(msg);
    verify(manager).forceDisconnect(reason);
  }

  @Test
  public void testLeaveCausesForcedDisconnect() throws Exception {
    String reason = "testing";
    initMocks();
    prepareAndInstallView(gmsJoinLeaveMemberId,
        createMemberList(gmsJoinLeaveMemberId, mockMembers[0]));
    gmsJoinLeave.getView().add(mockMembers[1]);
    LeaveRequestMessage msg =
        new LeaveRequestMessage(gmsJoinLeave.getMemberID(), gmsJoinLeave.getMemberID(), reason);
    msg.setSender(mockMembers[1]);
    gmsJoinLeave.processLeaveRequestMessage(msg);
    verify(manager).forceDisconnect(reason);
  }

  @Test
  public void testLeaveOfNonMemberIsNoOp() throws Exception {
    String reason = "testing";
    initMocks();
    prepareAndInstallView(mockMembers[0], createMemberList(mockMembers[0], gmsJoinLeaveMemberId));
    mockMembers[1].setVmViewId(gmsJoinLeave.getView().getViewId() - 1);
    LeaveRequestMessage msg =
        new LeaveRequestMessage(gmsJoinLeave.getMemberID(), mockMembers[1], reason);
    msg.setSender(mockMembers[1]);
    gmsJoinLeave.processLeaveRequestMessage(msg);
    assertTrue("Expected leave request from non-member to be ignored",
        gmsJoinLeave.getViewRequests().isEmpty());
  }

  @Test
  public void testBecomeCoordinatorOnStartup() throws Exception {
    initMocks();
    becomeCoordinatorForTest(gmsJoinLeave);
    await().until(() -> gmsJoinLeave.isCoordinator());
  }

  @Test
  public void testBecomeCoordinator() throws Exception {
    String reason = "testing";
    initMocks();
    prepareAndInstallView(mockMembers[0], createMemberList(mockMembers[0], gmsJoinLeaveMemberId));
    GMSMembershipView view = gmsJoinLeave.getView();
    view.add(gmsJoinLeaveMemberId);
    MemberIdentifier creator = view.getCreator();
    LeaveRequestMessage msg = new LeaveRequestMessage(creator, creator, reason);
    msg.setSender(creator);
    gmsJoinLeave.processLeaveRequestMessage(msg);
    assertTrue("Expected becomeCoordinator to be invoked", gmsJoinLeave.isCoordinator());
  }

  @Test
  public void testBecomeCoordinatorThroughRemove() throws Exception {
    String reason = "testing";
    initMocks();
    prepareAndInstallView(mockMembers[0], createMemberList(mockMembers[0], gmsJoinLeaveMemberId));
    GMSMembershipView view = gmsJoinLeave.getView();
    view.add(gmsJoinLeaveMemberId);
    MemberIdentifier creator = view.getCreator();
    RemoveMemberMessage msg = new RemoveMemberMessage(creator, creator, reason);
    msg.setSender(creator);
    gmsJoinLeave.processRemoveMemberMessage(msg);
    assertTrue("Expected becomeCoordinator to be invoked", gmsJoinLeave.isCoordinator());
  }

  /**
   * Given a view with [A, B, C, D, E] where C is coordinator, A failed availability checks and
   * C shuts down we should see B become the coordinator.
   */
  @Test
  public void testBecomeCoordinatorThroughShutdownWhenOlderMemberCrashed() throws Exception {
    initMocks();
    MemberIdentifier A = mockMembers[0],
        B = gmsJoinLeaveMemberId,
        C = mockMembers[1],
        D = mockMembers[2],
        E = mockMembers[3];
    prepareAndInstallView(C, createMemberList(A, B, C, D, E));
    LeaveRequestMessage msg = new LeaveRequestMessage(B, C, "leaving for test");
    msg.setSender(C);
    gmsJoinLeave.processLeaveRequestMessage(msg);
    RemoveMemberMessage removeMemberMessage = new RemoveMemberMessage(B, A, "removing for test");
    removeMemberMessage.setSender(B);
    gmsJoinLeave.processRemoveMemberMessage(removeMemberMessage);
    assertTrue("Expected becomeCoordinator to be invoked", gmsJoinLeave.isCoordinator());
  }

  /**
   * Given a view with [A, B, C, D] where C is coordinator, a failed availability check on A
   * and a pending join request from E show that B becomes coordinator when C sends it a
   * Leave request and that E is allowed into the system and retains its view ID.
   * See GEODE-6570
   */
  @Test
  public void testBecomeCoordinatorAndAcceptMemberWithViewID() throws Exception {
    initMocks();
    MemberIdentifier A = mockMembers[0],
        B = gmsJoinLeaveMemberId,
        C = mockMembers[1],
        D = mockMembers[2],
        E = mockMembers[3];

    prepareAndInstallView(C, createMemberList(A, B, C, D));
    messengerAckAllInstallViewMessages();

    E.setVmViewId(2);

    gmsJoinLeave.recordViewRequest(new LeaveRequestMessage(B, C, "removing for test"));

    gmsJoinLeave.processJoinRequestMessage(new JoinRequestMessage(B, E, null, 1, 1));

    RemoveMemberMessage msg = new RemoveMemberMessage(B, A, "crashed for test");
    msg.setSender(D);
    gmsJoinLeave.processRemoveMemberMessage(msg);

    await().until(() -> gmsJoinLeave.isCoordinator() && gmsJoinLeave.getViewRequests().isEmpty());

    // E should have joined and retained its view ID of 2
    await().until(() -> gmsJoinLeave.getView().contains(E));
    assertEquals(2, E.getVmViewId());
  }

