You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by je...@apache.org on 2015/12/07 18:06:14 UTC
[32/50] [abbrv] incubator-geode git commit: Adding code-coverage
testing for GMSMembershipManager and removing unused code.
Adding code-coverage testing for GMSMembershipManager and removing unused code.
Project: http://git-wip-us.apache.org/repos/asf/incubator-geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-geode/commit/38dd3ed8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-geode/tree/38dd3ed8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-geode/diff/38dd3ed8
Branch: refs/heads/feature/GEODE-17
Commit: 38dd3ed892be2acd1609c0b72ee5f4d4cfe85175
Parents: 253549d
Author: Bruce Schuchardt <bs...@pivotal.io>
Authored: Fri Nov 20 13:33:21 2015 -0800
Committer: Bruce Schuchardt <bs...@pivotal.io>
Committed: Tue Dec 1 08:47:51 2015 -0800
----------------------------------------------------------------------
.../internal/DistributionChannel.java | 2 +-
.../internal/DistributionManager.java | 29 +-
.../internal/HighPriorityAckedMessage.java | 4 +-
.../distributed/internal/StartupMessage.java | 19 -
.../distributed/internal/StartupOperation.java | 1 -
.../internal/direct/DirectChannel.java | 16 +-
.../internal/direct/DirectChannelListener.java | 22 ++
.../DistributedMembershipListener.java | 13 +-
.../internal/membership/MemberFactory.java | 8 -
.../internal/membership/MemberServices.java | 6 -
.../internal/membership/MembershipManager.java | 13 -
.../internal/membership/gms/GMSMember.java | 22 --
.../membership/gms/GMSMemberFactory.java | 12 +-
.../internal/membership/gms/GMSUtil.java | 19 +
.../membership/gms/fd/GMSHealthMonitor.java | 2 +-
.../gms/messenger/JGroupsMessenger.java | 9 +-
.../gms/mgr/GMSMembershipManager.java | 342 +++--------------
.../internal/logging/log4j/LogMarker.java | 1 -
.../gemfire/internal/tcp/Connection.java | 7 +
.../gemfire/internal/tcp/TCPConduit.java | 6 +-
.../membership/MembershipJUnitTest.java | 29 +-
.../locator/GMSLocatorRecoveryJUnitTest.java | 4 +-
.../gms/mgr/GMSMembershipManagerJUnitTest.java | 370 +++++++++++++++++++
23 files changed, 520 insertions(+), 436 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/38dd3ed8/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/DistributionChannel.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/DistributionChannel.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/DistributionChannel.java
index daccc9c..a31c92f 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/DistributionChannel.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/DistributionChannel.java
@@ -132,7 +132,7 @@ public class DistributionChannel {
public long getId() {
MembershipManager mgr = this.membershipManager;
if (mgr == null) {
- throw new DistributedSystemDisconnectedException(LocalizedStrings.DistributionChannel_I_NO_LONGER_HAVE_A_MEMBERSHIP_ID.toLocalizedString(), membershipManager.getShutdownCause());
+ throw new DistributedSystemDisconnectedException(LocalizedStrings.DistributionChannel_I_NO_LONGER_HAVE_A_MEMBERSHIP_ID.toLocalizedString());
}
InternalDistributedMember moi = mgr.getLocalMember();
if (moi == null) {
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/38dd3ed8/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/DistributionManager.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/DistributionManager.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/DistributionManager.java
index 837194c..5d3bdce 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/DistributionManager.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/DistributionManager.java
@@ -593,7 +593,7 @@ public class DistributionManager
}
}
}
- dm.addNewMember(id, null); // add ourselves
+ dm.addNewMember(id); // add ourselves
dm.selectElder(); // ShutdownException could be thrown here
}
@@ -1440,7 +1440,7 @@ public class DistributionManager
// Add them all to our view
Iterator<InternalDistributedMember> it = v.getMembers().iterator();
while (it.hasNext()) {
- addNewMember(it.next(), null);
+ addNewMember(it.next());
}
// Figure out who the elder is...
@@ -1610,15 +1610,6 @@ public class DistributionManager
}
/**
- * Returns a remote reference to the channel used for point-to-point
- * communications, or null if the normal channel is being used for
- * this.
- */
- protected Stub getDirectChannel() {
- return membershipManager.getDirectChannel();
- }
-
- /**
* Returns an unmodifiable set containing the identities of all of
* the known (non-admin-only) distribution managers.
*/
@@ -1820,16 +1811,16 @@ public class DistributionManager
}
}
- public void addNewMember(InternalDistributedMember member, Stub stub) {
+ public void addNewMember(InternalDistributedMember member) {
// This is the place to cleanup the zombieMembers
int vmType = member.getVmKind();
switch (vmType) {
case ADMIN_ONLY_DM_TYPE:
- handleConsoleStartup(member, stub);
+ handleConsoleStartup(member);
break;
case LOCATOR_DM_TYPE:
case NORMAL_DM_TYPE:
- handleManagerStartup(member, stub);
+ handleManagerStartup(member);
break;
default:
throw new InternalGemFireError(LocalizedStrings.DistributionManager_UNKNOWN_MEMBER_TYPE_0.toLocalizedString(Integer.valueOf(vmType)));
@@ -2795,7 +2786,7 @@ public class DistributionManager
if (unresponsiveElder) {
logger.warn(LocalizedMessage.create(
LocalizedStrings.DistributionManager_FORCING_AN_ELDER_JOIN_EVENT_SINCE_A_STARTUP_RESPONSE_WAS_NOT_RECEIVED_FROM_ELDER__0_, e));
- handleManagerStartup(e, null/*stub already registered*/);
+ handleManagerStartup(e);
}
} // an elder exists
} // someone didn't reply
@@ -3104,7 +3095,7 @@ public class DistributionManager
* The id of the distribution manager starting up
*
*/
- private void handleManagerStartup(InternalDistributedMember theId, Stub directChannel) {
+ private void handleManagerStartup(InternalDistributedMember theId) {
HashMap<InternalDistributedMember,InternalDistributedMember> tmp = null;
synchronized (this.membersLock) {
// Note test is under membersLock
@@ -3150,7 +3141,7 @@ public class DistributionManager
* the distributed cache.
*
*/
- private void handleConsoleStartup(InternalDistributedMember theId, Serializable directChannel) {
+ private void handleConsoleStartup(InternalDistributedMember theId) {
// if we have an all listener then notify it NOW.
HashSet tmp = null;
synchronized (this.membersLock) {
@@ -4400,12 +4391,12 @@ public class DistributionManager
handleIncomingDMsg(message);
}
- public void newMemberConnected(InternalDistributedMember member, Stub stub) {
+ public void newMemberConnected(InternalDistributedMember member) {
// Do not elect the elder here as surprise members invoke this callback
// without holding the view lock. That can cause a race condition and
// subsequent deadlock (#45566). Elder selection is now done when a view
// is installed.
- dm.addNewMember(member, stub);
+ dm.addNewMember(member);
}
public void memberDeparted(InternalDistributedMember theId, boolean crashed, String reason) {
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/38dd3ed8/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/HighPriorityAckedMessage.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/HighPriorityAckedMessage.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/HighPriorityAckedMessage.java
index cf0a856..66bd9bb 100755
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/HighPriorityAckedMessage.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/HighPriorityAckedMessage.java
@@ -69,7 +69,9 @@ public final class HighPriorityAckedMessage extends HighPriorityDistributionMess
if (ds != null) {
this.originDm = (DistributionManager)ds.getDistributionManager();
}
- this.id = this.originDm.getDistributionManagerId();
+ if (this.originDm != null) {
+ this.id = this.originDm.getDistributionManagerId();
+ }
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/38dd3ed8/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/StartupMessage.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/StartupMessage.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/StartupMessage.java
index 612d3fa..96f8b60 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/StartupMessage.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/StartupMessage.java
@@ -46,8 +46,6 @@ import com.gemstone.gemfire.internal.tcp.Stub;
public final class StartupMessage extends HighPriorityDistributionMessage implements AdminMessageType {
private static final Logger logger = LogService.getLogger();
- /** A stub for the direct channel for this manager */
- private Stub directChannel;
private String version = GemFireVersion.getGemFireVersion(); // added for bug 29005
private int replyProcessorId;
private boolean isMcastEnabled;
@@ -102,13 +100,6 @@ public final class StartupMessage extends HighPriorityDistributionMessage implem
/////////////////////// Instance Methods ///////////////////////
/**
- * Sets the id of the distribution manager that is starting up
- */
- void setDirectChannel(Stub directChannel) {
- this.directChannel = directChannel;
- }
-
- /**
* Sets the reply processor for this message
*/
void setReplyProcessorId(int proc) {
@@ -319,10 +310,6 @@ public final class StartupMessage extends HighPriorityDistributionMessage implem
@Override
public void toData(DataOutput out) throws IOException {
super.toData(out);
- out.writeBoolean(this.directChannel != null);
- if (this.directChannel != null) {
- InternalDataSerializer.invokeToData(this.directChannel, out);
- }
boolean pre9_0_0_0 = InternalDataSerializer.
getVersionForDataStream(out).compareTo(Version.GFE_90) < 0;
@@ -391,12 +378,6 @@ public final class StartupMessage extends HighPriorityDistributionMessage implem
public void fromData(DataInput in)
throws IOException, ClassNotFoundException {
super.fromData(in);
- boolean hasDirectChannel = in.readBoolean();
- if (hasDirectChannel) {
- this.directChannel = Stub.createFromData(in);
- } else {
- this.directChannel = null;
- }
boolean pre9_0_0_0 = InternalDataSerializer.
getVersionForDataStream(in).compareTo(Version.GFE_90) < 0;
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/38dd3ed8/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/StartupOperation.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/StartupOperation.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/StartupOperation.java
index 6d0ccb7..721f95b 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/StartupOperation.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/StartupOperation.java
@@ -65,7 +65,6 @@ public class StartupOperation {
msg.setDistributedSystemId(dm.getConfig().getDistributedSystemId());
msg.setRedundancyZone(redundancyZone);
msg.setEnforceUniqueZone(enforceUniqueZone);
- msg.setDirectChannel(dm.getDirectChannel());
msg.setMcastEnabled(transport.isMcastEnabled());
msg.setMcastPort(dm.getSystem().getOriginalConfig().getMcastPort());
msg.setMcastHostAddress(dm.getSystem().getOriginalConfig().getMcastAddress());
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/38dd3ed8/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/direct/DirectChannel.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/direct/DirectChannel.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/direct/DirectChannel.java
index fb7572a..f84813e 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/direct/DirectChannel.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/direct/DirectChannel.java
@@ -91,7 +91,7 @@ public class DirectChannel {
private volatile boolean disconnectCompleted = true;
/** this is the DistributionManager, most of the time */
- private final DistributedMembershipListener receiver;
+ private final DirectChannelListener receiver;
private final InetAddress address;
@@ -142,17 +142,17 @@ public class DirectChannel {
/**
* @param mgr
- * @param dm
+ * @param listener
* @param dc
* @param unused
* @throws ConnectionException
*/
- public DirectChannel(MembershipManager mgr, DistributedMembershipListener dm,
- DistributionConfig dc, Properties unused)
+ public DirectChannel(MembershipManager mgr, DirectChannelListener listener,
+ DistributionConfig dc)
throws ConnectionException {
- this.receiver = dm;
+ this.receiver = listener;
- this.address = initAddress(dm, dc);
+ this.address = initAddress(dc);
boolean isBindAddress = dc.getBindAddress() != null;
try {
int port = Integer.getInteger("tcpServerPort", 0).intValue();
@@ -889,7 +889,7 @@ public class DirectChannel {
}
/** returns the receiver to which this DirectChannel is delivering messages */
- protected DistributedMembershipListener getReceiver() {
+ protected DirectChannelListener getReceiver() {
return receiver;
}
@@ -909,7 +909,7 @@ public class DirectChannel {
return this.conduit;
}
- private InetAddress initAddress(DistributedMembershipListener dm, DistributionConfig dc) {
+ private InetAddress initAddress(DistributionConfig dc) {
String bindAddress = dc.getBindAddress();
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/38dd3ed8/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/direct/DirectChannelListener.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/direct/DirectChannelListener.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/direct/DirectChannelListener.java
new file mode 100755
index 0000000..28f481b
--- /dev/null
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/direct/DirectChannelListener.java
@@ -0,0 +1,22 @@
+package com.gemstone.gemfire.distributed.internal.direct;
+
+import com.gemstone.gemfire.distributed.internal.DistributionManager;
+import com.gemstone.gemfire.distributed.internal.DistributionMessage;
+
+public interface DirectChannelListener {
+
+ /**
+ * Event indicating a message has been delivered that we need to process.
+ *
+ * @param o the message that should be processed.
+ */
+ public void messageReceived(DistributionMessage o);
+
+
+ /**
+ * Return the distribution manager for this receiver
+ * @return the distribution manager
+ */
+ public DistributionManager getDM();
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/38dd3ed8/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/DistributedMembershipListener.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/DistributedMembershipListener.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/DistributedMembershipListener.java
index 5436b47..acfd6ba 100755
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/DistributedMembershipListener.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/DistributedMembershipListener.java
@@ -21,8 +21,9 @@ import java.util.Set;
import com.gemstone.gemfire.distributed.internal.DistributionManager;
import com.gemstone.gemfire.distributed.internal.DistributionMessage;
+import com.gemstone.gemfire.distributed.internal.direct.DirectChannelListener;
-public interface DistributedMembershipListener {
+public interface DistributedMembershipListener extends DirectChannelListener {
/** this method is invoked when the processing of a new view is completed */
public void viewInstalled(NetView view);
@@ -33,10 +34,8 @@ public interface DistributedMembershipListener {
/**
* Event indicating that a new member has joined the system.
* @param m the new member
- * @param stub the stub, if any, representing communication to this member
*/
- public void newMemberConnected(InternalDistributedMember m,
- com.gemstone.gemfire.internal.tcp.Stub stub);
+ public void newMemberConnected(InternalDistributedMember m);
/**
* Event indicating that a member has left the system
@@ -83,10 +82,4 @@ public interface DistributedMembershipListener {
*/
public String toString();
- /**
- * Return the distribution manager for this receiver
- * @return the distribution manager
- */
- public DistributionManager getDM();
-
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/38dd3ed8/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/MemberFactory.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/MemberFactory.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/MemberFactory.java
index c8a23e8..250a9a3 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/MemberFactory.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/MemberFactory.java
@@ -38,14 +38,6 @@ public class MemberFactory {
private static final MemberServices services = new GMSMemberFactory();
/**
- * Return a blank NetMember (used by externalization)
- * @return the new NetMember
- */
- static public NetMember newNetMember() {
- return services.newNetMember();
- }
-
- /**
* Return a new NetMember, possibly for a different host
*
* @param i the name of the host for the specified NetMember, the current host (hopefully)
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/38dd3ed8/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/MemberServices.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/MemberServices.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/MemberServices.java
index 8fb302e..3fb6ef2 100755
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/MemberServices.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/MemberServices.java
@@ -35,12 +35,6 @@ import com.gemstone.gemfire.internal.admin.remote.RemoteTransportConfig;
public interface MemberServices {
/**
- * Return a blank NetMember (used by externalization)
- * @return the new NetMember
- */
- public abstract NetMember newNetMember();
-
- /**
* Return a new NetMember, possibly for a different host
*
* @param i the name of the host for the specified NetMember, the current host (hopefully)
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/38dd3ed8/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/MembershipManager.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/MembershipManager.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/MembershipManager.java
index f72e6e2..54b82a7 100755
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/MembershipManager.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/MembershipManager.java
@@ -121,13 +121,6 @@ public interface MembershipManager {
*/
public void startEventProcessing();
- /**
- * Return the underlying proxy object, if any, associated with this
- * local side of this connection.
- *
- * @return the Stub
- */
- public Stub getDirectChannel();
/**
* @param destinations list of members to send the message to. A list of
@@ -345,10 +338,4 @@ public interface MembershipManager {
*/
public void releaseQuorumChecker(QuorumChecker checker);
- /**
- * sets the log writer for authentication logging
- * @param writer
- */
- public void setSecurityLogWriter(InternalLogWriter writer);
-
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/38dd3ed8/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/GMSMember.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/GMSMember.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/GMSMember.java
index e7c9315..f4784ed 100755
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/GMSMember.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/GMSMember.java
@@ -93,28 +93,6 @@ public class GMSMember implements NetMember, DataSerializableFixedID {
}
/**
- * This is the only constructor to refer to a CacheMember other
- * than the current host.
- */
- public GMSMember(GMSMember m) {
- udpPort=m.udpPort;
- preferredForCoordinator=m.preferredForCoordinator;
- splitBrainEnabled=m.splitBrainEnabled;
- memberWeight=m.memberWeight;
- inetAddr=m.inetAddr;
- processId=m.processId;
- vmKind=m.vmKind;
- vmViewId=m.vmViewId;
- directPort=m.directPort;
- name=m.name;
- durableClientAttributes=m.durableClientAttributes;
- groups=m.groups;
- versionOrdinal=m.versionOrdinal;
- uuidLSBs=m.uuidLSBs;
- uuidMSBs=m.uuidMSBs;
- }
-
- /**
* Create a CacheMember referring to the current host (as defined by
* the given string).
*
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/38dd3ed8/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/GMSMemberFactory.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/GMSMemberFactory.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/GMSMemberFactory.java
index 36311a2..2063d7c 100755
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/GMSMemberFactory.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/GMSMemberFactory.java
@@ -95,17 +95,6 @@ public class GMSMemberFactory implements MemberServices {
return newNetMember(inetAddr, p);
}
- /**
- * Return a new Member
- *
- * Used by externalization only.
- *
- * @return blank member for use with externalization
- */
- public NetMember newNetMember() {
- return new GMSMember();
- }
-
public MembershipManager newMembershipManager(DistributedMembershipListener listener,
DistributionConfig config,
RemoteTransportConfig transport, DMStats stats) throws DistributionException
@@ -140,4 +129,5 @@ public class GMSMemberFactory implements MemberServices {
return new GMSLocator(bindAddress, stateFile, locatorString, usePreferredCoordinators, networkPartitionDetectionEnabled, stats);
}
+
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/38dd3ed8/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 df8847c..6478c70 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
@@ -136,5 +136,24 @@ public class GMSUtil {
}
+ /** compareTo for InetAddresses */
+ public static int compareAddresses(InetAddress one, InetAddress two) {
+ byte[] oneBytes = one.getAddress();
+ byte[] twoBytes = two.getAddress();
+
+ if (oneBytes != twoBytes) {
+ for (int i = 0; i < oneBytes.length; i++) {
+ if (i >= twoBytes.length)
+ return -1; // same as far as they go, but shorter...
+ if (oneBytes[i] < twoBytes[i])
+ return -1;
+ if (oneBytes[i] > twoBytes[i])
+ return 1;
+ }
+ if (oneBytes.length > twoBytes.length)
+ return 1; // same as far as they go, but longer...
+ }
+ return 0;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/38dd3ed8/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/fd/GMSHealthMonitor.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/fd/GMSHealthMonitor.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/fd/GMSHealthMonitor.java
index bd60236..fcda1a0 100755
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/fd/GMSHealthMonitor.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/fd/GMSHealthMonitor.java
@@ -419,7 +419,7 @@ public class GMSHealthMonitor implements HealthMonitor, MessageHandler {
* @return
*/
private boolean doCheckMember(InternalDistributedMember member) {
- if (playingDead) {
+ if (playingDead || beingSick) {
// a member playingDead should not be sending messages to other
// members, so we avoid sending heartbeat requests or suspect
// messages by returning true.
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/38dd3ed8/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 91c32ae..3a00f62 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
@@ -605,7 +605,7 @@ public class JGroupsMessenger implements Messenger {
}
}
if (problem != null) {
- if (services.getManager().getShutdownCause() != null) {
+ if (services.getShutdownCause() != null) {
Throwable cause = services.getShutdownCause();
// If ForcedDisconnectException occurred then report it as actual
// problem.
@@ -901,13 +901,6 @@ public class JGroupsMessenger implements Messenger {
}
/**
- * returns the JChannel for test verification
- */
- public JChannel getJGroupsChannel() {
- return this.myChannel;
- }
-
- /**
* for unit testing we need to replace UDP with a fake UDP protocol
*/
public void setJGroupsStackConfigForTesting(String config) {
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/38dd3ed8/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 e7c937d..4e108be 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
@@ -65,12 +65,14 @@ import com.gemstone.gemfire.distributed.internal.SizeableRunnable;
import com.gemstone.gemfire.distributed.internal.StartupMessage;
import com.gemstone.gemfire.distributed.internal.ThrottlingMemLinkedQueueWithDMStats;
import com.gemstone.gemfire.distributed.internal.direct.DirectChannel;
+import com.gemstone.gemfire.distributed.internal.direct.DirectChannelListener;
import com.gemstone.gemfire.distributed.internal.membership.DistributedMembershipListener;
import com.gemstone.gemfire.distributed.internal.membership.InternalDistributedMember;
import com.gemstone.gemfire.distributed.internal.membership.MembershipManager;
import com.gemstone.gemfire.distributed.internal.membership.MembershipTestHook;
import com.gemstone.gemfire.distributed.internal.membership.NetView;
import com.gemstone.gemfire.distributed.internal.membership.QuorumChecker;
+import com.gemstone.gemfire.distributed.internal.membership.gms.GMSUtil;
import com.gemstone.gemfire.distributed.internal.membership.gms.Services;
import com.gemstone.gemfire.distributed.internal.membership.gms.SuspectMember;
import com.gemstone.gemfire.distributed.internal.membership.gms.fd.GMSHealthMonitor;
@@ -85,7 +87,6 @@ import com.gemstone.gemfire.internal.cache.GemFireCacheImpl;
import com.gemstone.gemfire.internal.cache.xmlcache.CacheServerCreation;
import com.gemstone.gemfire.internal.cache.xmlcache.CacheXmlGenerator;
import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
-import com.gemstone.gemfire.internal.logging.InternalLogWriter;
import com.gemstone.gemfire.internal.logging.log4j.AlertAppender;
import com.gemstone.gemfire.internal.logging.log4j.LocalizedMessage;
import com.gemstone.gemfire.internal.logging.log4j.LogMarker;
@@ -134,18 +135,16 @@ public class GMSMembershipManager implements MembershipManager, Manager
}
static class StartupEvent {
- static final int DEPARTURE = 1;
- static final int CONNECT = 2;
- static final int VIEW = 3;
- static final int MESSAGE = 4;
+ static final int SURPRISE_CONNECT = 1;
+ static final int VIEW = 2;
+ static final int MESSAGE = 3;
/**
* indicates whether the event is a departure, a surprise connect
* (i.e., before the view message arrived), a view, or a regular
* message
*
- * @see #DEPARTURE
- * @see #CONNECT
+ * @see #SURPRISE_CONNECT
* @see #VIEW
* @see #MESSAGE
*/
@@ -164,20 +163,7 @@ public class GMSMembershipManager implements MembershipManager, Manager
StringBuffer sb = new StringBuffer();
sb.append("kind=");
switch (kind) {
- case DEPARTURE:
- sb.append("departure; member = <")
- .append(member)
- .append(">; crashed = ")
- .append(crashed)
- .append("; reason = ");
- if (reason != null && (reason.indexOf("NoSuchMemberException") >= 0)) {
- sb.append(LocalizedStrings.GroupMembershipService_TCPIP_CONNECTIONS_CLOSED.toLocalizedString());
- }
- else {
- sb.append(reason);
- }
- break;
- case CONNECT:
+ case SURPRISE_CONNECT:
sb.append("connect; member = <" + member + ">; stub = " + stub);
break;
case VIEW:
@@ -193,42 +179,23 @@ public class GMSMembershipManager implements MembershipManager, Manager
}
return sb.toString();
}
- /**
- * Create a departure event
- * @param dm the member that left
- * @param crashed true if this member crashed
- * @param reason reason string, esp. if crashed
- */
- StartupEvent(InternalDistributedMember dm, boolean crashed, String reason) {
- this.kind = DEPARTURE;
- this.member = dm;
- this.crashed = crashed;
- this.reason = reason;
- }
- /**
- * Indicate if this is a departure
- * @return true if this is a departure event
- */
- boolean isDepartureEvent() {
- return this.kind == DEPARTURE;
- }
/**
- * Create a connect event
+ * Create a surprise connect event
* @param member the member connecting
* @param id the stub
*/
StartupEvent(final InternalDistributedMember member, final Stub id) {
- this.kind = CONNECT;
+ this.kind = SURPRISE_CONNECT;
this.member = member;
this.stub = id;
}
/**
- * Indicate if this is a connect event
+ * Indicate if this is a surprise connect event
* @return true if this is a connect event
*/
- boolean isConnect() {
- return this.kind == CONNECT;
+ boolean isSurpriseConnect() {
+ return this.kind == SURPRISE_CONNECT;
}
/**
@@ -398,11 +365,6 @@ public class GMSMembershipManager implements MembershipManager, Manager
*/
private final long suspectMemberTimeout = 180000;
- /** sleep period, in millis, that the user of this manager should slumber after creating
- the manager. This is advice from the JChannel itself when it detects a concurrent
- startup race condition that requires a settling period. */
- private long channelPause = 0;
-
/**
* Length of time, in seconds, that a member is retained in the zombie set
*
@@ -451,10 +413,10 @@ public class GMSMembershipManager implements MembershipManager, Manager
* @author jpenney
*
*/
- class MyDCReceiver implements DistributedMembershipListener
+ class MyDCReceiver implements DirectChannelListener
{
- DistributedMembershipListener upCall;
+ DirectChannelListener upCall;
/**
* Don't provide events until the caller has told us we are ready.
@@ -465,60 +427,17 @@ public class GMSMembershipManager implements MembershipManager, Manager
* client; we don't need to put this check before every call...
*
*/
- MyDCReceiver(DistributedMembershipListener up) {
+ MyDCReceiver(DirectChannelListener up) {
upCall = up;
}
- public void messageReceived(DistributionMessage msg)
- {
+ public void messageReceived(DistributionMessage msg) {
// bug 36851 - notify failure detection that we've had contact from a member
services.getHealthMonitor().contactedBy(msg.getSender());
handleOrDeferMessage(msg);
}
- public void newMemberConnected(final InternalDistributedMember member, final Stub id)
- {
- handleOrDeferSurpriseConnect(member, id);
- }
-
- public void memberDeparted(InternalDistributedMember id, boolean crashed, String reason)
- {
- try {
- handleOrDeferRemove(id, crashed, reason);
- }
- catch (DistributedSystemDisconnectedException ignore) {
- // ignore
- }
- catch (RuntimeException e) {
- logger.warn(LocalizedMessage.create(LocalizedStrings.GroupMembershipService_MEMBERSHIP_ERROR_HANDLING_MEMBER_DEPARTURE__0), e);
- }
- }
-
- public void quorumLost(Set<InternalDistributedMember> failures, List<InternalDistributedMember> remaining) {
- }
-
- public void memberSuspect(InternalDistributedMember suspect, InternalDistributedMember whoSuspected) {
- // the direct channel isn't currently a source of suspect events, though
- // it does request initiation of suspicion through the membership
- // manager
- }
-
- public boolean isShutdownMsgSent()
- {
- return upCall.isShutdownMsgSent();
- }
-
- public void membershipFailure(String reason, Throwable t)
- {
- upCall.membershipFailure(reason, t);
- }
-
- public void viewInstalled(NetView view) {
- upCall.viewInstalled(view);
- }
-
- public DistributionManager getDM()
- {
+ public DistributionManager getDM() {
return upCall.getDM();
}
@@ -547,9 +466,6 @@ public class GMSMembershipManager implements MembershipManager, Manager
msg.append("Membership: Processing view ");
msg.append(newView);
msg.append("} on " + address.toString());
- if (logger.isDebugEnabled()) {
- logger.debug(LogMarker.DM_VIEWS, msg);
- }
if (!newView.contains(address)) {
logger.info(LocalizedMessage.create(
LocalizedStrings.GroupMembershipService_THE_MEMBER_WITH_ID_0_IS_NO_LONGER_IN_MY_OWN_VIEW_1,
@@ -584,10 +500,6 @@ public class GMSMembershipManager implements MembershipManager, Manager
if (newViewId < latestViewId) {
// ignore this view since it is old news
- if (newViewId < latestViewId && logger.isDebugEnabled(LogMarker.DISTRIBUTION_VIEWS)) {
- logger.debug(LogMarker.DISTRIBUTION_VIEWS, "Membership: Ignoring view (with id {}) since it is older than the last view (with id {}); ignoredView={}",
- newViewId, latestViewId, newView);
- }
return;
}
@@ -641,8 +553,6 @@ public class GMSMembershipManager implements MembershipManager, Manager
}
if (shutdownInProgress()) {
- logger.info(LogMarker.DM_VIEWS, LocalizedMessage.create(
- LocalizedStrings.GroupMembershipService_MEMBERSHIP_SHUNNING_MEMBER__0__DURING_OUR_SHUTDOWN, m));
addShunnedMember(m);
continue; // no additions processed after shutdown begins
} else {
@@ -655,7 +565,7 @@ public class GMSMembershipManager implements MembershipManager, Manager
logger.info(LocalizedMessage.create(LocalizedStrings.GroupMembershipService_MEMBERSHIP_PROCESSING_ADDITION__0_, m));
try {
- listener.newMemberConnected(m, getStubForMember(m));
+ listener.newMemberConnected(m);
}
catch (VirtualMachineError err) {
SystemFailure.initiateFailure(err);
@@ -690,8 +600,6 @@ public class GMSMembershipManager implements MembershipManager, Manager
}
try {
- logger.info(LogMarker.DM_VIEWS, LocalizedMessage.create(
- LocalizedStrings.GroupMembershipService_MEMBERSHIP_PROCESSING_DEPARTING_MEMBER__0_, m));
removeWithViewLock(m,
newView.getCrashedMembers().contains(m) || suspectedMembers.containsKey(m)
, "departed membership view");
@@ -741,9 +649,6 @@ public class GMSMembershipManager implements MembershipManager, Manager
if (birthtime.longValue() < oldestAllowed) {
InternalDistributedMember m = (InternalDistributedMember)entry.getKey();
it.remove();
- if (logger.isTraceEnabled(LogMarker.DISTRIBUTION_VIEWS)) {
- logger.trace(LogMarker.DISTRIBUTION_VIEWS, "Membership: expiring suspect member <{}>", m);
- }
}
}
try {
@@ -755,8 +660,6 @@ public class GMSMembershipManager implements MembershipManager, Manager
} finally {
latestViewLock.writeLock().unlock();
}
- logger.info(LogMarker.DM_VIEWS, LocalizedMessage.create(
- LocalizedStrings.GroupMembershipService_MEMBERSHIP_FINISHED_VIEW_PROCESSING_VIEWID___0, Long.valueOf(newViewId)));
}
/**
@@ -869,7 +772,7 @@ public class GMSMembershipManager implements MembershipManager, Manager
int dcPort = 0;
if (!tcpDisabled) {
- directChannel = new DirectChannel(this, dcReceiver, config, null);
+ directChannel = new DirectChannel(this, dcReceiver, config);
dcPort = directChannel.getPort();
}
@@ -971,23 +874,6 @@ public class GMSMembershipManager implements MembershipManager, Manager
}
/**
- * Remove a member, or queue a startup operation to do so
- * @param dm the member to shun
- * @param crashed true if crashed
- * @param reason the reason, esp. if crashed
- */
- protected void handleOrDeferRemove(InternalDistributedMember dm,
- boolean crashed, String reason) {
- synchronized(startupLock) {
- if (!processingEvents) {
- startupMessages.add(new StartupEvent(dm, crashed, reason));
- return;
- }
- }
- removeMember(dm, crashed, reason);
- }
-
- /**
* Remove a member. {@link #latestViewLock} must be held
* before this method is called. If member is not already shunned,
* the uplevel event handler is invoked.
@@ -1007,10 +893,6 @@ public class GMSMembershipManager implements MembershipManager, Manager
return; // Explicit deletion, no upcall.
}
- if (logger.isTraceEnabled(LogMarker.DISTRIBUTION_VIEWS)) {
- logger.trace(LogMarker.DISTRIBUTION_VIEWS, "Membership: dispatching uplevel departure event for < {} >", dm);
- }
-
try {
listener.memberDeparted(dm, crashed, reason);
}
@@ -1020,48 +902,12 @@ public class GMSMembershipManager implements MembershipManager, Manager
}
/**
- * Automatic removal of a member (for internal
- * use only). Write-locks {@link #latestViewLock} and then deletes
- * the member.
- *
- * @param dm
- * @param crashed
- * @param reason
- */
- protected void removeMember(InternalDistributedMember dm,
- boolean crashed, String reason)
- {
- if (logger.isTraceEnabled(LogMarker.DISTRIBUTION_VIEWS)) {
- StringBuffer sb = new StringBuffer(200);
- sb.append("Membership: removing <")
- .append(dm)
- .append(">; crashed = ")
- .append(crashed)
- .append("; reason = ");
- if (reason != null && (reason.indexOf("NoSuchMemberException") >= 0)) {
- sb.append("tcp/ip connections closed");
- }
- else {
- sb.append(reason);
- }
- logger.trace(LogMarker.DISTRIBUTION_VIEWS, sb);
- }
- latestViewLock.writeLock().lock();
- try {
- removeWithViewLock(dm, crashed, reason);
- } finally {
- latestViewLock.writeLock().unlock();
- }
- }
-
-
- /**
* Process a surprise connect event, or place it on the startup queue.
* @param member the member
* @param stub its stub
*/
- protected void handleOrDeferSurpriseConnect(InternalDistributedMember member,
- Stub stub) {
+ protected void handleOrDeferSurpriseConnect(InternalDistributedMember member) {
+ Stub stub = new Stub(member.getInetAddress(), member.getDirectChannelPort(), member.getVmViewId());
synchronized (startupLock) {
if (!processingEvents) {
startupMessages.add(new StartupEvent(member, stub));
@@ -1113,7 +959,7 @@ public class GMSMembershipManager implements MembershipManager, Manager
return true;
}
if (member.getVmViewId() < 0) {
- logger.warn("attempt to add a surprise member that has not yet joined the distributed system: " + member, new Exception("stack trace"));
+ logger.warn("adding a surprise member that has not yet joined the distributed system: " + member, new Exception("stack trace"));
}
if (latestView.getViewId() > member.getVmViewId()) {
// tell the process that it should shut down distribution.
@@ -1133,20 +979,11 @@ public class GMSMembershipManager implements MembershipManager, Manager
return false;
}
- if (logger.isTraceEnabled(LogMarker.DISTRIBUTION_VIEWS)) {
- logger.trace(LogMarker.DISTRIBUTION_VIEWS, "Membership: Received message from surprise member: <{}>. My view number is {} it is {}",
- member, latestView.getViewId(), member.getVmViewId());
- }
-
// Adding him to this set ensures we won't remove him if a new
// view comes in and he's still not visible.
surpriseMembers.put(member, Long.valueOf(System.currentTimeMillis()));
if (shutdownInProgress()) {
- if (logger.isTraceEnabled(LogMarker.DISTRIBUTION_VIEWS)) {
- logger.trace(LogMarker.DISTRIBUTION_VIEWS, "Membership: new member during shutdown ignored: <{}>", member);
- }
-
// Force disconnect, esp. the TCPConduit
String msg = LocalizedStrings.GroupMembershipService_THIS_DISTRIBUTED_SYSTEM_IS_SHUTTING_DOWN.toLocalizedString();
if (directChannel != null) {
@@ -1166,10 +1003,6 @@ public class GMSMembershipManager implements MembershipManager, Manager
} else {
// Now that we're sure the member is new, add them.
- if (logger.isTraceEnabled(LogMarker.DM_VIEWS)) {
- logger.trace(LogMarker.DM_VIEWS, "Membership: Processing surprise addition <{}>", member);
- }
-
// make sure the surprise-member cleanup task is running
if (this.cleanupTimer == null) {
startCleanupTimer();
@@ -1206,7 +1039,7 @@ public class GMSMembershipManager implements MembershipManager, Manager
logger.warn(LocalizedMessage.create(
LocalizedStrings.GroupMembershipService_MEMBERSHIP_IGNORING_SURPRISE_CONNECT_FROM_SHUNNED_MEMBER_0, member));
} else {
- listener.newMemberConnected(member, s);
+ listener.newMemberConnected(member);
}
return !warn;
}
@@ -1364,10 +1197,6 @@ public class GMSMembershipManager implements MembershipManager, Manager
// to avoid blocking a reader thread
NetView newView = viewArg;
long newId = viewArg.getViewId();
- if (logger.isTraceEnabled(LogMarker.DM_VIEWS)) {
- logger.trace(LogMarker.DM_VIEWS, "Membership: queuing new view for processing, id = {}, view = {}",
- newId, newView);
- }
LocalViewMessage v = new LocalViewMessage(address, newId, newView,
GMSMembershipManager.this);
@@ -1445,10 +1274,7 @@ public class GMSMembershipManager implements MembershipManager, Manager
else if (o.isGmsView()) { // view event
processView(o.gmsView.getViewId(), o.gmsView);
}
- else if (o.isDepartureEvent()) { // departure
- removeMember(o.member, o.crashed, o.reason);
- }
- else if (o.isConnect()) { // connect
+ else if (o.isSurpriseConnect()) { // connect
processSurpriseConnect(o.member, o.stub);
}
@@ -1552,6 +1378,12 @@ public class GMSMembershipManager implements MembershipManager, Manager
}
}
+ /**
+ * for testing we need to validate the startup event list
+ */
+ public List<StartupEvent> getStartupEvents() {
+ return this.startupMessages;
+ }
public ReadWriteLock getViewLock() {
return this.latestViewLock;
@@ -1639,19 +1471,7 @@ public class GMSMembershipManager implements MembershipManager, Manager
return services;
}
- public void postConnect()
- {
- if (channelPause > 0) {
- logger.info(LocalizedMessage.create(
- LocalizedStrings.GroupMembershipService_MEMBERSHIP_PAUSING_TO_ALLOW_OTHER_CONCURRENT_PROCESSES_TO_JOIN_THE_DISTRIBUTED_SYSTEM));
- try {
- Thread.sleep(channelPause);
- }
- catch (InterruptedException ie) {
- Thread.currentThread().interrupt();
- }
- channelPause = 0;
- }
+ public void postConnect() {
}
/**
@@ -1936,7 +1756,7 @@ public class GMSMembershipManager implements MembershipManager, Manager
* all received it)
* @throws NotSerializableException if the message is not serializable
*/
- private Set<InternalDistributedMember> directChannelSend(InternalDistributedMember[] destinations,
+ protected Set<InternalDistributedMember> directChannelSend(InternalDistributedMember[] destinations,
DistributionMessage content,
DMStats theStats)
throws NotSerializableException
@@ -2122,7 +1942,7 @@ public class GMSMembershipManager implements MembershipManager, Manager
if (msg instanceof AdminMessageType
&& this.shutdownInProgress) {
- // no alerts while shutting down - this can cause threads to hang
+ // no admin messages while shutting down - this can cause threads to hang
return new HashSet(Arrays.asList(msg.getRecipients()));
}
@@ -2243,7 +2063,7 @@ public class GMSMembershipManager implements MembershipManager, Manager
if (result == null) {
// it may have not been added to the stub->idm map yet, so check the current view
for (InternalDistributedMember idm: latestView.getMembers()) {
- if (idm.getInetAddress().equals(s.getInetAddress())
+ if (GMSUtil.compareAddresses(idm.getInetAddress(), s.getInetAddress()) == 0
&& idm.getDirectChannelPort() == s.getPort()) {
addChannel(idm, s);
return idm;
@@ -2298,8 +2118,6 @@ public class GMSMembershipManager implements MembershipManager, Manager
*/
protected void destroyMember(final InternalDistributedMember member,
boolean crashed, final String reason) {
- if (logger.isTraceEnabled(LogMarker.DISTRIBUTION_VIEWS))
- logger.trace(LogMarker.DISTRIBUTION_VIEWS, "Membership: destroying < {} >", member);
// Clean up the maps
Stub theChannel = (Stub)memberToStubMap.remove(member);
@@ -2362,18 +2180,6 @@ public class GMSMembershipManager implements MembershipManager, Manager
}
}
- public Stub getDirectChannel()
- {
- Stub result;
- latestViewLock.readLock().lock();
- try {
- result = (Stub)memberToStubMap.get(address);
- } finally {
- latestViewLock.readLock().unlock();
- }
- return result;
- }
-
/**
* Indicate whether the given member is in the zombie list (dead or dying)
* @param m the member in question
@@ -2398,9 +2204,6 @@ public class GMSMembershipManager implements MembershipManager, Manager
return true;
// Oh, it _is_ stale. Remove it while we're here.
- if (logger.isTraceEnabled(LogMarker.DISTRIBUTION_VIEWS)) {
- logger.debug("Membership: no longer shunning < {} >", m);
- }
endShun(m);
return false;
} finally {
@@ -2459,14 +2262,6 @@ public class GMSMembershipManager implements MembershipManager, Manager
return this.surpriseMemberTimeout;
}
- /**
- * returns the shunned member shunset interval, in milliseconds
- */
- public int getShunnedMemberTimeout() {
- return SHUNNED_SUNSET * 1000;
- }
-
-
private boolean endShun(DistributedMember m) {
boolean wasShunned = (shunnedMembers.remove(m) != null);
shunnedAndWarnedMembers.remove(m);
@@ -2490,8 +2285,6 @@ public class GMSMembershipManager implements MembershipManager, Manager
// Update the shunned set.
if (!isShunned(m)) {
shunnedMembers.put(m, Long.valueOf(System.currentTimeMillis()));
- if (logger.isTraceEnabled(LogMarker.DISTRIBUTION_VIEWS))
- logger.trace(LogMarker.DISTRIBUTION_VIEWS, "Membership: added shunned member < {} >", m);
}
// Remove really really old shunned members.
@@ -2535,13 +2328,6 @@ public class GMSMembershipManager implements MembershipManager, Manager
}
/**
- * Set thread-local data for hydra
- */
- public void setThreadLocalData(Object data) {
- Map dataMap = (Map)data;
- }
-
- /**
* for testing verification purposes, this return the port for the
* direct channel, or zero if there is no direct
* channel
@@ -2550,6 +2336,14 @@ public class GMSMembershipManager implements MembershipManager, Manager
return directChannel == null? 0 : directChannel.getPort();
}
+ /**
+ * for mock testing this allows insertion of a DirectChannel mock
+ */
+ protected void setDirectChannel(DirectChannel dc) {
+ this.directChannel = dc;
+ this.tcpDisabled = false;
+ }
+
/* non-thread-owned serial channels and high priority channels are not
* included
*/
@@ -2664,6 +2458,8 @@ public class GMSMembershipManager implements MembershipManager, Manager
}
+ // TODO remove this overly complex method and replace its use with
+ // waitForViewChange using the remote member's view ID
public boolean waitForMembershipCheck(InternalDistributedMember remoteId) {
boolean foundRemoteId = false;
CountDownLatch currentLatch = null;
@@ -2689,8 +2485,6 @@ public class GMSMembershipManager implements MembershipManager, Manager
}
if (!foundRemoteId) {
- // ARB: wait for hardcoded 1000 ms for latch to open.
- // if-stmt precondition: currentLatch is non-null
try {
if (currentLatch.await(membershipCheckTimeout, TimeUnit.MILLISECONDS)) {
foundRemoteId = true;
@@ -2783,9 +2577,7 @@ public class GMSMembershipManager implements MembershipManager, Manager
public synchronized void beSick() {
if (!beingSick) {
beingSick = true;
- if (logger.isInfoEnabled()) {
- logger.info("GroupMembershipService.beSick invoked for {} - simulating sickness", this.address);
- }
+ logger.info("GroupMembershipService.beSick invoked for {} - simulating sickness", this.address);
services.getJoinLeave().beSick();
services.getHealthMonitor().beSick();
if (directChannel != null) {
@@ -2800,9 +2592,7 @@ public class GMSMembershipManager implements MembershipManager, Manager
public synchronized void playDead() {
if (!playingDead) {
playingDead = true;
- if (logger.isDebugEnabled()) {
- logger.debug("GroupMembershipService.playDead invoked for {}", this.address);
- }
+ logger.info("GroupMembershipService.playDead invoked for {}", this.address);
services.getJoinLeave().playDead();
services.getHealthMonitor().playDead();
services.getMessenger().playDead();
@@ -2816,9 +2606,7 @@ public class GMSMembershipManager implements MembershipManager, Manager
if (beingSick || playingDead) {
beingSick = false;
playingDead = false;
- if (logger.isDebugEnabled()) {
- logger.debug("GroupMembershipService.beHealthy invoked for {} - recovering health now", this.address);
- }
+ logger.info("GroupMembershipService.beHealthy invoked for {} - recovering health now", this.address);
if (directChannel != null) {
directChannel.beHealthy();
}
@@ -2955,11 +2743,6 @@ public class GMSMembershipManager implements MembershipManager, Manager
services.getJoinLeave().disableDisconnectOnQuorumLossForTesting();
}
- @Override
- public void setSecurityLogWriter(InternalLogWriter writer) {
- Services.setSecurityLogWriter(writer);
- }
-
/**
* Class <code>BoundedLinkedHashMap</code> is a bounded
@@ -2980,29 +2763,6 @@ public class GMSMembershipManager implements MembershipManager, Manager
/**
* 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) {
@@ -3010,14 +2770,6 @@ public class GMSMembershipManager implements MembershipManager, Manager
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/38dd3ed8/gemfire-core/src/main/java/com/gemstone/gemfire/internal/logging/log4j/LogMarker.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/logging/log4j/LogMarker.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/logging/log4j/LogMarker.java
index 04b4223..3b95ca6 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/logging/log4j/LogMarker.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/logging/log4j/LogMarker.java
@@ -73,7 +73,6 @@ public interface LogMarker {
public static final Marker DISTRIBUTION_BRIDGE_SERVER = MarkerManager.getMarker("DISTRIBUTION_BRIDGE_SERVER").addParents(DISTRIBUTION, BRIDGE_SERVER);
public static final Marker DISTRIBUTION_VIEWS = MarkerManager.getMarker("DISTRIBUTION_VIEWS").addParents(DISTRIBUTION);
public static final Marker DM = MarkerManager.getMarker("DM").addParents(DISTRIBUTION);
- public static final Marker DM_VIEWS = MarkerManager.getMarker("DM_VIEWS").addParents(DM, DISTRIBUTION_VIEWS);
public static final Marker DM_BRIDGE_SERVER = MarkerManager.getMarker("DM_BRIDGE").addParents(BRIDGE_SERVER, DM);
public static final Marker DA = MarkerManager.getMarker("DA").addParents(DISTRIBUTION);
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/38dd3ed8/gemfire-core/src/main/java/com/gemstone/gemfire/internal/tcp/Connection.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/tcp/Connection.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/tcp/Connection.java
index c592133..30962e7 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/tcp/Connection.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/tcp/Connection.java
@@ -771,6 +771,13 @@ public class Connection implements Runnable {
String peerName;
if (this.remoteAddr != null) {
peerName = this.remoteAddr.toString();
+ // late in the life of jdk 1.7 we started seeing connections accepted
+ // when accept() was not even being called. This started causing timeouts
+ // to occur in the handshake threads instead of causing failures in
+ // connection-formation. So, we need to initiate suspect processing here
+ owner.getDM().getMembershipManager().suspectMember(this.remoteAddr,
+ LocalizedStrings.Connection_CONNECTION_HANDSHAKE_WITH_0_TIMED_OUT_AFTER_WAITING_1_MILLISECONDS.toLocalizedString(
+ new Object[] {peerName, Integer.valueOf(HANDSHAKE_TIMEOUT_MS)}));
}
else {
peerName = "socket " + this.socket.getRemoteSocketAddress().toString()
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/38dd3ed8/gemfire-core/src/main/java/com/gemstone/gemfire/internal/tcp/TCPConduit.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/tcp/TCPConduit.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/tcp/TCPConduit.java
index f4fab74..12a03fd 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/tcp/TCPConduit.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/tcp/TCPConduit.java
@@ -645,9 +645,9 @@ public class TCPConduit implements Runnable {
continue;
}
if (inhibitNewConnections) {
- if (logger.isTraceEnabled(LogMarker.QA)) {
- logger.trace(LogMarker.QA, "Test hook: inhibiting acceptance of connection {}", othersock);
- }
+// if (logger.isTraceEnabled(LogMarker.QA)) {
+ logger.info("Test hook: inhibiting acceptance of connection {}", othersock);
+// }
othersock.close();
while (inhibitNewConnections && !stopped) {
this.stopper.checkCancelInProgress(null);
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/38dd3ed8/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 fd5aaa7..91889df 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
@@ -19,6 +19,10 @@ package com.gemstone.gemfire.distributed.internal.membership;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.assertEquals;
+
import java.io.File;
import java.net.InetAddress;
import java.util.ArrayList;
@@ -43,6 +47,7 @@ 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.InternalLocator;
+import com.gemstone.gemfire.distributed.internal.membership.gms.GMSUtil;
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.interfaces.JoinLeave;
@@ -54,13 +59,9 @@ import com.gemstone.gemfire.internal.admin.remote.RemoteTransportConfig;
import com.gemstone.gemfire.test.junit.categories.IntegrationTest;
@Category(IntegrationTest.class)
-public class MembershipJUnitTest extends TestCase {
+public class MembershipJUnitTest {
static Level baseLogLevel;
- public MembershipJUnitTest(String name) {
- super(name);
- }
-
@BeforeClass
public static void setupClass() {
// baseLogLevel = LogService.getBaseLogLevel();
@@ -68,7 +69,7 @@ public class MembershipJUnitTest extends TestCase {
}
@AfterClass
- protected void tearDown() throws Exception {
+ public static void tearDown() throws Exception {
// LogService.setBaseLogLevel(baseLogLevel);
}
@@ -263,11 +264,25 @@ public class MembershipJUnitTest extends TestCase {
GMSJoinLeave joinLeave = new GMSJoinLeave();
try {
joinLeave.init(services);
- fail("expected a GemFireConfigException to be thrown because no locators are configured");
+ throw new Error("expected a GemFireConfigException to be thrown because no locators are configured");
} catch (GemFireConfigException e) {
// expected
}
}
+ /**
+ * test the GMSUtil.formatBytes() method
+ */
+ @Test
+ public void testFormatBytes() throws Exception {
+ byte[] bytes = new byte[200];
+ for (int i=0; i<bytes.length; i++) {
+ bytes[i] = (byte)(i%255);
+ }
+ String str = GMSUtil.formatBytes(bytes, 0, bytes.length);
+ System.out.println(str);
+ assertEquals(600+4, str.length());
+ }
+
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/38dd3ed8/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/internal/membership/gms/locator/GMSLocatorRecoveryJUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/internal/membership/gms/locator/GMSLocatorRecoveryJUnitTest.java b/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/internal/membership/gms/locator/GMSLocatorRecoveryJUnitTest.java
index 2d042fc..7badce6 100644
--- a/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/internal/membership/gms/locator/GMSLocatorRecoveryJUnitTest.java
+++ b/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/internal/membership/gms/locator/GMSLocatorRecoveryJUnitTest.java
@@ -152,8 +152,8 @@ public class GMSLocatorRecoveryJUnitTest {
nonDefault.put(DistributionConfig.MCAST_PORT_NAME, "0");
nonDefault.put(DistributionConfig.LOG_FILE_NAME, "");
nonDefault.put(DistributionConfig.LOG_LEVEL_NAME, "fine");
- nonDefault.put(DistributionConfig.LOCATORS_NAME, localHost.getHostAddress()+'['+port+']');
- nonDefault.put(DistributionConfig.BIND_ADDRESS_NAME, localHost.getHostAddress());
+ nonDefault.put(DistributionConfig.LOCATORS_NAME, localHost.getHostName()+'['+port+']');
+ nonDefault.put(DistributionConfig.BIND_ADDRESS_NAME, localHost.getHostName());
DistributionConfigImpl config = new DistributionConfigImpl(nonDefault);
RemoteTransportConfig transport = new RemoteTransportConfig(config,
DistributionManager.NORMAL_DM_TYPE);
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/38dd3ed8/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/internal/membership/gms/mgr/GMSMembershipManagerJUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/internal/membership/gms/mgr/GMSMembershipManagerJUnitTest.java b/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/internal/membership/gms/mgr/GMSMembershipManagerJUnitTest.java
new file mode 100755
index 0000000..2b59ca5
--- /dev/null
+++ b/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/internal/membership/gms/mgr/GMSMembershipManagerJUnitTest.java
@@ -0,0 +1,370 @@
+/*
+ * 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 com.gemstone.gemfire.distributed.internal.membership.gms.mgr;
+
+import static org.mockito.Mockito.*;
+import static org.junit.Assert.*;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Date;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+import java.util.Timer;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import com.gemstone.gemfire.distributed.DistributedSystemDisconnectedException;
+import com.gemstone.gemfire.distributed.internal.AdminMessageType;
+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.HighPriorityAckedMessage;
+import com.gemstone.gemfire.distributed.internal.HighPriorityDistributionMessage;
+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.NetView;
+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.SuspectMember;
+import com.gemstone.gemfire.distributed.internal.membership.gms.interfaces.Authenticator;
+import com.gemstone.gemfire.distributed.internal.membership.gms.interfaces.HealthMonitor;
+import com.gemstone.gemfire.distributed.internal.membership.gms.interfaces.JoinLeave;
+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.membership.GMSJoinLeave;
+import com.gemstone.gemfire.distributed.internal.membership.gms.mgr.GMSMembershipManager;
+import com.gemstone.gemfire.distributed.internal.membership.gms.mgr.GMSMembershipManager.StartupEvent;
+import com.gemstone.gemfire.internal.AvailablePortHelper;
+import com.gemstone.gemfire.internal.Version;
+import com.gemstone.gemfire.internal.admin.remote.AdminRequest;
+import com.gemstone.gemfire.internal.admin.remote.AdminResponse;
+import com.gemstone.gemfire.internal.admin.remote.AlertListenerMessage;
+import com.gemstone.gemfire.internal.admin.remote.RemoteTransportConfig;
+import com.gemstone.gemfire.internal.tcp.ConnectExceptions;
+import com.gemstone.gemfire.internal.tcp.Stub;
+import com.gemstone.gemfire.test.junit.categories.UnitTest;
+
+@Category(UnitTest.class)
+public class GMSMembershipManagerJUnitTest {
+ private Services services;
+ private ServiceConfig mockConfig;
+ private DistributionConfig distConfig;
+ private Authenticator authenticator;
+ private HealthMonitor healthMonitor;
+ private InternalDistributedMember myMemberId;
+ private InternalDistributedMember[] mockMembers;
+ private Messenger messenger;
+ private JoinLeave joinLeave;
+ private Stopper stopper;
+ DistributedMembershipListener listener;
+ private GMSMembershipManager manager;
+ private List<InternalDistributedMember> members;
+ private Set<InternalDistributedMember> emptyMembersSet = new HashSet<>();
+ private DirectChannel dc;
+
+ @Before
+ public void initMocks() throws Exception {
+ Properties nonDefault = new Properties();
+ nonDefault.put(DistributionConfig.DISABLE_TCP_NAME, "true");
+ nonDefault.put(DistributionConfig.MCAST_PORT_NAME, "0");
+ nonDefault.put(DistributionConfig.MCAST_TTL_NAME, "0");
+ nonDefault.put(DistributionConfig.LOG_FILE_NAME, "");
+ nonDefault.put(DistributionConfig.LOG_LEVEL_NAME, "fine");
+ nonDefault.put(DistributionConfig.MEMBER_TIMEOUT_NAME, "2000");
+ nonDefault.put(DistributionConfig.LOCATORS_NAME, "localhost[10344]");
+ distConfig = new DistributionConfigImpl(nonDefault);
+ RemoteTransportConfig tconfig = new RemoteTransportConfig(distConfig,
+ DistributionManager.NORMAL_DM_TYPE);
+
+
+ mockConfig = mock(ServiceConfig.class);
+ when(mockConfig.getDistributionConfig()).thenReturn(distConfig);
+ when(mockConfig.getTransport()).thenReturn(tconfig);
+
+ authenticator = mock(Authenticator.class);
+ myMemberId = new InternalDistributedMember("localhost", 8887);
+
+ messenger = mock(Messenger.class);
+ when(messenger.getMemberID()).thenReturn(myMemberId);
+
+ stopper = mock(Stopper.class);
+ when(stopper.isCancelInProgress()).thenReturn(false);
+
+ healthMonitor = mock(HealthMonitor.class);
+ when(healthMonitor.getFailureDetectionPort()).thenReturn(Integer.valueOf(-1));
+
+ joinLeave = mock(JoinLeave.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.getHealthMonitor()).thenReturn(healthMonitor);
+ when(services.getJoinLeave()).thenReturn(joinLeave);
+
+ Timer t = new Timer(true);
+ when(services.getTimer()).thenReturn(t);
+
+ mockMembers = new InternalDistributedMember[5];
+ for (int i = 0; i < mockMembers.length; i++) {
+ mockMembers[i] = new InternalDistributedMember("localhost", 8888 + i);
+ }
+ members = new ArrayList<>(Arrays.asList(mockMembers));
+
+ listener = mock(DistributedMembershipListener.class);
+
+ manager = new GMSMembershipManager(listener);
+ manager.init(services);
+ when(services.getManager()).thenReturn(manager);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ if (manager != null) {
+ manager.stop();
+ manager.stopped();
+ }
+ }
+
+ @Test
+ public void testSendMessage() throws Exception {
+ HighPriorityAckedMessage m = new HighPriorityAckedMessage();
+ m.setRecipient(mockMembers[0]);
+ manager.start();
+ manager.started();
+ manager.installView(new NetView(myMemberId, 1, members, emptyMembersSet, emptyMembersSet));
+ Set<InternalDistributedMember> failures = manager.send(m);
+ verify(messenger).send(m);
+ if (failures != null) {
+ assertEquals(0, failures.size());
+ }
+ }
+
+ @Test
+ public void testSendAdminMessageFailsDuringShutdown() throws Exception {
+ AlertListenerMessage m = AlertListenerMessage.create(mockMembers[0], 1,
+ new Date(System.currentTimeMillis()), "thread", "", 1L, "", "");
+ manager.start();
+ manager.started();
+ manager.installView(new NetView(myMemberId, 1, members, emptyMembersSet, emptyMembersSet));
+ manager.setShutdown();
+ Set<InternalDistributedMember> failures = manager.send(m);
+ verify(messenger, never()).send(m);
+ assertEquals(1, failures.size());
+ assertEquals(mockMembers[0], failures.iterator().next());
+ }
+
+ @Test
+ public void testSendToEmptyListIsRejected() throws Exception {
+ InternalDistributedMember[] emptyList = new InternalDistributedMember[0];
+ HighPriorityAckedMessage m = new HighPriorityAckedMessage();
+ m.setRecipient(mockMembers[0]);
+ manager.start();
+ manager.started();
+ manager.installView(new NetView(myMemberId, 1, members, emptyMembersSet, emptyMembersSet));
+ Set<InternalDistributedMember> failures = manager.send(null, m, null);
+ verify(messenger, never()).send(m);
+ reset(messenger);
+ failures = manager.send(emptyList, m, null);
+ verify(messenger, never()).send(m);
+ }
+
+ @Test
+ public void testStartupEvents() throws Exception {
+ manager.start();
+ manager.started();
+ manager.isJoining = true;
+
+ List<InternalDistributedMember> viewmembers = Arrays.asList(new InternalDistributedMember[] {mockMembers[0], myMemberId});
+ manager.installView(new NetView(myMemberId, 2, viewmembers, emptyMembersSet, emptyMembersSet));
+
+ // add a surprise member that will be shunned due to it's having
+ // an old view ID
+ InternalDistributedMember surpriseMember = mockMembers[2];
+ surpriseMember.setVmViewId(1);
+ manager.handleOrDeferSurpriseConnect(surpriseMember);
+ assertEquals(1, manager.getStartupEvents().size());
+
+ // add a surprise member that will be accepted
+ InternalDistributedMember surpriseMember2 = mockMembers[3];
+ surpriseMember2.setVmViewId(3);
+ manager.handleOrDeferSurpriseConnect(surpriseMember2);
+ assertEquals(2, manager.getStartupEvents().size());
+
+ // suspect a member
+ InternalDistributedMember suspectMember = mockMembers[1];
+ manager.handleOrDeferSuspect(new SuspectMember(mockMembers[0], suspectMember));
+ // suspect messages aren't queued - they're ignored before joining the system
+ assertEquals(2, manager.getStartupEvents().size());
+ verify(listener, never()).memberSuspect(suspectMember, mockMembers[0]);
+
+ HighPriorityAckedMessage m = new HighPriorityAckedMessage();
+ mockMembers[0].setVmViewId(1);
+ m.setRecipient(mockMembers[0]);
+ m.setSender(mockMembers[1]);
+ manager.handleOrDeferMessage(m);
+ assertEquals(3, manager.getStartupEvents().size());
+
+ // this view officially adds surpriseMember2
+ viewmembers = Arrays.asList(
+ new InternalDistributedMember[] {mockMembers[0], myMemberId, surpriseMember2});
+ manager.handleOrDeferViewEvent(new NetView(myMemberId, 3, viewmembers, emptyMembersSet, emptyMembersSet));
+ assertEquals(4, manager.getStartupEvents().size());
+
+ // add a surprise member that will be shunned due to it's having
+ // an old view ID
+ InternalDistributedMember surpriseMember3 = mockMembers[4];
+ surpriseMember.setVmViewId(1);
+ manager.handleOrDeferSurpriseConnect(surpriseMember);
+ assertEquals(5, manager.getStartupEvents().size());
+
+
+ // process a new view after we finish joining but before event processing has started
+ manager.isJoining = false;
+ mockMembers[4].setVmViewId(4);
+ viewmembers = Arrays.asList(new InternalDistributedMember[] {mockMembers[0], myMemberId, surpriseMember2, mockMembers[4]});
+ manager.handleOrDeferViewEvent(new NetView(myMemberId, 4, viewmembers, emptyMembersSet, emptyMembersSet));
+ assertEquals(6, manager.getStartupEvents().size());
+
+ // exercise the toString methods for code coverage
+ for (StartupEvent ev: manager.getStartupEvents()) {
+ ev.toString();
+ }
+
+ manager.startEventProcessing();
+
+ // all startup events should have been processed
+ assertEquals(0, manager.getStartupEvents().size());
+ // the new view should have been installed
+ assertEquals(4, manager.getView().getViewId());
+ // supriseMember2 should have been announced
+ verify(listener).newMemberConnected(surpriseMember2);
+ // supriseMember should have been rejected (old view ID)
+ verify(listener, never()).newMemberConnected(surpriseMember);
+
+ // for code coverage also install a view after we finish joining but before
+ // event processing has started. This should notify the distribution manager
+ // with a LocalViewMessage to process the view
+ reset(listener);
+ manager.handleOrDeferViewEvent(new NetView(myMemberId, 5, viewmembers, emptyMembersSet, emptyMembersSet));
+ assertEquals(0, manager.getStartupEvents().size());
+ verify(listener).messageReceived(isA(LocalViewMessage.class));
+
+ // process a suspect now - it will be passed to the listener
+ reset(listener);
+ suspectMember = mockMembers[1];
+ manager.handleOrDeferSuspect(new SuspectMember(mockMembers[0], suspectMember));
+ verify(listener).memberSuspect(suspectMember, mockMembers[0]);
+
+ InternalDistributedMember mbr = manager.getMemberForStub(new Stub(myMemberId.getInetAddress(), 2033, 20), false);
+ assertTrue(mbr == null);
+ myMemberId.setDirectChannelPort(2033);
+ mbr = manager.getMemberForStub(new Stub(myMemberId.getInetAddress(), 2033, 20), false);
+ assertTrue(mbr != null);
+ assertEquals(mbr, myMemberId);
+ }
+
+ /**
+ * Some tests require a DirectChannel mock
+ */
+ private void setUpDirectChannelMock() throws Exception {
+ dc = mock(DirectChannel.class);
+ when(dc.send(any(GMSMembershipManager.class), any(mockMembers.getClass()), any(DistributionMessage.class), anyInt(), anyInt()))
+ .thenReturn(100);
+
+ manager.start();
+ manager.started();
+
+ manager.setDirectChannel(dc);
+
+ NetView view = new NetView(myMemberId, 1, members, emptyMembersSet, emptyMembersSet);
+ manager.installView(view);
+ when(joinLeave.getView()).thenReturn(view);
+
+ manager.startEventProcessing();
+ }
+
+ @Test
+ public void testDirectChannelSend() throws Exception {
+ setUpDirectChannelMock();
+ HighPriorityAckedMessage m = new HighPriorityAckedMessage();
+ InternalDistributedMember[] recipients = new InternalDistributedMember[] {mockMembers[2], mockMembers[3]};
+ m.setRecipients(Arrays.asList(recipients));
+ Set<InternalDistributedMember> failures = manager.directChannelSend(recipients, m, null);
+ assertTrue(failures == null);
+ verify(dc).send(isA(GMSMembershipManager.class), isA(mockMembers.getClass()), isA(DistributionMessage.class), anyInt(), anyInt());
+ }
+
+ @Test
+ public void testDirectChannelSendFailureToOneRecipient() throws Exception {
+ setUpDirectChannelMock();
+ HighPriorityAckedMessage m = new HighPriorityAckedMessage();
+ InternalDistributedMember[] recipients = new InternalDistributedMember[] {mockMembers[2], mockMembers[3]};
+ m.setRecipients(Arrays.asList(recipients));
+ Set<InternalDistributedMember> failures = manager.directChannelSend(recipients, m, null);
+
+ ConnectExceptions exception = new ConnectExceptions();
+ exception.addFailure(recipients[0], new Exception("testing"));
+ when(dc.send(any(GMSMembershipManager.class), any(mockMembers.getClass()), any(DistributionMessage.class), anyInt(), anyInt()))
+ .thenThrow(exception);
+ failures = manager.directChannelSend(recipients, m, null);
+ assertTrue(failures != null);
+ assertEquals(1, failures.size());
+ assertEquals(recipients[0], failures.iterator().next());
+ }
+
+ @Test
+ public void testDirectChannelSendFailureToAll() throws Exception {
+ setUpDirectChannelMock();
+ HighPriorityAckedMessage m = new HighPriorityAckedMessage();
+ InternalDistributedMember[] recipients = new InternalDistributedMember[] {mockMembers[2], mockMembers[3]};
+ m.setRecipients(Arrays.asList(recipients));
+ Set<InternalDistributedMember> failures = manager.directChannelSend(recipients, m, null);
+ when(dc.send(any(GMSMembershipManager.class), any(mockMembers.getClass()), any(DistributionMessage.class), anyInt(), anyInt()))
+ .thenReturn(0);
+ when(stopper.cancelInProgress()).thenReturn("stopping for test");
+ try {
+ manager.directChannelSend(recipients, m, null);
+ throw new RuntimeException("expected directChannelSend to throw an exception");
+ } catch (DistributedSystemDisconnectedException e) {
+ // expected
+ }
+ }
+
+ @Test
+ public void testDirectChannelSendAllRecipients() throws Exception {
+ setUpDirectChannelMock();
+ HighPriorityAckedMessage m = new HighPriorityAckedMessage();
+ m.setRecipient(DistributionMessage.ALL_RECIPIENTS);
+ assertTrue(m.forAll());
+ Set<InternalDistributedMember> failures = manager.directChannelSend(null, m, null);
+ assertTrue(failures == null);
+ verify(dc).send(isA(GMSMembershipManager.class), isA(mockMembers.getClass()), isA(DistributionMessage.class), anyInt(), anyInt());
+ }
+
+}
+