You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by bs...@apache.org on 2015/08/21 22:20:30 UTC
incubator-geode git commit: GEODE-77 adding new unit tests for
GMSJoinLeave
Repository: incubator-geode
Updated Branches:
refs/heads/feature/GEODE-77 62b6ce15e -> b8f40f4c1
GEODE-77 adding new unit tests for GMSJoinLeave
This adds a bunch of new unit tests to GMSJoinLeaveJUnitTest and
fixes a number of problems exposed by the new tests.
Members were in the view multiple times.
Members were in the shutdownMembers collection multiple times.
Members could be in the crashedMembers collection multiple times.
View preparation was actually installing the view in the coordinator.
A rogue process could request removal of a member.
Project: http://git-wip-us.apache.org/repos/asf/incubator-geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-geode/commit/b8f40f4c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-geode/tree/b8f40f4c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-geode/diff/b8f40f4c
Branch: refs/heads/feature/GEODE-77
Commit: b8f40f4c1205fe56ab823e50dad962ddd29089d3
Parents: 62b6ce1
Author: Bruce Schuchardt <bs...@pivotal.io>
Authored: Fri Aug 21 13:12:47 2015 -0700
Committer: Bruce Schuchardt <bs...@pivotal.io>
Committed: Fri Aug 21 13:20:10 2015 -0700
----------------------------------------------------------------------
.../cache/util/BoundedLinkedHashMap.java | 3 +
.../distributed/internal/InternalLocator.java | 4 +-
.../internal/membership/gms/GMSUtil.java | 21 ++
.../internal/membership/gms/Services.java | 5 +
.../membership/gms/membership/GMSJoinLeave.java | 106 +++++--
.../gms/messages/LeaveRequestMessage.java | 7 +
.../gms/messenger/JGroupsMessenger.java | 34 +--
.../gms/mgr/GMSMembershipManager.java | 91 ++++--
.../membership/MembershipJUnitTest.java | 126 ++++-----
.../membership/gms/MembershipManagerHelper.java | 2 +-
.../gms/membership/GMSJoinLeaveJUnitTest.java | 279 ++++++++++++++++++-
11 files changed, 534 insertions(+), 144 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/b8f40f4c/gemfire-core/src/main/java/com/gemstone/gemfire/cache/util/BoundedLinkedHashMap.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/util/BoundedLinkedHashMap.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/util/BoundedLinkedHashMap.java
index 1abf24f..db15f46 100755
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/util/BoundedLinkedHashMap.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/util/BoundedLinkedHashMap.java
@@ -17,6 +17,9 @@ import java.util.Map;
* can contain.
*
* @since 4.2
+ * @deprecated as of 5.7 create your own class that extends {@link LinkedHashMap}
+ * and implement {@link LinkedHashMap#removeEldestEntry}
+ * to enforce a maximum number of entries.
*/
public class BoundedLinkedHashMap extends LinkedHashMap
{
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/b8f40f4c/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/InternalLocator.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/InternalLocator.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/InternalLocator.java
index 32dee46..cfda513 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/InternalLocator.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/InternalLocator.java
@@ -331,6 +331,9 @@ public class InternalLocator extends Locator implements ConnectListener {
* Creates a distribution locator that runs in this VM on the given
* port and bind address.
*
+ * This is for internal use only as it does not create a distributed
+ * system unless told to do so.
+ *
* @param port
* the tcp/ip port to listen on
* @param logFile
@@ -351,7 +354,6 @@ public class InternalLocator extends Locator implements ConnectListener {
* the name to give to clients for connecting to this locator
* @param loadSharedConfigFromDir TODO:CONFIG
* @throws IOException
- * @deprecated as of 7.0 use startLocator(int, File, File, InternalLogWriter, InternalLogWriter, InetAddress, java.util.Properties, boolean, boolean, String) instead.
*/
public static InternalLocator startLocator(
int port,
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/b8f40f4c/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/GMSUtil.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/GMSUtil.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/GMSUtil.java
index 9e6922e..17a6721 100755
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/GMSUtil.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/GMSUtil.java
@@ -79,4 +79,25 @@ public class GMSUtil {
return result;
}
+ /**
+ * replaces all occurrences of a given string in the properties argument with the
+ * given value
+ */
+ public static String replaceStrings(String properties, String property, String value) {
+ StringBuffer sb = new StringBuffer();
+ int start = 0;
+ int index = properties.indexOf(property);
+ while (index != -1) {
+ sb.append(properties.substring(start, index));
+ sb.append(value);
+
+ start = index + property.length();
+ index = properties.indexOf(property, start);
+ }
+ sb.append(properties.substring(start));
+ return sb.toString();
+ }
+
+
+
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/b8f40f4c/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/Services.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/Services.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/Services.java
index 5f63a9a..aa7bc1e 100755
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/Services.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/Services.java
@@ -49,6 +49,7 @@ public class Services {
final private ServiceConfig config;
final private DMStats stats;
final private Stopper cancelCriterion;
+ private volatile boolean stopping;
private InternalLogWriter logWriter;
private InternalLogWriter securityLogWriter;
@@ -152,6 +153,10 @@ public class Services {
public void stop() {
logger.info("Membership: stopping services");
+ if (stopping) {
+ return;
+ }
+ stopping = true;
this.joinLeave.stop();
this.healthMon.stop();
this.auth.stop();
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/b8f40f4c/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/membership/GMSJoinLeave.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/membership/GMSJoinLeave.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/membership/GMSJoinLeave.java
index d3075bb..9450091 100755
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/membership/GMSJoinLeave.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/membership/GMSJoinLeave.java
@@ -287,17 +287,26 @@ logger.info("received join response {}", response);
private void processLeaveRequest(LeaveRequestMessage incomingRequest) {
logger.info("received leave request from {} for {}", incomingRequest.getSender(), incomingRequest.getMemberID());
-
+
+
NetView v = currentView;
+ InternalDistributedMember mbr = incomingRequest.getMemberID();
+
if (logger.isDebugEnabled()) {
logger.debug("JoinLeave.processLeaveRequest invoked. isCoordinator="+isCoordinator+ "; isStopping="+isStopping
+"; cancelInProgress="+services.getCancelCriterion().isCancelInProgress());
}
+
+ if (!v.contains(mbr) && mbr.getVmViewId() < v.getViewId()) {
+ logger.debug("ignoring leave request from old member");
+ return;
+ }
if (incomingRequest.getMemberID().equals(this.localAddress)) {
logger.info("I am being told to leave the distributed system");
- services.getManager().forceDisconnect(incomingRequest.getReason());
+ forceDisconnect(incomingRequest.getReason());
}
+
if (!isCoordinator && !isStopping && !services.getCancelCriterion().isCancelInProgress()) {
logger.debug("JoinLeave is checking to see if I should become coordinator");
NetView check = new NetView(v, v.getViewId()+1);
@@ -339,7 +348,7 @@ logger.info("received join response {}", response);
if (mbr.equals(this.localAddress)) {
// oops - I've been kicked out
- services.getManager().forceDisconnect(incomingRequest.getReason());
+ forceDisconnect(incomingRequest.getReason());
return;
}
@@ -367,17 +376,22 @@ logger.info("received join response {}", response);
}
}
- //for testing purposes, returns a copy of the view requests for verification
+ // for testing purposes, returns a copy of the view requests for verification
List<DistributionMessage> getViewRequests() {
synchronized(viewRequests) {
return new LinkedList<DistributionMessage>(viewRequests);
}
}
+ // for testing purposes, returns the view-creation thread
+ ViewCreator getViewCreator() {
+ return viewCreator;
+ }
+
/**
* Yippeee - I get to be the coordinator
*/
- private void becomeCoordinator() {
+ void becomeCoordinator() { // package access for unit testing
becomeCoordinator(null);
}
@@ -405,7 +419,9 @@ logger.info("received join response {}", response);
synchronized(viewInstallationLock) {
int viewNumber = currentView.getViewId() + 5;
List<InternalDistributedMember> mbrs = new ArrayList<InternalDistributedMember>(currentView.getMembers());
- mbrs.add(localAddress);
+ if (!mbrs.contains(localAddress)) {
+ mbrs.add(localAddress);
+ }
List<InternalDistributedMember> leaving = new ArrayList<InternalDistributedMember>();
if (oldCoordinator != null) {
leaving.add(oldCoordinator);
@@ -452,17 +468,26 @@ logger.info("received join response {}", response);
int id = view.getViewId();
InstallViewMessage msg = new InstallViewMessage(view, services.getAuthenticator().getCredentials(this.localAddress), preparing);
Set<InternalDistributedMember> recips = new HashSet<InternalDistributedMember>(view.getMembers());
+
recips.removeAll(newMembers); // new members get the view in a JoinResponseMessage
recips.remove(this.localAddress); // no need to send it to ourselves
- installView(view);
+
recips.addAll(view.getCrashedMembers());
+
+ logger.info((preparing? "preparing" : "sending") + " new view " + view);
+
+ if (preparing) {
+ this.preparedView = view;
+ } else {
+ installView(view);
+ }
+
if (recips.isEmpty()) {
return true;
}
+
msg.setRecipients(recips);
rp.initialize(id, recips);
-
- logger.info((preparing? "preparing" : "sending") + " new view " + view);
services.getMessenger().send(msg);
// only wait for responses during preparation
@@ -512,7 +537,7 @@ logger.info("received join response {}", response);
else { // !preparing
if (currentView != null && !view.contains(this.localAddress)) {
if (quorumRequired) {
- services.getManager().forceDisconnect("This node is no longer in the membership view");
+ forceDisconnect("This node is no longer in the membership view");
}
}
else {
@@ -522,6 +547,11 @@ logger.info("received join response {}", response);
}
}
+ private void forceDisconnect(String reason) {
+ this.isStopping = true;
+ services.getManager().forceDisconnect(reason);
+ }
+
private void ackView(InstallViewMessage m) {
if (m.getView().contains(m.getView().getCreator())) {
@@ -622,13 +652,16 @@ logger.info("received join response {}", response);
}
public void installView(NetView newView) {
+
+ logger.info("received new view: {}", newView);
+
synchronized(viewInstallationLock) {
if (currentView != null && currentView.getViewId() >= newView.getViewId()) {
// old view - ignore it
return;
}
- if (checkForPartition(newView)) {
+ if (isNetworkPartition(newView)) {
if (quorumRequired) {
List<InternalDistributedMember> crashes = newView.getActualCrashedMembers(currentView);
services.getManager().forceDisconnect(
@@ -680,11 +713,33 @@ logger.info("received join response {}", response);
}
}
+ /**
+ * returns true if this member thinks it is the membership coordinator
+ * for the distributed system
+ */
+ public boolean isCoordinator() {
+ return this.isCoordinator;
+ }
+
+ /**
+ * return true if we're stopping or are stopped
+ */
+ public boolean isStopping() {
+ return this.isStopping;
+ }
+
+ /**
+ * returns the currently prepared view, if any
+ */
+ public NetView getPreparedView() {
+ return this.preparedView;
+ }
+
/**
* check to see if the new view shows a drop of 51% or more
*/
- private boolean checkForPartition(NetView newView) {
+ private boolean isNetworkPartition(NetView newView) {
if (currentView == null) {
return false;
}
@@ -696,6 +751,8 @@ logger.info("received join response {}", response);
}
int failurePoint = (int)(Math.round(51 * oldWeight) / 100.0);
if (failedWeight > failurePoint) {
+ logger.warn("total weight lost in this view change is {} of {}. Quorum has been lost!",
+ failedWeight, oldWeight);
services.getManager().quorumLost(newView.getActualCrashedMembers(currentView), currentView);
return true;
}
@@ -805,6 +862,7 @@ logger.info("received join response {}", response);
@Override
public void remove(InternalDistributedMember m, String reason) {
NetView v = this.currentView;
+
if (v != null) {
RemoveMemberMessage msg = new RemoveMemberMessage(v.getCoordinator(), m,
reason);
@@ -980,6 +1038,7 @@ logger.info("received join response {}", response);
class ViewCreator extends Thread {
boolean shutdown = false;
+ volatile boolean waiting = false;
ViewCreator(String name, ThreadGroup tg) {
super(tg, name);
@@ -996,6 +1055,10 @@ logger.info("received join response {}", response);
boolean isShutdown() {
return shutdown;
}
+
+ boolean isWaiting() {
+ return waiting;
+ }
@Override
public void run() {
@@ -1011,9 +1074,12 @@ logger.info("received join response {}", response);
if (viewRequests.isEmpty()) {
try {
logger.debug("View Creator is waiting for requests");
+ waiting = true;
viewRequests.wait();
} catch (InterruptedException e) {
return;
+ } finally {
+ waiting = false;
}
} else {
if (System.currentTimeMillis() < okayToCreateView) {
@@ -1072,25 +1138,24 @@ logger.info("received join response {}", response);
if (msg instanceof JoinRequestMessage) {
mbr = ((JoinRequestMessage)msg).getMemberID();
- if (!oldMembers.contains(mbr)) {
+ if (!oldMembers.contains(mbr) && !joinReqs.contains(mbr)) {
joinReqs.add(mbr);
}
}
else if (msg instanceof LeaveRequestMessage) {
mbr = ((LeaveRequestMessage) msg).getMemberID();
- if (oldMembers.contains(mbr)) {
+ if (oldMembers.contains(mbr) && !leaveReqs.contains(mbr)) {
leaveReqs.add(mbr);
}
}
else if (msg instanceof RemoveMemberMessage) {
mbr = ((RemoveMemberMessage) msg).getMemberID();
- if (oldMembers.contains(mbr)) {
+ if (oldMembers.contains(mbr) && !leaveReqs.contains(mbr) && !removalReqs.contains(mbr)) {
removalReqs.add(mbr);
removalReasons.add(((RemoveMemberMessage) msg).getReason());
}
}
else {
- // TODO: handle removals
logger.warn("Unknown membership request encountered: {}", msg);
}
}
@@ -1119,9 +1184,16 @@ logger.info("received join response {}", response);
// getting messages from members that have been kicked out
sendRemoveMessages(removalReqs, removalReasons, newView);
+ // if there are no membership changes then abort creation of
+ // the new view
+ if (newView.getMembers().equals(currentView.getMembers())) {
+ logger.info("membership hasn't changed - aborting new view {}", newView);
+ return true;
+ }
+
// we want to always check for quorum loss but don't act on it
// unless network-partition-detection is enabled
- if ( !(checkForPartition(newView) && quorumRequired) ) {
+ if ( !(isNetworkPartition(newView) && quorumRequired) ) {
sendJoinResponses(joinReqs, newView);
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/b8f40f4c/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messages/LeaveRequestMessage.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messages/LeaveRequestMessage.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messages/LeaveRequestMessage.java
index 61183a8..4cadf19 100755
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messages/LeaveRequestMessage.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messages/LeaveRequestMessage.java
@@ -18,6 +18,7 @@ public class LeaveRequestMessage extends HighPriorityDistributionMessage {
super();
setRecipient(coord);
this.memberID = id;
+ this.reason = reason;
}
public LeaveRequestMessage() {
@@ -59,4 +60,10 @@ public class LeaveRequestMessage extends HighPriorityDistributionMessage {
reason = DataSerializer.readString(in);
}
+ @Override
+ public String toString() {
+ return getShortClassName() + "(" + memberID
+ + "; reason=" + reason + ")";
+ }
+
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/b8f40f4c/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/JGroupsMessenger.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/JGroupsMessenger.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/JGroupsMessenger.java
index 656230a..a653110 100755
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/JGroupsMessenger.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/JGroupsMessenger.java
@@ -1,5 +1,7 @@
package com.gemstone.gemfire.distributed.internal.membership.gms.messenger;
+import static com.gemstone.gemfire.distributed.internal.membership.gms.GMSUtil.replaceStrings;
+
import it.unimi.dsi.fastutil.ints.Int2ObjectOpenHashMap;
import java.io.BufferedReader;
@@ -571,7 +573,7 @@ public class JGroupsMessenger implements Messenger {
* @param version the version of the recipient
* @return the new message
*/
- private Message createJGMessage(DistributionMessage gfmsg, JGAddress src, short version) {
+ Message createJGMessage(DistributionMessage gfmsg, JGAddress src, short version) {
if(gfmsg instanceof DirectReplyMessage) {
((DirectReplyMessage) gfmsg).registerProcessor();
}
@@ -723,6 +725,13 @@ public class JGroupsMessenger implements Messenger {
public String getJGroupsStackConfig() {
return this.jgStackConfig;
}
+
+ /**
+ * for unit testing we need to replace UDP with a fake UDP protocol
+ */
+ public void setJGroupsStackConfigForTesting(String config) {
+ this.jgStackConfig = config;
+ }
/**
* returns the member ID for the given GMSMember object
@@ -750,34 +759,15 @@ public class JGroupsMessenger implements Messenger {
}
/**
- * replaces all occurrences of a given string in the properties argument with the
- * given value
- */
- private String replaceStrings(String properties, String property, String value) {
- StringBuffer sb = new StringBuffer();
- int start = 0;
- int index = properties.indexOf(property);
- while (index != -1) {
- sb.append(properties.substring(start, index));
- sb.append(value);
-
- start = index + property.length();
- index = properties.indexOf(property, start);
- }
- sb.append(properties.substring(start));
- return sb.toString();
- }
-
-
- /**
* Puller receives incoming JGroups messages and passes them to a handler
*/
class JGroupsReceiver implements Receiver {
@Override
public void receive(Message jgmsg) {
- if (services.getManager().shutdownInProgress())
+ if (services.getManager().shutdownInProgress()) {
return;
+ }
if (logger.isDebugEnabled()) {
logger.debug("JGroupsMessenger received {} headers: {}", jgmsg, jgmsg.getHeaders());
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/b8f40f4c/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/mgr/GMSMembershipManager.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/mgr/GMSMembershipManager.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/mgr/GMSMembershipManager.java
index 76d83f7..ec2bd53 100755
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/mgr/GMSMembershipManager.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/mgr/GMSMembershipManager.java
@@ -18,6 +18,7 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
+import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
@@ -40,7 +41,6 @@ import com.gemstone.gemfire.SystemFailure;
import com.gemstone.gemfire.ToDataException;
import com.gemstone.gemfire.cache.Cache;
import com.gemstone.gemfire.cache.server.CacheServer;
-import com.gemstone.gemfire.cache.util.BoundedLinkedHashMap;
import com.gemstone.gemfire.distributed.DistributedMember;
import com.gemstone.gemfire.distributed.DistributedSystem;
import com.gemstone.gemfire.distributed.DistributedSystemDisconnectedException;
@@ -59,7 +59,6 @@ import com.gemstone.gemfire.distributed.internal.ThrottlingMemLinkedQueueWithDMS
import com.gemstone.gemfire.distributed.internal.direct.DirectChannel;
import com.gemstone.gemfire.distributed.internal.membership.DistributedMembershipListener;
import com.gemstone.gemfire.distributed.internal.membership.InternalDistributedMember;
-import com.gemstone.gemfire.distributed.internal.membership.MemberAttributes;
import com.gemstone.gemfire.distributed.internal.membership.MembershipManager;
import com.gemstone.gemfire.distributed.internal.membership.MembershipTestHook;
import com.gemstone.gemfire.distributed.internal.membership.NetView;
@@ -125,16 +124,6 @@ public class GMSMembershipManager implements MembershipManager, Manager
}
}
- /**
- * Trick class to make the view lock more visible
- * in stack traces
- *
- */
- static class ViewLock extends ReentrantReadWriteLock {
- public ViewLock() {
- }
- }
-
static class StartupEvent {
static final int DEPARTURE = 1;
static final int CONNECT = 2;
@@ -289,7 +278,7 @@ public class GMSMembershipManager implements MembershipManager, Manager
*
* @see #latestView
*/
- protected ViewLock latestViewLock = new ViewLock();
+ protected ReadWriteLock latestViewLock = new ReentrantReadWriteLock();
/**
* This is the listener that accepts our membership events
@@ -565,7 +554,8 @@ public class GMSMembershipManager implements MembershipManager, Manager
// }
// We perform the update under a global lock so that other
// incoming events will not be lost in terms of our global view.
- synchronized (latestViewLock) {
+ latestViewLock.writeLock().lock();
+ try {
// first determine the version for multicast message serialization
Version version = Version.CURRENT;
for (Iterator<Map.Entry<InternalDistributedMember, Long>> it=surpriseMembers.entrySet().iterator(); it.hasNext(); ) {
@@ -753,7 +743,9 @@ public class GMSMembershipManager implements MembershipManager, Manager
}
catch (DistributedSystemDisconnectedException se) {
}
- } // synchronized
+ } finally {
+ latestViewLock.writeLock().unlock();
+ }
logger.info(LogMarker.DM_VIEWS, LocalizedMessage.create(
LocalizedStrings.GroupMembershipService_MEMBERSHIP_FINISHED_VIEW_PROCESSING_VIEWID___0, Long.valueOf(newViewId)));
}
@@ -1013,7 +1005,7 @@ public class GMSMembershipManager implements MembershipManager, Manager
/**
* Automatic removal of a member (for internal
- * use only). Synchronizes on {@link #latestViewLock} and then deletes
+ * use only). Write-locks {@link #latestViewLock} and then deletes
* the member.
*
* @param dm
@@ -2265,8 +2257,8 @@ public class GMSMembershipManager implements MembershipManager, Manager
}
/**
- * Add a mapping from the given member to the given stub. Must be
- * synchronized on {@link #latestViewLock} by caller.
+ * Add a mapping from the given member to the given stub. Must
+ * be called with {@link #latestViewLock} held.
*
* @param member
* @param theChannel
@@ -2946,4 +2938,67 @@ public class GMSMembershipManager implements MembershipManager, Manager
}
+ /**
+ * Class <code>BoundedLinkedHashMap</code> is a bounded
+ * <code>LinkedHashMap</code>. The bound is the maximum
+ * number of entries the <code>BoundedLinkedHashMap</code>
+ * can contain.
+ */
+ static class BoundedLinkedHashMap extends LinkedHashMap
+ {
+ private static final long serialVersionUID = -3419897166186852692L;
+
+ /**
+ * The maximum number of entries allowed in this
+ * <code>BoundedLinkedHashMap</code>
+ */
+ protected int _maximumNumberOfEntries;
+
+ /**
+ * Constructor.
+ *
+ * @param initialCapacity The initial capacity.
+ * @param loadFactor The load factor
+ * @param maximumNumberOfEntries The maximum number of allowed entries
+ */
+ public BoundedLinkedHashMap(int initialCapacity, float loadFactor, int maximumNumberOfEntries) {
+ super(initialCapacity, loadFactor);
+ this._maximumNumberOfEntries = maximumNumberOfEntries;
+ }
+
+ /**
+ * Constructor.
+ *
+ * @param initialCapacity The initial capacity.
+ * @param maximumNumberOfEntries The maximum number of allowed entries
+ */
+ public BoundedLinkedHashMap(int initialCapacity, int maximumNumberOfEntries) {
+ super(initialCapacity);
+ this._maximumNumberOfEntries = maximumNumberOfEntries;
+ }
+
+ /**
+ * Constructor.
+ *
+ * @param maximumNumberOfEntries The maximum number of allowed entries
+ */
+ public BoundedLinkedHashMap(int maximumNumberOfEntries) {
+ super();
+ this._maximumNumberOfEntries = maximumNumberOfEntries;
+ }
+
+ /**
+ * Returns the maximum number of entries.
+ * @return the maximum number of entries
+ */
+ public int getMaximumNumberOfEntries(){
+ return this._maximumNumberOfEntries;
+ }
+
+ @Override
+ protected boolean removeEldestEntry(Map.Entry entry) {
+ return size() > this._maximumNumberOfEntries;
+ }
+ }
+
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/b8f40f4c/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/internal/membership/MembershipJUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/internal/membership/MembershipJUnitTest.java b/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/internal/membership/MembershipJUnitTest.java
index 7da7780..b668f7a 100755
--- a/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/internal/membership/MembershipJUnitTest.java
+++ b/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/internal/membership/MembershipJUnitTest.java
@@ -7,42 +7,42 @@
*/
package com.gemstone.gemfire.distributed.internal.membership;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
import java.io.File;
import java.net.InetAddress;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Properties;
-import java.util.Set;
import junit.framework.TestCase;
import org.apache.logging.log4j.Level;
import org.apache.logging.log4j.core.config.ConfigurationFactory;
-import org.junit.After;
import org.junit.AfterClass;
-import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.experimental.categories.Category;
+import com.gemstone.gemfire.GemFireConfigException;
import com.gemstone.gemfire.distributed.Locator;
import com.gemstone.gemfire.distributed.internal.DMStats;
import com.gemstone.gemfire.distributed.internal.DistributionConfig;
import com.gemstone.gemfire.distributed.internal.DistributionConfigImpl;
import com.gemstone.gemfire.distributed.internal.DistributionManager;
-import com.gemstone.gemfire.distributed.internal.DistributionMessage;
-import com.gemstone.gemfire.distributed.internal.LonerDistributionManager.DummyDMStats;
-import com.gemstone.gemfire.distributed.internal.MembershipListener;
-import com.gemstone.gemfire.distributed.internal.PoolStatHelper;
+import com.gemstone.gemfire.distributed.internal.InternalLocator;
+import com.gemstone.gemfire.distributed.internal.membership.gms.ServiceConfig;
+import com.gemstone.gemfire.distributed.internal.membership.gms.Services;
+import com.gemstone.gemfire.distributed.internal.membership.gms.membership.GMSJoinLeave;
import com.gemstone.gemfire.internal.AvailablePortHelper;
import com.gemstone.gemfire.internal.SocketCreator;
import com.gemstone.gemfire.internal.admin.remote.RemoteTransportConfig;
import com.gemstone.gemfire.internal.logging.LogService;
-import com.gemstone.gemfire.internal.tcp.Stub;
-import com.gemstone.gemfire.test.junit.categories.UnitTest;
+import com.gemstone.gemfire.test.junit.categories.IntegrationTest;
-@Category(UnitTest.class)
+@Category(IntegrationTest.class)
public class MembershipJUnitTest extends TestCase {
static Level baseLogLevel;
@@ -52,13 +52,13 @@ public class MembershipJUnitTest extends TestCase {
@BeforeClass
public static void setupClass() {
- baseLogLevel = LogService.getBaseLogLevel();
- LogService.setBaseLogLevel(Level.DEBUG);
+// baseLogLevel = LogService.getBaseLogLevel();
+// LogService.setBaseLogLevel(Level.DEBUG);
}
@AfterClass
protected void tearDown() throws Exception {
- LogService.setBaseLogLevel(baseLogLevel);
+// LogService.setBaseLogLevel(baseLogLevel);
}
/**
@@ -92,11 +92,12 @@ public class MembershipJUnitTest extends TestCase {
members[i].setVmKind(DistributionManager.NORMAL_DM_TYPE);
members[i++].getNetMember().setPreferredForCoordinator(false);
- List<InternalDistributedMember> vmbrs = new ArrayList(members.length);
+ List<InternalDistributedMember> vmbrs = new ArrayList<>(members.length);
for (i=0; i<members.length; i++) {
vmbrs.add(members[i]);
}
- NetView lastView = new NetView(members[0], 4, vmbrs, Collections.EMPTY_LIST, Collections.EMPTY_LIST);
+ List<InternalDistributedMember> empty = Collections.emptyList();
+ NetView lastView = new NetView(members[0], 4, vmbrs, empty, empty);
InternalDistributedMember leader = members[2];
assertTrue(!leader.getNetMember().preferredForCoordinator());
@@ -113,7 +114,7 @@ public class MembershipJUnitTest extends TestCase {
failedMembers.add(members[members.length-2]); // admin
List<InternalDistributedMember> newMbrs = new ArrayList<InternalDistributedMember>(lastView.getMembers());
newMbrs.removeAll(failedMembers);
- NetView newView = new NetView(members[0], 5, newMbrs, Collections.EMPTY_LIST, failedMembers);
+ NetView newView = new NetView(members[0], 5, newMbrs, empty, failedMembers);
int failedWeight = newView.getCrashedMemberWeight(lastView);
// System.out.println("last view = " + lastView);
@@ -134,7 +135,7 @@ public class MembershipJUnitTest extends TestCase {
* and makes more assertions.
*/
@Test
- public void testJoinLeave() throws Exception {
+ public void testMultipleManagersInSameProcess() throws Exception {
MembershipManager m1=null, m2=null;
Locator l = null;
@@ -147,7 +148,10 @@ public class MembershipJUnitTest extends TestCase {
// this locator will hook itself up with the first MembershipManager
// to be created
- l = Locator.startLocator(port, new File(""), localHost);
+// l = Locator.startLocator(port, new File(""), localHost);
+ l = InternalLocator.startLocator(port, new File(""), null,
+ null, null, localHost, false, new Properties(), true, false, null,
+ false);
// create configuration objects
Properties nonDefault = new Properties();
@@ -161,13 +165,13 @@ public class MembershipJUnitTest extends TestCase {
DistributionManager.NORMAL_DM_TYPE);
// start the first membership manager
- MembershipListener listener1 = new MembershipListener();
- DMStats stats1 = new MyStats();
+ DistributedMembershipListener listener1 = mock(DistributedMembershipListener.class);
+ DMStats stats1 = mock(DMStats.class);
m1 = MemberFactory.newMembershipManager(listener1, config, transport, stats1);
// start the second membership manager
- MembershipListener listener2 = new MembershipListener();
- DMStats stats2 = new MyStats();
+ DistributedMembershipListener listener2 = mock(DistributedMembershipListener.class);
+ DMStats stats2 = mock(DMStats.class);
m2 = MemberFactory.newMembershipManager(listener2, config, transport, stats2);
assert m2.getView().size() == 2;
@@ -180,8 +184,6 @@ public class MembershipJUnitTest extends TestCase {
assert m1.getView().size() == 1;
}
finally {
- System.getProperties().remove(ConfigurationFactory.CONFIGURATION_FILE_PROPERTY);
- LogService.reconfigure();
if (m2 != null) {
m2.shutdown();
@@ -193,59 +195,35 @@ public class MembershipJUnitTest extends TestCase {
l.stop();
}
}
-
}
-
- static class MembershipListener implements DistributedMembershipListener {
-
- @Override
- public void viewInstalled(NetView view) {
- }
-
- @Override
- public void quorumLost(Set<InternalDistributedMember> failures,
- List<InternalDistributedMember> remainingMembers) {
- }
-
- @Override
- public void newMemberConnected(InternalDistributedMember m, Stub stub) {
- }
-
- @Override
- public void memberDeparted(InternalDistributedMember id, boolean crashed,
- String reason) {
- }
-
- @Override
- public void memberSuspect(InternalDistributedMember suspect,
- InternalDistributedMember whoSuspected) {
- }
-
- @Override
- public void messageReceived(DistributionMessage o) {
- }
-
- @Override
- public boolean isShutdownMsgSent() {
- return false;
- }
- @Override
- public void membershipFailure(String reason, Throwable t) {
- }
-
- @Override
- public DistributionManager getDM() {
- return null;
- }
+ @Test
+ public void testMulticastDiscoveryNotAllowed() {
+ Properties nonDefault = new Properties();
+ nonDefault.put(DistributionConfig.DISABLE_TCP_NAME, "true");
+ nonDefault.put(DistributionConfig.MCAST_PORT_NAME, "12345");
+ nonDefault.put(DistributionConfig.LOG_FILE_NAME, "");
+ nonDefault.put(DistributionConfig.LOG_LEVEL_NAME, "fine");
+ nonDefault.put(DistributionConfig.LOCATORS_NAME, "");
+ DistributionConfigImpl config = new DistributionConfigImpl(nonDefault);
+ RemoteTransportConfig transport = new RemoteTransportConfig(config,
+ DistributionManager.NORMAL_DM_TYPE);
+
+ ServiceConfig serviceConfig = mock(ServiceConfig.class);
+ when(serviceConfig.getDistributionConfig()).thenReturn(config);
+ when(serviceConfig.getTransport()).thenReturn(transport);
+ Services services = mock(Services.class);
+ when(services.getConfig()).thenReturn(serviceConfig);
+
+ GMSJoinLeave joinLeave = new GMSJoinLeave();
+ try {
+ joinLeave.init(services);
+ fail("expected a GemFireConfigException to be thrown because no locators are configured");
+ } catch (GemFireConfigException e) {
+ // expected
+ }
}
- static class MyStats extends DummyDMStats {
- }
-
- public static class DummyPoolStatHelper implements PoolStatHelper {
- public void startJob() {}
- public void endJob(){}
- }
+
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/b8f40f4c/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/internal/membership/gms/MembershipManagerHelper.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/internal/membership/gms/MembershipManagerHelper.java b/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/internal/membership/gms/MembershipManagerHelper.java
index a4eaa40..19396f6 100644
--- a/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/internal/membership/gms/MembershipManagerHelper.java
+++ b/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/internal/membership/gms/MembershipManagerHelper.java
@@ -147,7 +147,7 @@ public class MembershipManagerHelper
public static void crashDistributedSystem(final DistributedSystem msys) {
MembershipManagerHelper.inhibitForcedDisconnectLogging(true);
MembershipManagerHelper.playDead(msys);
- getMembershipManager(msys).uncleanShutdown("test is forcing disconnect", new ForcedDisconnectException("test is forcing disconnect"));
+ ((GMSMembershipManager)getMembershipManager(msys)).forceDisconnect("for testing");
MembershipManagerHelper.inhibitForcedDisconnectLogging(false);
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/b8f40f4c/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/internal/membership/gms/membership/GMSJoinLeaveJUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/internal/membership/gms/membership/GMSJoinLeaveJUnitTest.java b/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/internal/membership/gms/membership/GMSJoinLeaveJUnitTest.java
index a769aa9..cbabc20 100644
--- a/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/internal/membership/gms/membership/GMSJoinLeaveJUnitTest.java
+++ b/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/internal/membership/gms/membership/GMSJoinLeaveJUnitTest.java
@@ -2,6 +2,8 @@ package com.gemstone.gemfire.distributed.internal.membership.gms.membership;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
@@ -21,14 +23,17 @@ import org.mockito.stubbing.Answer;
import com.gemstone.gemfire.distributed.internal.DistributionConfig;
import com.gemstone.gemfire.distributed.internal.membership.InternalDistributedMember;
import com.gemstone.gemfire.distributed.internal.membership.NetView;
+import com.gemstone.gemfire.distributed.internal.membership.gms.GMSMember;
import com.gemstone.gemfire.distributed.internal.membership.gms.ServiceConfig;
import com.gemstone.gemfire.distributed.internal.membership.gms.Services;
+import com.gemstone.gemfire.distributed.internal.membership.gms.Services.Stopper;
import com.gemstone.gemfire.distributed.internal.membership.gms.interfaces.Authenticator;
import com.gemstone.gemfire.distributed.internal.membership.gms.interfaces.Manager;
import com.gemstone.gemfire.distributed.internal.membership.gms.interfaces.Messenger;
import com.gemstone.gemfire.distributed.internal.membership.gms.messages.InstallViewMessage;
import com.gemstone.gemfire.distributed.internal.membership.gms.messages.JoinRequestMessage;
import com.gemstone.gemfire.distributed.internal.membership.gms.messages.JoinResponseMessage;
+import com.gemstone.gemfire.distributed.internal.membership.gms.messages.LeaveRequestMessage;
import com.gemstone.gemfire.distributed.internal.membership.gms.messages.RemoveMemberMessage;
import com.gemstone.gemfire.distributed.internal.membership.gms.messages.ViewAckMessage;
import com.gemstone.gemfire.internal.Version;
@@ -47,6 +52,8 @@ public class GMSJoinLeaveJUnitTest {
private Object credentials = new Object();
private Messenger messenger;
private GMSJoinLeave gmsJoinLeave;
+ private Manager manager;
+ private Stopper stopper;
public void initMocks() throws IOException {
initMocks(false);
@@ -66,11 +73,19 @@ public class GMSJoinLeaveJUnitTest {
messenger = mock(Messenger.class);
when(messenger.getMemberID()).thenReturn(gmsJoinLeaveMemberId);
+
+ stopper = mock(Stopper.class);
+ when(stopper.isCancelInProgress()).thenReturn(false);
+
+ manager = mock(Manager.class);
services = mock(Services.class);
+ when(services.getAuthenticator()).thenReturn(authenticator);
when(services.getConfig()).thenReturn(mockConfig);
when(services.getMessenger()).thenReturn(messenger);
-
+ when(services.getCancelCriterion()).thenReturn(stopper);
+ when(services.getManager()).thenReturn(manager);
+
mockMembers = new InternalDistributedMember[4];
for (int i = 0; i < mockMembers.length; i++) {
mockMembers[i] = new InternalDistributedMember("localhost", 8888 + i);
@@ -165,7 +180,7 @@ public class GMSJoinLeaveJUnitTest {
}
- @Test
+ @Test
public void testRejectOlderView() throws IOException {
initMocks();
prepareAndInstallView();
@@ -207,24 +222,266 @@ public class GMSJoinLeaveJUnitTest {
verify(mockManager).forceDisconnect(any(String.class));
}
- public void testOlderPreparedViewBeforeFirstViewInstalled() throws IOException {
+ private class MethodExecuted implements Answer {
+ private boolean methodExecuted = false;
+ @Override
+ public Object answer(InvocationOnMock invocation) {
+ //do we only expect a join response on a failure?
+ methodExecuted = true;
+ return null;
+ }
+
+ public boolean isMethodExecuted() {
+ return methodExecuted;
+ }
+ }
+
+ @Test
+ public void testNonMemberCantRemoveMember() throws Exception {
+ String reason = "testing";
initMocks();
- prepareView();
+ prepareAndInstallView();
+ // test that a non-member can't remove another member
+ RemoveMemberMessage msg = new RemoveMemberMessage(mockMembers[0], mockMembers[1], reason);
+ msg.setSender(new InternalDistributedMember("localhost", 9000));
+ gmsJoinLeave.processMessage(msg);
+ Assert.assertTrue("RemoveMemberMessage should not have been added to view requests", gmsJoinLeave.getViewRequests().size() == 0);
+ }
+
+ @Test
+ public void testDuplicateLeaveRequestDoesNotCauseNewView() throws Exception {
+ String reason = "testing";
+ initMocks();
+ prepareAndInstallView();
+ gmsJoinLeave.getView().add(gmsJoinLeaveMemberId);
+ gmsJoinLeave.becomeCoordinator();
+
+ LeaveRequestMessage msg = new LeaveRequestMessage(gmsJoinLeave.getMemberID(), mockMembers[0], reason);
+ msg.setSender(mockMembers[0]);
+ gmsJoinLeave.processMessage(msg);
+ msg = new LeaveRequestMessage(gmsJoinLeave.getMemberID(), mockMembers[0], reason);
+ msg.setSender(mockMembers[0]);
+ gmsJoinLeave.processMessage(msg);
+
+ waitForViewAndNoRequestsInProgress(7);
+
+ NetView view = gmsJoinLeave.getView();
+ Assert.assertTrue("expected member to be removed: " + mockMembers[0] + "; view: " + view,
+ !view.contains(mockMembers[0]));
+ Assert.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();
+ gmsJoinLeave.getView().add(gmsJoinLeaveMemberId);
+ gmsJoinLeave.getView().add(mockMembers[1]);
+ gmsJoinLeave.becomeCoordinator();
+ RemoveMemberMessage msg = new RemoveMemberMessage(gmsJoinLeave.getMemberID(), mockMembers[0], reason);
+ msg.setSender(mockMembers[0]);
+ gmsJoinLeave.processMessage(msg);
+ msg = new RemoveMemberMessage(gmsJoinLeave.getMemberID(), mockMembers[0], reason);
+ msg.setSender(mockMembers[0]);
+ gmsJoinLeave.processMessage(msg);
+
+ waitForViewAndNoRequestsInProgress(7);
+
+ NetView view = gmsJoinLeave.getView();
+ Assert.assertTrue("expected member to be removed: " + mockMembers[0] + "; view: " + view,
+ !view.contains(mockMembers[0]));
+ Assert.assertTrue("expected member to be in crashedMembers collection: " + mockMembers[0] + "; view: " + view,
+ view.getCrashedMembers().contains(mockMembers[0]));
+ }
+
+
+ @Test
+ public void testDuplicateJoinRequestDoesNotCauseNewView() throws Exception {
+ initMocks();
+ prepareAndInstallView();
+ gmsJoinLeave.getView().add(gmsJoinLeaveMemberId);
+ gmsJoinLeave.getView().add(mockMembers[1]);
+ gmsJoinLeave.becomeCoordinator();
+ JoinRequestMessage msg = new JoinRequestMessage(gmsJoinLeaveMemberId, mockMembers[2], null);
+ msg.setSender(mockMembers[2]);
+ gmsJoinLeave.processMessage(msg);
+ msg = new JoinRequestMessage(gmsJoinLeaveMemberId, mockMembers[2], null);
+ msg.setSender(mockMembers[2]);
+ gmsJoinLeave.processMessage(msg);
+
+ waitForViewAndNoRequestsInProgress(7);
+
+ NetView view = gmsJoinLeave.getView();
+ Assert.assertTrue("expected member to be added: " + mockMembers[2] + "; view: " + view,
+ view.contains(mockMembers[2]));
+ List<InternalDistributedMember> members = view.getMembers();
+ int occurrences = 0;
+ for (InternalDistributedMember mbr: members) {
+ if (mbr.equals(mockMembers[2])) {
+ occurrences += 1;
+ }
+ }
+ Assert.assertTrue("expected member to only be in the view once: " + mockMembers[2] + "; view: " + view,
+ occurrences == 1);
+ }
+
+
+ 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) {
+ System.out.println("view requests: " + gmsJoinLeave.getViewRequests());
+ System.out.println("current view: " + gmsJoinLeave.getView());
+ throw new RuntimeException("timeout waiting for view #" + viewId);
+ }
+ Thread.sleep(1000);
+ }
+ }
+
+ @Test
+ public void testRemoveCausesForcedDisconnect() throws Exception {
+ String reason = "testing";
+ initMocks();
+ prepareAndInstallView();
+ gmsJoinLeave.getView().add(mockMembers[1]);
+ RemoveMemberMessage msg = new RemoveMemberMessage(mockMembers[0], gmsJoinLeave.getMemberID(), reason);
+ msg.setSender(mockMembers[1]);
+ gmsJoinLeave.processMessage(msg);
+ verify(manager).forceDisconnect(reason);
+ }
+
+ @Test
+ public void testLeaveCausesForcedDisconnect() throws Exception {
+ String reason = "testing";
+ initMocks();
+ prepareAndInstallView();
+ gmsJoinLeave.getView().add(gmsJoinLeaveMemberId);
+ gmsJoinLeave.getView().add(mockMembers[1]);
+ LeaveRequestMessage msg = new LeaveRequestMessage(gmsJoinLeave.getMemberID(), gmsJoinLeave.getMemberID(), reason);
+ msg.setSender(mockMembers[1]);
+ gmsJoinLeave.processMessage(msg);
+ verify(manager).forceDisconnect(reason);
+ }
+
+ @Test
+ public void testLeaveOfNonMemberIsNoOp() throws Exception {
+ String reason = "testing";
+ initMocks();
+ prepareAndInstallView();
+ mockMembers[1].setVmViewId(gmsJoinLeave.getView().getViewId()-1);
+ LeaveRequestMessage msg = new LeaveRequestMessage(gmsJoinLeave.getMemberID(), mockMembers[1], reason);
+ msg.setSender(mockMembers[1]);
+ gmsJoinLeave.processMessage(msg);
+ Assert.assertTrue("Expected leave request from non-member to be ignored", gmsJoinLeave.getViewRequests().isEmpty());
+ }
+
+ @Test
+ public void testBecomeCoordinator() throws Exception {
+ String reason = "testing";
+ initMocks();
+ prepareAndInstallView();
+ NetView view = gmsJoinLeave.getView();
+ view.add(gmsJoinLeaveMemberId);
+ InternalDistributedMember creator = view.getCreator();
+ LeaveRequestMessage msg = new LeaveRequestMessage(creator, creator, reason);
+ msg.setSender(creator);
+ gmsJoinLeave.processMessage(msg);
+ Assert.assertTrue("Expected becomeCoordinator to be invoked", gmsJoinLeave.isCoordinator());
+ }
+
+ @Test
+ public void testNetworkPartitionDetected() throws IOException {
+ initMocks(true);
+ prepareAndInstallView();
+
+ // set up a view with sufficient members, then create a new view
+ // where enough weight is lost to cause a network partition
- int viewId = 0;
List<InternalDistributedMember> mbrs = new LinkedList<InternalDistributedMember>();
List<InternalDistributedMember> shutdowns = new LinkedList<InternalDistributedMember>();
List<InternalDistributedMember> crashes = new LinkedList<InternalDistributedMember>();
+ mbrs.add(mockMembers[0]);
mbrs.add(mockMembers[1]);
mbrs.add(mockMembers[2]);
- mbrs.add(mockMembers[3]);
-
- //install the view
- NetView netView = new NetView(mockMembers[0], viewId, mbrs, shutdowns, crashes);
- InstallViewMessage installViewMessage = new InstallViewMessage(netView, credentials, true);
+ mbrs.add(gmsJoinLeaveMemberId);
+
+ ((GMSMember)mockMembers[1].getNetMember()).setMemberWeight((byte)20);
+
+ NetView newView = new NetView(mockMembers[0], gmsJoinLeave.getView().getViewId()+1, mbrs, shutdowns, crashes);
+ InstallViewMessage installViewMessage = new InstallViewMessage(newView, credentials, false);
gmsJoinLeave.processMessage(installViewMessage);
- Assert.assertNotEquals(netView, gmsJoinLeave.getView());
+ crashes = new LinkedList<>(crashes);
+ crashes.add(mockMembers[1]);
+ crashes.add(mockMembers[2]);
+ mbrs = new LinkedList<>(mbrs);
+ mbrs.remove(mockMembers[1]);
+ mbrs.remove(mockMembers[2]);
+ NetView partitionView = new NetView(mockMembers[0], newView.getViewId()+1, mbrs, shutdowns, crashes);
+ installViewMessage = new InstallViewMessage(partitionView, credentials, false);
+ gmsJoinLeave.processMessage(installViewMessage);
+
+ verify(manager).forceDisconnect(any(String.class));
+ verify(manager).quorumLost(crashes, newView);
+ }
+
+ @Test
+ public void testQuorumLossNotificationWithNetworkPartitionDetectionDisabled() throws IOException {
+ initMocks(false);
+ prepareAndInstallView();
+
+ // set up a view with sufficient members, then create a new view
+ // where enough weight is lost to cause a network partition
+
+ List<InternalDistributedMember> mbrs = new LinkedList<InternalDistributedMember>();
+ List<InternalDistributedMember> shutdowns = new LinkedList<InternalDistributedMember>();
+ List<InternalDistributedMember> crashes = new LinkedList<InternalDistributedMember>();
+ mbrs.add(mockMembers[0]);
+ mbrs.add(mockMembers[1]);
+ mbrs.add(mockMembers[2]);
+ mbrs.add(gmsJoinLeaveMemberId);
+
+ ((GMSMember)mockMembers[1].getNetMember()).setMemberWeight((byte)20);
+
+ NetView newView = new NetView(mockMembers[0], gmsJoinLeave.getView().getViewId()+1, mbrs, shutdowns, crashes);
+ InstallViewMessage installViewMessage = new InstallViewMessage(newView, credentials, false);
+ gmsJoinLeave.processMessage(installViewMessage);
+
+ crashes = new LinkedList<>(crashes);
+ crashes.add(mockMembers[1]);
+ crashes.add(mockMembers[2]);
+ mbrs = new LinkedList<>(mbrs);
+ mbrs.remove(mockMembers[1]);
+ mbrs.remove(mockMembers[2]);
+ NetView partitionView = new NetView(mockMembers[0], newView.getViewId()+1, mbrs, shutdowns, crashes);
+ installViewMessage = new InstallViewMessage(partitionView, credentials, false);
+ gmsJoinLeave.processMessage(installViewMessage);
+
+ verify(manager, never()).forceDisconnect(any(String.class));
+ verify(manager).quorumLost(crashes, newView);
}
+
+ @Test
+ public void testConflictingPrepare() throws Exception {
+ initMocks(true);
+ prepareAndInstallView();
+
+ NetView gmsView = gmsJoinLeave.getView();
+ NetView newView = new NetView(gmsView, gmsView.getViewId()+6);
+ InstallViewMessage msg = new InstallViewMessage(newView, null, true);
+ gmsJoinLeave.processMessage(msg);
+
+ NetView alternateView = new NetView(gmsView, gmsView.getViewId()+1);
+ msg = new InstallViewMessage(alternateView, null, true);
+ gmsJoinLeave.processMessage(msg);
+
+ Assert.assertTrue(gmsJoinLeave.getPreparedView().equals(newView));
+ }
+
+
}