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 2016/08/05 18:33:26 UTC
[2/2] incubator-geode git commit: GEODE-1726 Clean up compilation
warnings in new GMS
GEODE-1726 Clean up compilation warnings in new GMS
Project: http://git-wip-us.apache.org/repos/asf/incubator-geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-geode/commit/5f840549
Tree: http://git-wip-us.apache.org/repos/asf/incubator-geode/tree/5f840549
Diff: http://git-wip-us.apache.org/repos/asf/incubator-geode/diff/5f840549
Branch: refs/heads/develop
Commit: 5f84054955dc873add294cc0115f9d52e420e168
Parents: f3db3e8
Author: Bruce Schuchardt <bs...@pivotal.io>
Authored: Fri Aug 5 11:26:43 2016 -0700
Committer: Bruce Schuchardt <bs...@pivotal.io>
Committed: Fri Aug 5 11:32:35 2016 -0700
----------------------------------------------------------------------
.../internal/membership/NetView.java | 80 ++--
.../internal/membership/gms/GMSUtil.java | 30 +-
.../internal/membership/gms/ServiceConfig.java | 56 +--
.../internal/membership/gms/Services.java | 3 +-
.../membership/gms/fd/GMSHealthMonitor.java | 218 +++++-----
.../membership/gms/locator/GMSLocator.java | 19 +-
.../membership/gms/membership/GMSJoinLeave.java | 53 ++-
.../gms/messenger/GMSQuorumChecker.java | 43 +-
.../gms/messenger/JGroupsMessenger.java | 77 ++--
.../gms/mgr/GMSMembershipManager.java | 413 +++++++------------
.../gms/membership/GMSJoinLeaveJUnitTest.java | 2 +-
.../sanctionedDataSerializables.txt | 4 +-
.../codeAnalysis/sanctionedSerializables.txt | 2 +-
13 files changed, 387 insertions(+), 613 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/5f840549/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/NetView.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/NetView.java b/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/NetView.java
index 4c33e08..f3c04f7 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/NetView.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/NetView.java
@@ -28,8 +28,8 @@ import java.util.Iterator;
import java.util.List;
import java.util.Random;
import java.util.Set;
+import java.util.stream.*;
-import com.gemstone.gemfire.internal.logging.LogService;
import org.apache.logging.log4j.Logger;
import com.gemstone.gemfire.DataSerializer;
@@ -46,6 +46,7 @@ import com.gemstone.gemfire.internal.Version;
*
* @since GemFire 5.5
*/
+@SuppressWarnings("SynchronizeOnNonFinalField")
public class NetView implements DataSerializableFixedID {
private int viewId;
@@ -60,7 +61,7 @@ public class NetView implements DataSerializableFixedID {
public NetView() {
viewId = 0;
- members = new ArrayList<InternalDistributedMember>(4);
+ members = new ArrayList<>(4);
this.hashedMembers = new HashSet<>(members);
shutdownMembers = Collections.emptySet();
crashedMembers = new HashSet<>();
@@ -70,7 +71,7 @@ public class NetView implements DataSerializableFixedID {
public NetView(InternalDistributedMember creator) {
viewId = 0;
- members = new ArrayList<InternalDistributedMember>(4);
+ members = new ArrayList<>(4);
members.add(creator);
hashedMembers = new HashSet<>(members);
shutdownMembers = new HashSet<>();
@@ -81,7 +82,7 @@ public class NetView implements DataSerializableFixedID {
public NetView(InternalDistributedMember creator, int viewId, List<InternalDistributedMember> members) {
this.viewId = viewId;
- this.members = new ArrayList<InternalDistributedMember>(members);
+ this.members = new ArrayList<>(members);
hashedMembers = new HashSet<>(this.members);
shutdownMembers = new HashSet<>();
crashedMembers = Collections.emptySet();
@@ -90,11 +91,15 @@ public class NetView implements DataSerializableFixedID {
}
- // legacy method for JGMM
+ /**
+ * Test method
+ * @param size size of the view, used for presizing collections
+ * @param viewId the ID of the view
+ */
public NetView(int size, long viewId) {
this.viewId = (int) viewId;
- members = new ArrayList<InternalDistributedMember>(size);
- this.hashedMembers = new HashSet<InternalDistributedMember>();
+ members = new ArrayList<>(size);
+ this.hashedMembers = new HashSet<>();
shutdownMembers = new HashSet<>();
crashedMembers = Collections.emptySet();
creator = null;
@@ -104,18 +109,16 @@ public class NetView implements DataSerializableFixedID {
/**
* Create a new view with the contents of the given view and the
* specified view ID
- * @param other
- * @param viewId
*/
public NetView(NetView other, int viewId) {
this.creator = other.creator;
this.viewId = viewId;
- this.members = new ArrayList<InternalDistributedMember>(other.members);
- this.hashedMembers = new HashSet<InternalDistributedMember>(other.members);
+ this.members = new ArrayList<>(other.members);
+ this.hashedMembers = new HashSet<>(other.members);
this.failureDetectionPorts = new int[other.failureDetectionPorts.length];
System.arraycopy(other.failureDetectionPorts, 0, this.failureDetectionPorts, 0, other.failureDetectionPorts.length);
- this.shutdownMembers = new HashSet<InternalDistributedMember>(other.shutdownMembers);
- this.crashedMembers = new HashSet<InternalDistributedMember>(other.crashedMembers);
+ this.shutdownMembers = new HashSet<>(other.shutdownMembers);
+ this.crashedMembers = new HashSet<>(other.crashedMembers);
}
public NetView(InternalDistributedMember creator, int viewId, List<InternalDistributedMember> mbrs, Set<InternalDistributedMember> shutdowns,
@@ -123,7 +126,7 @@ public class NetView implements DataSerializableFixedID {
this.creator = creator;
this.viewId = viewId;
this.members = mbrs;
- this.hashedMembers = new HashSet<InternalDistributedMember>(mbrs);
+ this.hashedMembers = new HashSet<>(mbrs);
this.shutdownMembers = shutdowns;
this.crashedMembers = crashes;
this.failureDetectionPorts = new int[mbrs.size()+10];
@@ -165,7 +168,6 @@ public class NetView implements DataSerializableFixedID {
/**
* Transfer the failure-detection ports from another view to this one
- * @param otherView
*/
public void setFailureDetectionPorts(NetView otherView) {
int[] ports = otherView.getFailureDetectionPorts();
@@ -209,7 +211,7 @@ public class NetView implements DataSerializableFixedID {
* return members that are i this view but not the given old view
*/
public List<InternalDistributedMember> getNewMembers(NetView olderView) {
- List<InternalDistributedMember> result = new ArrayList<InternalDistributedMember>(members);
+ List<InternalDistributedMember> result = new ArrayList<>(members);
result.removeAll(olderView.getMembers());
return result;
}
@@ -218,12 +220,9 @@ public class NetView implements DataSerializableFixedID {
* return members added in this view
*/
public List<InternalDistributedMember> getNewMembers() {
- List<InternalDistributedMember> result = new ArrayList<InternalDistributedMember>(5);
- for (InternalDistributedMember mbr : this.members) {
- if (mbr.getVmViewId() == this.viewId) {
- result.add(mbr);
- }
- }
+ List<InternalDistributedMember> result = new ArrayList<>(5);
+ result.addAll(this.members.stream().filter(mbr -> mbr.getVmViewId()
+ == this.viewId).collect(Collectors.toList()));
return result;
}
@@ -243,10 +242,6 @@ public class NetView implements DataSerializableFixedID {
this.crashedMembers.addAll(mbr);
}
- public void addShutdownMembers(Set<InternalDistributedMember> mbr) {
- this.shutdownMembers.addAll(mbr);
- }
-
public boolean remove(InternalDistributedMember mbr) {
this.hashedMembers.remove(mbr);
int idx = this.members.indexOf(mbr);
@@ -259,12 +254,11 @@ public class NetView implements DataSerializableFixedID {
public void removeAll(Collection<InternalDistributedMember> ids) {
this.hashedMembers.removeAll(ids);
- for (InternalDistributedMember mbr: ids) {
- remove(mbr);
- }
+ ids.forEach(this::remove);
}
public boolean contains(DistributedMember mbr) {
+ assert mbr instanceof InternalDistributedMember;
return this.hashedMembers.contains(mbr);
}
@@ -325,13 +319,13 @@ public class NetView implements DataSerializableFixedID {
* And local member.
*
* @param filter Suspect member set.
- * @param localAddress
+ * @param localAddress the address of this member
* @param maxNumberDesired number of preferred coordinators to return
* @return list of preferred coordinators
*/
public List<InternalDistributedMember> getPreferredCoordinators(Set<InternalDistributedMember> filter, InternalDistributedMember localAddress, int maxNumberDesired) {
- List<InternalDistributedMember> results = new ArrayList<InternalDistributedMember>();
- List<InternalDistributedMember> notPreferredCoordinatorList = new ArrayList<InternalDistributedMember>();
+ List<InternalDistributedMember> results = new ArrayList<>();
+ List<InternalDistributedMember> notPreferredCoordinatorList = new ArrayList<>();
synchronized (members) {
for (InternalDistributedMember addr : members) {
@@ -449,13 +443,10 @@ public class NetView implements DataSerializableFixedID {
*/
public Set<InternalDistributedMember> getActualCrashedMembers(NetView oldView) {
Set<InternalDistributedMember> result = new HashSet<>(this.crashedMembers.size());
- for (InternalDistributedMember mbr : this.crashedMembers) {
- if ((mbr.getVmKind() != DistributionManager.ADMIN_ONLY_DM_TYPE)) {
- if (oldView == null || oldView.contains(mbr)) {
- result.add(mbr);
- }
- }
- }
+ result.addAll(this.crashedMembers.stream()
+ .filter(mbr -> (mbr.getVmKind() != DistributionManager.ADMIN_ONLY_DM_TYPE))
+ .filter(mbr -> oldView == null || oldView.contains(mbr))
+ .collect(Collectors.toList()));
return result;
}
@@ -554,14 +545,14 @@ public class NetView implements DataSerializableFixedID {
}
@Override
- public synchronized boolean equals(Object arg0) {
- if (arg0 == this) {
+ public synchronized boolean equals(Object other) {
+ if (other == this) {
return true;
}
- if (!(arg0 instanceof NetView)) {
+ if (!(other instanceof NetView)) {
return false;
}
- return this.members.equals(((NetView) arg0).getMembers());
+ return this.members.equals(((NetView) other).getMembers());
}
@Override
@@ -584,7 +575,8 @@ public class NetView implements DataSerializableFixedID {
creator = DataSerializer.readObject(in);
viewId = in.readInt();
members = DataSerializer.readArrayList(in);
- this.hashedMembers = new HashSet<InternalDistributedMember>(members);
+ assert members != null;
+ this.hashedMembers = new HashSet<>(members);
shutdownMembers = InternalDataSerializer.readHashSet(in);
crashedMembers = InternalDataSerializer.readHashSet(in);
failureDetectionPorts = DataSerializer.readIntArray(in);
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/5f840549/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/GMSUtil.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/GMSUtil.java b/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/GMSUtil.java
old mode 100755
new mode 100644
index 6478c70..d489431
--- a/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/GMSUtil.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/GMSUtil.java
@@ -26,11 +26,9 @@ import java.util.StringTokenizer;
import org.apache.logging.log4j.Logger;
import com.gemstone.gemfire.GemFireConfigException;
-import com.gemstone.gemfire.distributed.internal.membership.MemberAttributes;
import com.gemstone.gemfire.internal.SocketCreator;
public class GMSUtil {
- static Logger logger = Services.getLogger();
public static List<InetSocketAddress> parseLocators(String locatorsString, String bindAddress) {
InetAddress addr = null;
@@ -49,11 +47,11 @@ public class GMSUtil {
public static List<InetSocketAddress> parseLocators(String locatorsString, InetAddress bindAddress) {
- List<InetSocketAddress> result=new ArrayList<InetSocketAddress>(2);
+ List<InetSocketAddress> result= new ArrayList<>(2);
String host;
int port;
boolean checkLoopback = (bindAddress != null);
- boolean isLoopback = (checkLoopback? bindAddress.isLoopbackAddress() : false);
+ boolean isLoopback = (checkLoopback && bindAddress.isLoopbackAddress());
StringTokenizer parts=new StringTokenizer(locatorsString, ",");
while(parts.hasMoreTokens()) {
@@ -87,7 +85,7 @@ public class GMSUtil {
result.add(isa);
}
catch(NumberFormatException e) {
- // this shouldln't happen because the config has already been parsed and
+ // this shouldn't happen because the config has already been parsed and
// validated
}
}
@@ -100,7 +98,7 @@ public class GMSUtil {
* given value
*/
public static String replaceStrings(String properties, String property, String value) {
- StringBuffer sb = new StringBuffer();
+ StringBuilder sb = new StringBuilder();
int start = 0;
int index = properties.indexOf(property);
while (index != -1) {
@@ -136,24 +134,4 @@ 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/5f840549/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/ServiceConfig.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/ServiceConfig.java b/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/ServiceConfig.java
old mode 100755
new mode 100644
index 2a41574..6df0c7b
--- a/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/ServiceConfig.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/ServiceConfig.java
@@ -30,21 +30,21 @@ public class ServiceConfig {
public static final long MEMBER_REQUEST_COLLECTION_INTERVAL = Long.getLong(DistributionConfig.GEMFIRE_PREFIX + "member-request-collection-interval", 300);
/** various settings from Geode configuration */
- private long joinTimeout;
- private int[] membershipPortRange;
- private int udpRecvBufferSize;
- private int udpSendBufferSize;
- private long memberTimeout;
+ private final long joinTimeout;
+ private final int[] membershipPortRange;
+ private final int udpRecvBufferSize;
+ private final int udpSendBufferSize;
+ private final long memberTimeout;
private Integer lossThreshold;
- private Integer memberWeight;
+ private final Integer memberWeight;
private boolean networkPartitionDetectionEnabled;
- private int locatorWaitTime;
+ private final int locatorWaitTime;
/** the configuration for the distributed system */
- private DistributionConfig dconfig;
+ private final DistributionConfig dconfig;
/** the transport config from the distribution manager */
- private RemoteTransportConfig transport;
+ private final RemoteTransportConfig transport;
public int getLocatorWaitTime() {
@@ -62,16 +62,6 @@ public class ServiceConfig {
}
- public int getUdpRecvBufferSize() {
- return udpRecvBufferSize;
- }
-
-
- public int getUdpSendBufferSize() {
- return udpSendBufferSize;
- }
-
-
public long getMemberTimeout() {
return memberTimeout;
}
@@ -95,30 +85,6 @@ public class ServiceConfig {
this.networkPartitionDetectionEnabled = enabled;
}
- /**
- * returns the address that will be used by the DirectChannel to
- * identify this member
- */
- public InetAddress getInetAddress() {
- String bindAddress = this.dconfig.getBindAddress();
-
- try {
- /* note: had to change the following to make sure the prop wasn't empty
- in addition to not null for admin.DistributedSystemFactory */
- if (bindAddress != null && bindAddress.length() > 0) {
- return InetAddress.getByName(bindAddress);
-
- }
- else {
- return SocketCreator.getLocalHost();
- }
- }
- catch (java.net.UnknownHostException unhe) {
- throw new RuntimeException(unhe);
-
- }
- }
-
public boolean areLocatorsPreferredAsCoordinators() {
boolean locatorsAreCoordinators = false;
@@ -151,7 +117,7 @@ public class ServiceConfig {
this.transport = transport;
long defaultJoinTimeout = 24000;
- if (theConfig.getLocators().length() > 0 && !Locator.hasLocators()) {
+ if (theConfig.getLocators().length() > 0 && !Locator.hasLocator()) {
defaultJoinTimeout = 60000;
}
@@ -160,7 +126,7 @@ public class ServiceConfig {
long minimumJoinTimeout = dconfig.getMemberTimeout() * 2 + MEMBER_REQUEST_COLLECTION_INTERVAL;
if (defaultJoinTimeout < minimumJoinTimeout) {
defaultJoinTimeout = minimumJoinTimeout;
- };
+ }
joinTimeout = Long.getLong("p2p.joinTimeout", defaultJoinTimeout).longValue();
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/5f840549/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/Services.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/Services.java b/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/Services.java
index 7d1f2d9..5bc006f 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/Services.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/Services.java
@@ -41,6 +41,7 @@ import org.apache.logging.log4j.Logger;
import java.util.Timer;
+@SuppressWarnings("ConstantConditions")
public class Services {
private static final Logger logger = LogService.getLogger();
@@ -66,7 +67,7 @@ public class Services {
private InternalLogWriter logWriter;
private InternalLogWriter securityLogWriter;
- private Timer timer = new Timer("Geode Membership Timer", true);
+ private final Timer timer = new Timer("Geode Membership Timer", true);
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/5f840549/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/fd/GMSHealthMonitor.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/fd/GMSHealthMonitor.java b/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/fd/GMSHealthMonitor.java
index 4b63d4f..9fdbb64 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/fd/GMSHealthMonitor.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/fd/GMSHealthMonitor.java
@@ -19,7 +19,6 @@ package com.gemstone.gemfire.distributed.internal.membership.gms.fd;
import static com.gemstone.gemfire.internal.DataSerializableFixedID.HEARTBEAT_REQUEST;
import static com.gemstone.gemfire.internal.DataSerializableFixedID.HEARTBEAT_RESPONSE;
import static com.gemstone.gemfire.internal.DataSerializableFixedID.SUSPECT_MEMBERS_MESSAGE;
-import static com.sun.corba.se.impl.naming.cosnaming.NamingUtils.debug;
import java.io.DataInputStream;
import java.io.DataOutputStream;
@@ -41,6 +40,7 @@ import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.*;
import org.apache.logging.log4j.Logger;
import org.jgroups.util.UUID;
@@ -49,7 +49,6 @@ import com.gemstone.gemfire.CancelException;
import com.gemstone.gemfire.GemFireConfigException;
import com.gemstone.gemfire.SystemConnectException;
import com.gemstone.gemfire.distributed.DistributedMember;
-import com.gemstone.gemfire.distributed.DistributedSystemDisconnectedException;
import com.gemstone.gemfire.distributed.internal.DMStats;
import com.gemstone.gemfire.distributed.internal.DistributionMessage;
import com.gemstone.gemfire.distributed.internal.membership.InternalDistributedMember;
@@ -87,6 +86,7 @@ import com.gemstone.gemfire.internal.Version;
* for that member.
*
* */
+@SuppressWarnings({ "SynchronizationOnLocalVariableOrMethodParameter", "NullableProblems" })
public class GMSHealthMonitor implements HealthMonitor, MessageHandler {
private Services services;
@@ -98,7 +98,7 @@ public class GMSHealthMonitor implements HealthMonitor, MessageHandler {
private final AtomicInteger requestId = new AtomicInteger();
/** membership logger */
- private static Logger logger = Services.getLogger();
+ private static final Logger logger = Services.getLogger();
/**
* The number of recipients of periodic heartbeats. The recipients will
@@ -116,7 +116,7 @@ public class GMSHealthMonitor implements HealthMonitor, MessageHandler {
/** stall time to wait for members leaving concurrently */
public static final long MEMBER_SUSPECT_COLLECTION_INTERVAL = Long.getLong("geode.suspect-member-collection-interval", 200);
- volatile long currentTimeStamp;
+ private volatile long currentTimeStamp;
/** this member's ID */
private InternalDistributedMember localAddress;
@@ -156,10 +156,10 @@ public class GMSHealthMonitor implements HealthMonitor, MessageHandler {
private ScheduledFuture<?> monitorFuture;
/** test hook */
- volatile boolean playingDead = false;
+ private volatile boolean playingDead = false;
/** test hook */
- volatile boolean beingSick = false;
+ private volatile boolean beingSick = false;
// For TCP check
private ExecutorService serverSocketExecutor;
@@ -169,7 +169,7 @@ public class GMSHealthMonitor implements HealthMonitor, MessageHandler {
private volatile ServerSocket serverSocket;
/** Statistics about health monitor */
- protected DMStats stats;
+ private DMStats stats;
/**
* this class is to avoid garbage
@@ -261,7 +261,7 @@ public class GMSHealthMonitor implements HealthMonitor, MessageHandler {
class ClientSocketHandler implements Runnable {
- private Socket socket;
+ private final Socket socket;
public ClientSocketHandler(Socket socket) {
this.socket = socket;
@@ -272,7 +272,7 @@ public class GMSHealthMonitor implements HealthMonitor, MessageHandler {
socket.setTcpNoDelay(true);
DataInputStream in = new DataInputStream(socket.getInputStream());
OutputStream out = socket.getOutputStream();
- @SuppressWarnings("unused")
+ @SuppressWarnings("UnusedAssignment")
short version = in.readShort();
int vmViewId = in.readInt();
long uuidLSBs = in.readLong();
@@ -337,6 +337,7 @@ public class GMSHealthMonitor implements HealthMonitor, MessageHandler {
}
+ @SuppressWarnings("EmptyMethod")
public static void loadEmergencyClasses() {
}
@@ -380,27 +381,23 @@ public class GMSHealthMonitor implements HealthMonitor, MessageHandler {
setNextNeighbor(cv, mbr);
// we need to check this member
- checkExecutor.execute(new Runnable() {
-
- @Override
- public void run() {
- boolean pinged = false;
- try {
- pinged = GMSHealthMonitor.this.doCheckMember(mbr, true);
- } catch (CancelException e) {
- return;
- }
-
- if (!pinged) {
- suspectedMemberInView.put(mbr, currentView);
- String reason = "Member isn't responding to heartbeat requests";
- GMSHealthMonitor.this.initiateSuspicion(mbr, reason);
- } else {
- logger.trace("Setting next neighbor as member {} has responded.", mbr);
- suspectedMemberInView.remove(mbr);
- // back to previous one
- setNextNeighbor(GMSHealthMonitor.this.currentView, null);
- }
+ checkExecutor.execute(() -> {
+ boolean pinged = false;
+ try {
+ pinged = GMSHealthMonitor.this.doCheckMember(mbr, true);
+ } catch (CancelException e) {
+ return;
+ }
+
+ if (!pinged) {
+ suspectedMemberInView.put(mbr, currentView);
+ String reason = "Member isn't responding to heartbeat requests";
+ GMSHealthMonitor.this.initiateSuspicion(mbr, reason);
+ } else {
+ logger.trace("Setting next neighbor as member {} has responded.", mbr);
+ suspectedMemberInView.remove(mbr);
+ // back to previous one
+ setNextNeighbor(GMSHealthMonitor.this.currentView, null);
}
});
@@ -411,7 +408,7 @@ public class GMSHealthMonitor implements HealthMonitor, MessageHandler {
return;
}
SuspectRequest sr = new SuspectRequest(mbr, reason);
- List<SuspectRequest> sl = new ArrayList<SuspectRequest>();
+ List<SuspectRequest> sl = new ArrayList<>();
sl.add(sr);
sendSuspectRequest(sl);
}
@@ -586,17 +583,14 @@ public class GMSHealthMonitor implements HealthMonitor, MessageHandler {
}
public void start() {
- scheduler = Executors.newScheduledThreadPool(1, new ThreadFactory() {
- @Override
- public Thread newThread(Runnable r) {
- Thread th = new Thread(Services.getThreadGroup(), r, "Geode Failure Detection Scheduler");
- th.setDaemon(true);
- return th;
- }
+ scheduler = Executors.newScheduledThreadPool(1, r -> {
+ Thread th = new Thread(Services.getThreadGroup(), r, "Geode Failure Detection Scheduler");
+ th.setDaemon(true);
+ return th;
});
checkExecutor = Executors.newCachedThreadPool(new ThreadFactory() {
- AtomicInteger threadIdx = new AtomicInteger();
+ final AtomicInteger threadIdx = new AtomicInteger();
@Override
public Thread newThread(Runnable r) {
@@ -619,10 +613,10 @@ public class GMSHealthMonitor implements HealthMonitor, MessageHandler {
// }
// }, MEMBER_SUSPECT_COLLECTION_INTERVAL);
// suspectRequestCollectorThread.setDaemon(true);
-// suspectRequestCollectorThread.start();
+// suspectRequestCollectorThread.start()
serverSocketExecutor = Executors.newCachedThreadPool(new ThreadFactory() {
- AtomicInteger threadIdx = new AtomicInteger();
+ final AtomicInteger threadIdx = new AtomicInteger();
@Override
public Thread newThread(Runnable r) {
@@ -636,14 +630,12 @@ public class GMSHealthMonitor implements HealthMonitor, MessageHandler {
}
ServerSocket createServerSocket(InetAddress socketAddress, int[] portRange) {
- ServerSocket serverSocket = null;
+ ServerSocket serverSocket;
try {
serverSocket = SocketCreator.getDefaultInstance().createServerSocketUsingPortRange(socketAddress, 50/*backlog*/,
true/*isBindAddress*/, false/*useNIO*/, 65536/*tcpBufferSize*/, portRange, false);
socketPort = serverSocket.getLocalPort();
- } catch (IOException e) {
- throw new GemFireConfigException("Unable to allocate a failure detection port in the membership-port range", e);
- } catch (SystemConnectException e) {
+ } catch (IOException | SystemConnectException e) {
throw new GemFireConfigException("Unable to allocate a failure detection port in the membership-port range", e);
}
return serverSocket;
@@ -657,48 +649,45 @@ public class GMSHealthMonitor implements HealthMonitor, MessageHandler {
// allocate a socket here so there are no race conditions between knowing the FD
// socket port and joining the system
- serverSocketExecutor.execute(new Runnable() {
- @Override
- public void run() {
- logger.info("Started failure detection server thread on {}:{}.", ssocket.getInetAddress(), socketPort);
- Socket socket = null;
- try {
- while (!services.getCancelCriterion().isCancelInProgress()
- && !GMSHealthMonitor.this.isStopping) {
+ serverSocketExecutor.execute(() -> {
+ logger.info("Started failure detection server thread on {}:{}.", ssocket.getInetAddress(), socketPort);
+ Socket socket = null;
+ try {
+ while (!services.getCancelCriterion().isCancelInProgress()
+ && !GMSHealthMonitor.this.isStopping) {
+ try {
+ socket = ssocket.accept();
+ if (GMSHealthMonitor.this.playingDead) {
+ continue;
+ }
+ serverSocketExecutor.execute(new ClientSocketHandler(socket)); //start(); [bruce] I'm seeing a lot of failures due to this thread not being created fast enough, sometimes as long as 30 seconds
+
+ } catch (RejectedExecutionException e) {
+ // this can happen during shutdown
+
+ } catch (IOException e) {
+ if (!isStopping) {
+ logger.trace("Unexpected exception", e);
+ }
try {
- socket = ssocket.accept();
- if (GMSHealthMonitor.this.playingDead) {
- continue;
- }
- serverSocketExecutor.execute(new ClientSocketHandler(socket)); //start(); [bruce] I'm seeing a lot of failures due to this thread not being created fast enough, sometimes as long as 30 seconds
-
- } catch (RejectedExecutionException e) {
- // this can happen during shutdown
-
- } catch (IOException e) {
- if (!isStopping) {
- logger.trace("Unexpected exception", e);
- }
- try {
- if (socket != null) {
- socket.close();
- }
- } catch (IOException ioe) {
- logger.trace("Unexpected exception", ioe);
+ if (socket != null) {
+ socket.close();
}
+ } catch (IOException ioe) {
+ logger.trace("Unexpected exception", ioe);
}
}
- logger.info("GMSHealthMonitor server thread exiting");
- } finally {
- // close the server socket
- if (ssocket != null && !ssocket.isClosed()) {
- try {
- ssocket.close();
- serverSocket = null;
- logger.info("GMSHealthMonitor server socket closed.");
- } catch (IOException e) {
- logger.debug("Unexpected exception", e);
- }
+ }
+ logger.info("GMSHealthMonitor server thread exiting");
+ } finally {
+ // close the server socket
+ if (!ssocket.isClosed()) {
+ try {
+ ssocket.close();
+ serverSocket = null;
+ logger.info("GMSHealthMonitor server socket closed.");
+ } catch (IOException e) {
+ logger.debug("Unexpected exception", e);
}
}
}
@@ -829,7 +818,7 @@ public class GMSHealthMonitor implements HealthMonitor, MessageHandler {
List<InternalDistributedMember> allMembers = newView.getMembers();
- Set<?> checkAllSuspected = new HashSet<>(allMembers);
+ Set<InternalDistributedMember> checkAllSuspected = new HashSet<>(allMembers);
checkAllSuspected.removeAll(suspectedMemberInView.keySet());
checkAllSuspected.remove(localAddress);
if (checkAllSuspected.isEmpty() && allMembers.size() > 1) {
@@ -899,8 +888,7 @@ public class GMSHealthMonitor implements HealthMonitor, MessageHandler {
}
Collection<Response> val = requestIdVsResponse.values();
- for (Iterator<Response> it = val.iterator(); it.hasNext();) {
- Response r = it.next();
+ for (Response r : val) {
synchronized (r) {
r.notify();
}
@@ -1066,8 +1054,6 @@ public class GMSHealthMonitor implements HealthMonitor, MessageHandler {
* to become the new membership coordinator.
* it will to final check on that member and then it will send remove request
* for that member
- *
- * @param incomingRequest
*/
private void processSuspectMembersRequest(SuspectMembersMessage incomingRequest) {
@@ -1118,13 +1104,11 @@ public class GMSHealthMonitor implements HealthMonitor, MessageHandler {
else {
NetView check = new NetView(cv, cv.getViewId() + 1);
- ArrayList<SuspectRequest> smbr = new ArrayList<SuspectRequest>();
+ ArrayList<SuspectRequest> smbr = new ArrayList<>();
synchronized (viewVsSuspectedMembers) {
recordSuspectRequests(sMembers, cv);
Set<SuspectRequest> viewVsMembers = viewVsSuspectedMembers.get(cv);
- Iterator<SuspectRequest> itr = viewVsMembers.iterator();
- while (itr.hasNext()) {
- SuspectRequest sr = itr.next();
+ for (final SuspectRequest sr : viewVsMembers) {
check.remove(sr.getSuspectMember());
smbr.add(sr);
}
@@ -1148,16 +1132,14 @@ public class GMSHealthMonitor implements HealthMonitor, MessageHandler {
/***
* This method make sure that records suspectRequest. We need to make sure this
* on preferred coordinators, as elder coordinator might be in suspected list next.
- * @param sMembers
- * @param cv
*/
private void recordSuspectRequests(List<SuspectRequest> sMembers, NetView cv) {
// record suspect requests
- Set<SuspectRequest> viewVsMembers = null;
+ Set<SuspectRequest> viewVsMembers;
synchronized (viewVsSuspectedMembers) {
viewVsMembers = viewVsSuspectedMembers.get(cv);
if (viewVsMembers == null) {
- viewVsMembers = new HashSet<SuspectRequest>();
+ viewVsMembers = new HashSet<>();
viewVsSuspectedMembers.put(cv, viewVsMembers);
}
for (SuspectRequest sr: sMembers) {
@@ -1171,16 +1153,11 @@ public class GMSHealthMonitor implements HealthMonitor, MessageHandler {
* socket information is available for the member (in the view) then
* we attempt to connect to its socket and ask if it's the expected member.
* Otherwise we send a heartbeat request and wait for a reply.
- *
- * @param initiator
- * @param sMembers
- * @param cv
*/
private void checkIfAvailable(final InternalDistributedMember initiator,
List<SuspectRequest> sMembers, final NetView cv) {
- for (int i = 0; i < sMembers.size(); i++) {
- final SuspectRequest sr = sMembers.get(i);
+ for (final SuspectRequest sr : sMembers) {
final InternalDistributedMember mbr = sr.getSuspectMember();
if (!cv.contains(mbr) || membersInFinalCheck.contains(mbr)) {
@@ -1203,25 +1180,16 @@ public class GMSHealthMonitor implements HealthMonitor, MessageHandler {
final String reason = sr.getReason();
logger.debug("Scheduling final check for member {}; reason={}", mbr, reason);
// its a coordinator
- checkExecutor.execute(new Runnable() {
-
- @Override
- public void run() {
- try {
- inlineCheckIfAvailable(initiator, cv, true, mbr,
- reason);
- } catch (DistributedSystemDisconnectedException e) {
- return;
- } catch (CancelException e) {
- // shutting down
- } catch (Exception e) {
- logger.info("Unexpected exception while verifying member", e);
- } finally {
- GMSHealthMonitor.this.suspectedMemberInView.remove(mbr);
- }
+ checkExecutor.execute(() -> {
+ try {
+ inlineCheckIfAvailable(initiator, cv, true, mbr, reason);
+ } catch (CancelException e) {
+ // shutting down
+ } catch (Exception e) {
+ logger.info("Unexpected exception while verifying member", e);
+ } finally {
+ GMSHealthMonitor.this.suspectedMemberInView.remove(mbr);
}
-
-
});
// }// scheduling for final check and removing it..
}
@@ -1317,13 +1285,11 @@ public class GMSHealthMonitor implements HealthMonitor, MessageHandler {
logger.debug("Sending suspect request for members {}", requests);
List<InternalDistributedMember> recipients;
if (currentView.size() > 4) {
- HashSet<InternalDistributedMember> filter = new HashSet<InternalDistributedMember>();
+ HashSet<InternalDistributedMember> filter = new HashSet<>();
for (Enumeration<InternalDistributedMember> e = suspectedMemberInView.keys(); e.hasMoreElements();) {
filter.add(e.nextElement());
}
- for (int i = 0; i < requests.size(); i++) {
- filter.add(requests.get(i).getSuspectMember());
- }
+ filter.addAll(requests.stream().map(SuspectRequest::getSuspectMember).collect(Collectors.toList()));
recipients = currentView.getPreferredCoordinators(filter, services.getJoinLeave().getMemberID(), 5);
} else {
recipients = currentView.getMembers();
@@ -1344,9 +1310,9 @@ public class GMSHealthMonitor implements HealthMonitor, MessageHandler {
}
private static class ConnectTimeoutTask extends TimerTask implements ConnectionWatcher {
- Timer scheduler;
+ final Timer scheduler;
Socket socket;
- long timeout;
+ final long timeout;
ConnectTimeoutTask(Timer scheduler, long timeout) {
this.scheduler = scheduler;
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/5f840549/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/locator/GMSLocator.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/locator/GMSLocator.java b/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/locator/GMSLocator.java
old mode 100755
new mode 100644
index aab9002..1065214
--- a/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/locator/GMSLocator.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/locator/GMSLocator.java
@@ -72,7 +72,7 @@ public class GMSLocator implements Locator, NetLocator {
private final LocatorStats stats;
private InternalDistributedMember localAddress;
- private Set<InternalDistributedMember> registrants = new HashSet<InternalDistributedMember>();
+ private final Set<InternalDistributedMember> registrants = new HashSet<>();
/**
* The current membership view, or one recovered from disk.
@@ -99,7 +99,7 @@ public class GMSLocator implements Locator, NetLocator {
this.networkPartitionDetectionEnabled = networkPartitionDetectionEnabled;
this.locatorString = locatorString;
if (this.locatorString == null || this.locatorString.length() == 0) {
- this.locators = new ArrayList<InetSocketAddress>(0);
+ this.locators = new ArrayList<>(0);
} else {
this.locators = GMSUtil.parseLocators(locatorString, bindAddress);
}
@@ -178,7 +178,6 @@ public class GMSLocator implements Locator, NetLocator {
}
boolean fromView = false;
- int viewId = -1;
NetView v = this.view;
if (v != null) {
@@ -193,10 +192,10 @@ public class GMSLocator implements Locator, NetLocator {
break;
}
}
- viewId = v.getViewId();
+ int viewId = v.getViewId();
if (viewId > findRequest.getLastViewId()) {
// ignore the requests rejectedCoordinators if the view has changed
- coord = v.getCoordinator(Collections.<InternalDistributedMember>emptyList());
+ coord = v.getCoordinator(Collections.emptyList());
} else {
coord = v.getCoordinator(findRequest.getRejectedCoordinators());
}
@@ -229,7 +228,7 @@ public class GMSLocator implements Locator, NetLocator {
synchronized(registrants) {
response = new FindCoordinatorResponse(coord, localAddress,
- fromView, view, new HashSet<InternalDistributedMember>(registrants),
+ fromView, view, new HashSet<>(registrants),
this.networkPartitionDetectionEnabled, this.usePreferredCoordinators);
}
}
@@ -240,7 +239,7 @@ public class GMSLocator implements Locator, NetLocator {
return response;
}
- public void saveView(NetView view) {
+ private void saveView(NetView view) {
if (viewFile == null) {
return;
}
@@ -287,10 +286,10 @@ public class GMSLocator implements Locator, NetLocator {
// test hook
public List<InternalDistributedMember> getMembers() {
if (view != null) {
- return new ArrayList<InternalDistributedMember>(view.getMembers());
+ return new ArrayList<>(view.getMembers());
} else {
synchronized(registrants) {
- return new ArrayList<InternalDistributedMember>(registrants);
+ return new ArrayList<>(registrants);
}
}
}
@@ -301,7 +300,7 @@ public class GMSLocator implements Locator, NetLocator {
setMembershipManager(((InternalDistributedSystem)ds).getDM().getMembershipManager());
}
- public void recover() throws InternalGemFireException {
+ private void recover() throws InternalGemFireException {
if (!recoverFromOthers()) {
recoverFromFile(viewFile);
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/5f840549/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/membership/GMSJoinLeave.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/membership/GMSJoinLeave.java b/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/membership/GMSJoinLeave.java
index 4b82fa0..58b794a 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/membership/GMSJoinLeave.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/membership/GMSJoinLeave.java
@@ -162,7 +162,7 @@ public class GMSJoinLeave implements JoinLeave, MessageHandler {
/**
* collects the response to a join request
*/
- private JoinResponseMessage[] joinResponse = new JoinResponseMessage[1];
+ private final JoinResponseMessage[] joinResponse = new JoinResponseMessage[1];
/**
* collects responses to new views
@@ -227,7 +227,7 @@ public class GMSJoinLeave implements JoinLeave, MessageHandler {
int locatorsContacted = 0;
boolean hasContactedAJoinedLocator;
NetView view;
- Set<FindCoordinatorResponse> responses = new HashSet<>();
+ final Set<FindCoordinatorResponse> responses = new HashSet<>();
void cleanup() {
alreadyTried.clear();
@@ -366,7 +366,7 @@ public class GMSJoinLeave implements JoinLeave, MessageHandler {
services.getMessenger().send(req);
}
- JoinResponseMessage response = null;
+ JoinResponseMessage response;
try {
response = waitForJoinResponse();
} catch (InterruptedException e) {
@@ -393,7 +393,7 @@ public class GMSJoinLeave implements JoinLeave, MessageHandler {
}
//there is no way we can rech here right now
- throw new RuntimeException("Join Request Failed with response " + joinResponse );
+ throw new RuntimeException("Join Request Failed with response " + joinResponse[0] );
}
private JoinResponseMessage waitForJoinResponse() throws InterruptedException {
@@ -474,7 +474,7 @@ public class GMSJoinLeave implements JoinLeave, MessageHandler {
return;
}
Object creds = incomingRequest.getCredentials();
- String rejection = null;
+ String rejection;
try {
rejection = services.getAuthenticator().authenticate(incomingRequest.getMemberID(), (Properties)creds);
} catch (Exception e) {
@@ -699,7 +699,7 @@ public class GMSJoinLeave implements JoinLeave, MessageHandler {
startViewBroadcaster();
} else {
// create and send out a new view
- NetView newView = addMemberToNetView(this.currentView, oldCoordinator);
+ NetView newView = addMemberToNetView(oldCoordinator);
createAndStartViewCreator(newView);
startViewBroadcaster();
}
@@ -717,9 +717,9 @@ public class GMSJoinLeave implements JoinLeave, MessageHandler {
}
}
- private NetView addMemberToNetView(NetView netView, InternalDistributedMember oldCoordinator) {
+ private NetView addMemberToNetView(InternalDistributedMember oldCoordinator) {
boolean testing = unitTesting.contains("noRandomViewChange");
- NetView newView = null;
+ NetView newView;
Set<InternalDistributedMember> leaving = new HashSet<>();
Set<InternalDistributedMember> removals;
synchronized (viewInstallationLock) {
@@ -758,7 +758,7 @@ public class GMSJoinLeave implements JoinLeave, MessageHandler {
}
}
- private void sendRemoveMessages(List<InternalDistributedMember> removals, List<String> reasons, NetView newView, Set<InternalDistributedMember> oldIds) {
+ private void sendRemoveMessages(List<InternalDistributedMember> removals, List<String> reasons, Set<InternalDistributedMember> oldIds) {
Iterator<String> reason = reasons.iterator();
for (InternalDistributedMember mbr : removals) {
//if olds not contains mbr then send remove request
@@ -772,15 +772,14 @@ public class GMSJoinLeave implements JoinLeave, MessageHandler {
}
boolean prepareView(NetView view, List<InternalDistributedMember> newMembers) throws InterruptedException {
- return sendView(view, newMembers, true, this.prepareProcessor);
+ return sendView(view, true, this.prepareProcessor);
}
void sendView(NetView view, List<InternalDistributedMember> newMembers) throws InterruptedException {
- sendView(view, newMembers, false, this.viewProcessor);
+ sendView(view, false, this.viewProcessor);
}
- private boolean sendView(NetView view, List<InternalDistributedMember> newMembers, boolean preparing,
- ViewReplyProcessor viewReplyProcessor) throws InterruptedException {
+ private boolean sendView(NetView view, boolean preparing, ViewReplyProcessor viewReplyProcessor) throws InterruptedException {
int id = view.getViewId();
InstallViewMessage msg = new InstallViewMessage(view, services.getAuthenticator().getCredentials(this.localAddress), preparing);
@@ -1428,8 +1427,8 @@ public class GMSJoinLeave implements JoinLeave, MessageHandler {
@Override
public void memberSuspected(InternalDistributedMember initiator, InternalDistributedMember suspect, String reason) {
- prepareProcessor.memberSuspected(initiator, suspect);
- viewProcessor.memberSuspected(initiator, suspect);
+ prepareProcessor.memberSuspected(suspect);
+ viewProcessor.memberSuspected(suspect);
}
@Override
@@ -1440,7 +1439,7 @@ public class GMSJoinLeave implements JoinLeave, MessageHandler {
stopCoordinatorServices();
if (view != null) {
if (view.size() > 1) {
- List<InternalDistributedMember> coords = view.getPreferredCoordinators(Collections.<InternalDistributedMember>emptySet(), localAddress, 5);
+ List<InternalDistributedMember> coords = view.getPreferredCoordinators(Collections.emptySet(), localAddress, 5);
logger.debug("Sending my leave request to {}", coords);
LeaveRequestMessage m = new LeaveRequestMessage(coords, this.localAddress, "this member is shutting down");
services.getMessenger().send(m);
@@ -1463,7 +1462,7 @@ public class GMSJoinLeave implements JoinLeave, MessageHandler {
processRemoveRequest(msg);
if (!this.isCoordinator) {
msg.resetRecipients();
- msg.setRecipients(v.getPreferredCoordinators(Collections.<InternalDistributedMember> emptySet(), localAddress, 10));
+ msg.setRecipients(v.getPreferredCoordinators(Collections.emptySet(), localAddress, 10));
services.getMessenger().send(msg);
}
} else {
@@ -1642,7 +1641,7 @@ public class GMSJoinLeave implements JoinLeave, MessageHandler {
}
}
- synchronized void memberSuspected(InternalDistributedMember initiator, InternalDistributedMember suspect) {
+ synchronized void memberSuspected(InternalDistributedMember suspect) {
if (waiting) {
// we will do a final check on this member if it hasn't already
// been done, so stop waiting for it now
@@ -1707,7 +1706,7 @@ public class GMSJoinLeave implements JoinLeave, MessageHandler {
}
Set<InternalDistributedMember> waitForResponses() throws InterruptedException {
- Set<InternalDistributedMember> result = null;
+ Set<InternalDistributedMember> result;
long endOfWait = System.currentTimeMillis() + viewAckTimeout;
try {
while (System.currentTimeMillis() < endOfWait && (!services.getCancelCriterion().isCancelInProgress())) {
@@ -1797,7 +1796,7 @@ public class GMSJoinLeave implements JoinLeave, MessageHandler {
/**
* initial joining members. guarded by synch on ViewCreator
*/
- List<InternalDistributedMember> initialJoins = Collections.<InternalDistributedMember>emptyList();
+ List<InternalDistributedMember> initialJoins = Collections.emptyList();
/**
* initial leaving members guarded by synch on ViewCreator
*/
@@ -2041,7 +2040,7 @@ public class GMSJoinLeave implements JoinLeave, MessageHandler {
for (DistributionMessage msg : requests) {
logger.debug("processing request {}", msg);
- InternalDistributedMember mbr = null;
+ InternalDistributedMember mbr;
switch (msg.getDSFID()) {
case JOIN_REQUEST:
JoinRequestMessage jmsg = (JoinRequestMessage) msg;
@@ -2085,10 +2084,8 @@ public class GMSJoinLeave implements JoinLeave, MessageHandler {
removalReqs.add(mbr);
removalReasons.add(((RemoveMemberMessage) msg).getReason());
} else {
- sendRemoveMessages(Collections.<InternalDistributedMember>singletonList(mbr),
- Collections.<String>singletonList(((RemoveMemberMessage) msg).getReason()),
- currentView,
- new HashSet<InternalDistributedMember>());
+ sendRemoveMessages(Collections.singletonList(mbr),
+ Collections.singletonList(((RemoveMemberMessage) msg).getReason()), new HashSet<InternalDistributedMember>());
}
}
break;
@@ -2150,7 +2147,7 @@ public class GMSJoinLeave implements JoinLeave, MessageHandler {
}
// send removal messages before installing the view so we stop
// getting messages from members that have been kicked out
- sendRemoveMessages(removalReqs, removalReasons, newView, oldIDs);
+ sendRemoveMessages(removalReqs, removalReasons, oldIDs);
prepareAndSendView(newView, joinReqs, leaveReqs, newView.getCrashedMembers());
@@ -2163,7 +2160,7 @@ public class GMSJoinLeave implements JoinLeave, MessageHandler {
*/
void prepareAndSendView(NetView newView, List<InternalDistributedMember> joinReqs, Set<InternalDistributedMember> leaveReqs,
Set<InternalDistributedMember> removalReqs) throws InterruptedException {
- boolean prepared = false;
+ boolean prepared;
do {
if (this.shutdown || Thread.currentThread().isInterrupted()) {
return;
@@ -2255,7 +2252,7 @@ public class GMSJoinLeave implements JoinLeave, MessageHandler {
for (int i = 0; i < size; i++) {
reasons.add("Failed to acknowledge a new membership view and then failed tcp/ip connection attempt");
}
- sendRemoveMessages(failures, reasons, newView, new HashSet<InternalDistributedMember>());
+ sendRemoveMessages(failures, reasons, new HashSet<InternalDistributedMember>());
}
// if there is no conflicting view then we can count
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/5f840549/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/GMSQuorumChecker.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/GMSQuorumChecker.java b/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/GMSQuorumChecker.java
index 1f8ed2e..63f4aa1 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/GMSQuorumChecker.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/GMSQuorumChecker.java
@@ -48,13 +48,13 @@ public class GMSQuorumChecker implements QuorumChecker {
private Set<InternalDistributedMember> receivedAcks;
- private NetView lastView;
+ private final NetView lastView;
// guarded by this
private boolean quorumAchieved = false;
- private JChannel channel;
+ private final JChannel channel;
private JGAddress myAddress;
- private int partitionThreshold;
+ private final int partitionThreshold;
public GMSQuorumChecker(NetView jgView, int partitionThreshold, JChannel channel) {
this.lastView = jgView;
@@ -63,7 +63,7 @@ public class GMSQuorumChecker implements QuorumChecker {
}
public void initialize() {
- receivedAcks = new ConcurrentHashSet<InternalDistributedMember>();
+ receivedAcks = new ConcurrentHashSet<>();
pingPonger = new GMSPingPonger();
// UUID logicalAddress = (UUID) channel.getAddress();
@@ -72,7 +72,7 @@ public class GMSQuorumChecker implements QuorumChecker {
// myAddress = new JGAddress(logicalAddress, ipaddr);
myAddress = (JGAddress)channel.down(new Event(Event.GET_LOCAL_ADDRESS));
- addressConversionMap = new ConcurrentHashMap<SocketAddress, InternalDistributedMember>(this.lastView.size());
+ addressConversionMap = new ConcurrentHashMap<>(this.lastView.size());
List<InternalDistributedMember> members = this.lastView.getMembers();
for (InternalDistributedMember addr : members) {
SocketAddress sockaddr = new InetSocketAddress(addr.getNetMember().getInetAddress(), addr.getPort());
@@ -92,15 +92,11 @@ public class GMSQuorumChecker implements QuorumChecker {
if (isDebugEnabled) {
logger.debug("beginning quorum check with {}", this);
}
- try {
- sendPingMessages();
- quorumAchieved = waitForResponses(lastView.getMembers().size(), timeout);
- // If we did not achieve full quorum, calculate if we achieved quorum
- if (!quorumAchieved) {
- quorumAchieved = calculateQuorum();
- }
- } finally {
-
+ sendPingMessages();
+ quorumAchieved = waitForResponses(lastView.getMembers().size(), timeout);
+ // If we did not achieve full quorum, calculate if we achieved quorum
+ if (!quorumAchieved) {
+ quorumAchieved = calculateQuorum();
}
return quorumAchieved;
}
@@ -209,18 +205,15 @@ public class GMSQuorumChecker implements QuorumChecker {
@Override
public void receive(Message msg) {
- Object contents = msg.getBuffer();
- if (contents instanceof byte[]) {
- byte[] msgBytes = (byte[]) contents;
- if (pingPonger.isPingMessage(msgBytes)) {
- try {
- pingPonger.sendPongMessage(channel, myAddress, msg.getSrc());
- } catch (Exception e) {
- logger.debug("Failed sending Pong message to " + msg.getSrc());
- }
- } else if (pingPonger.isPongMessage(msgBytes)) {
- pongReceived(msg.getSrc());
+ byte[] msgBytes = msg.getBuffer();
+ if (pingPonger.isPingMessage(msgBytes)) {
+ try {
+ pingPonger.sendPongMessage(channel, myAddress, msg.getSrc());
+ } catch (Exception e) {
+ logger.debug("Failed sending Pong message to " + msg.getSrc());
}
+ } else if (pingPonger.isPongMessage(msgBytes)) {
+ pongReceived(msg.getSrc());
}
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/5f840549/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/JGroupsMessenger.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/JGroupsMessenger.java b/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/JGroupsMessenger.java
index 879b0a5..a119bb5 100644
--- a/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/JGroupsMessenger.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/JGroupsMessenger.java
@@ -59,11 +59,13 @@ import java.net.UnknownHostException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.*;
import static com.gemstone.gemfire.distributed.internal.membership.gms.GMSUtil.replaceStrings;
import static com.gemstone.gemfire.internal.DataSerializableFixedID.JOIN_REQUEST;
import static com.gemstone.gemfire.internal.DataSerializableFixedID.JOIN_RESPONSE;
+@SuppressWarnings("StatementWithEmptyBody")
public class JGroupsMessenger implements Messenger {
private static final Logger logger = Services.getLogger();
@@ -79,26 +81,26 @@ public class JGroupsMessenger implements Messenger {
private static final String JGROUPS_MCAST_CONFIG_FILE_NAME = "com/gemstone/gemfire/distributed/internal/membership/gms/messenger/jgroups-mcast.xml";
/** JG magic numbers for types added to the JG ClassConfigurator */
- public static final short JGROUPS_TYPE_JGADDRESS = 2000;
- public static final short JGROUPS_PROTOCOL_TRANSPORT = 1000;
+ private static final short JGROUPS_TYPE_JGADDRESS = 2000;
+ private static final short JGROUPS_PROTOCOL_TRANSPORT = 1000;
public static boolean THROW_EXCEPTION_ON_START_HOOK;
- String jgStackConfig;
+ private String jgStackConfig;
JChannel myChannel;
InternalDistributedMember localAddress;
JGAddress jgAddress;
- Services services;
+ private Services services;
/** handlers that receive certain classes of messages instead of the Manager */
- Map<Class, MessageHandler> handlers = new ConcurrentHashMap<Class, MessageHandler>();
+ private final Map<Class, MessageHandler> handlers = new ConcurrentHashMap<>();
private volatile NetView view;
- private GMSPingPonger pingPonger = new GMSPingPonger();
+ private final GMSPingPonger pingPonger = new GMSPingPonger();
- protected AtomicLong pongsReceived = new AtomicLong(0);
+ protected final AtomicLong pongsReceived = new AtomicLong(0);
/**
* A set that contains addresses that we have logged JGroups IOExceptions for in the
@@ -106,7 +108,7 @@ public class JGroupsMessenger implements Messenger {
* reduces the amount of suspect processing initiated by IOExceptions and the
* amount of exceptions logged
*/
- private Set<Address> addressesWithIoExceptionsProcessed = Collections.synchronizedSet(new HashSet<Address>());
+ private final Set<Address> addressesWithIoExceptionsProcessed = Collections.synchronizedSet(new HashSet<Address>());
static {
// register classes that we've added to jgroups that are put on the wire
@@ -132,9 +134,9 @@ public class JGroupsMessenger implements Messenger {
}
System.setProperty("jgroups.resolve_dns", String.valueOf(!b));
- InputStream is= null;
+ InputStream is;
- String r = null;
+ String r;
if (transport.isMcastEnabled()) {
r = JGROUPS_MCAST_CONFIG_FILE_NAME;
} else {
@@ -149,7 +151,7 @@ public class JGroupsMessenger implements Messenger {
try {
//PlainConfigurator config = PlainConfigurator.getInstance(is);
//properties = config.getProtocolStackString();
- StringBuffer sb = new StringBuffer(3000);
+ StringBuilder sb = new StringBuilder(3000);
BufferedReader br;
br = new BufferedReader(new InputStreamReader(is, "US-ASCII"));
String input;
@@ -238,7 +240,7 @@ public class JGroupsMessenger implements Messenger {
myChannel = (JChannel)oldChannel;
// scrub the old channel
ViewId vid = new ViewId(new JGAddress(), 0);
- View jgv = new View(vid, new ArrayList<Address>());
+ View jgv = new View(vid, new ArrayList<>());
this.myChannel.down(new Event(Event.VIEW_CHANGE, jgv));
UUID logicalAddress = (UUID)myChannel.getAddress();
if (logicalAddress instanceof JGAddress) {
@@ -334,12 +336,10 @@ public class JGroupsMessenger implements Messenger {
if (this.jgAddress.getVmViewId() < 0) {
this.jgAddress.setVmViewId(this.localAddress.getVmViewId());
}
- List<JGAddress> mbrs = new ArrayList<JGAddress>(v.size());
- for (InternalDistributedMember idm: v.getMembers()) {
- mbrs.add(new JGAddress(idm));
- }
+ List<JGAddress> mbrs = new ArrayList<>(v.size());
+ mbrs.addAll(v.getMembers().stream().map(JGAddress::new).collect(Collectors.toList()));
ViewId vid = new ViewId(new JGAddress(v.getCoordinator()), v.getViewId());
- View jgv = new View(vid, new ArrayList<Address>(mbrs));
+ View jgv = new View(vid, new ArrayList<>(mbrs));
logger.trace("installing JGroups view: {}", jgv);
this.myChannel.down(new Event(Event.VIEW_CHANGE, jgv));
@@ -353,6 +353,7 @@ public class JGroupsMessenger implements Messenger {
* recipient.<p>
* see Transport._send()
*/
+ @SuppressWarnings("UnusedParameters")
public void handleJGroupsIOException(IOException e, Address dest) {
if (services.getManager().shutdownInProgress()) { // GEODE-634 - don't log IOExceptions during shutdown
return;
@@ -473,6 +474,7 @@ public class JGroupsMessenger implements Messenger {
long pongsSnapshot = pongsReceived.longValue();
JGAddress dest = null;
try {
+ //noinspection ConstantConditions
pingPonger.sendPingMessage(myChannel, jgAddress, dest);
} catch (Exception e) {
logger.warn("unable to send multicast message: {}", (jgAddress==null? "multicast recipients":jgAddress),
@@ -538,9 +540,7 @@ public class JGroupsMessenger implements Messenger {
long now = System.currentTimeMillis();
if (!warned && now >= warnTime) {
warned = true;
- if (senderSeqnos != null) {
- received = String.valueOf(senderSeqnos[0]);
- }
+ received = String.valueOf(senderSeqnos[0]);
logger.warn("{} seconds have elapsed while waiting for multicast messages from {}. Received {} but expecting at least {}.",
Long.toString((warnTime-startTime)/1000L), sender, received, seqno);
}
@@ -561,7 +561,7 @@ public class JGroupsMessenger implements Messenger {
return send(msg, true);
}
- public Set<InternalDistributedMember> send(DistributionMessage msg, boolean reliably) {
+ private Set<InternalDistributedMember> send(DistributionMessage msg, boolean reliably) {
// perform the same jgroups messaging as in 8.2's GMSMembershipManager.send() method
@@ -608,7 +608,7 @@ public class JGroupsMessenger implements Messenger {
Message jmsg = createJGMessage(msg, local, Version.CURRENT_ORDINAL);
theStats.endMsgSerialization(startSer);
- Exception problem = null;
+ Exception problem;
try {
jmsg.setTransientFlag(TransientFlag.DONT_LOOPBACK);
if (!reliably) {
@@ -655,7 +655,7 @@ public class JGroupsMessenger implements Messenger {
// Construct the list
calculatedLen = v.size();
- calculatedMembers = new LinkedList<GMSMember>();
+ calculatedMembers = new LinkedList<>();
for (int i = 0; i < calculatedLen; i ++) {
InternalDistributedMember m = (InternalDistributedMember)v.get(i);
calculatedMembers.add((GMSMember)m.getNetMember());
@@ -663,7 +663,7 @@ public class JGroupsMessenger implements Messenger {
} // send to all
else { // send to explicit list
calculatedLen = len;
- calculatedMembers = new LinkedList<GMSMember>();
+ calculatedMembers = new LinkedList<>();
for (int i = 0; i < calculatedLen; i ++) {
calculatedMembers.add((GMSMember)destinations[i].getNetMember());
}
@@ -671,10 +671,9 @@ public class JGroupsMessenger implements Messenger {
Int2ObjectOpenHashMap<Message> messages = new Int2ObjectOpenHashMap<>();
long startSer = theStats.startMsgSerialization();
boolean firstMessage = true;
- for (Iterator<GMSMember> it=calculatedMembers.iterator(); it.hasNext(); ) {
- GMSMember mbr = it.next();
+ for (GMSMember mbr : calculatedMembers) {
short version = mbr.getVersionOrdinal();
- if ( !messages.containsKey(version) ) {
+ if (!messages.containsKey(version)) {
Message jmsg = createJGMessage(msg, local, version);
messages.put(version, jmsg);
if (firstMessage) {
@@ -689,7 +688,7 @@ public class JGroupsMessenger implements Messenger {
for (GMSMember mbr: calculatedMembers) {
JGAddress to = new JGAddress(mbr);
short version = mbr.getVersionOrdinal();
- Message jmsg = (Message)messages.get(version);
+ Message jmsg = messages.get(version);
Exception problem = null;
try {
Message tmp = (i < (calculatedLen-1)) ? jmsg.copy(true) : jmsg;
@@ -732,11 +731,10 @@ public class JGroupsMessenger implements Messenger {
if (msg.forAll()) {
return Collections.emptySet();
}
- Set<InternalDistributedMember> result = new HashSet<InternalDistributedMember>();
+ Set<InternalDistributedMember> result = new HashSet<>();
NetView newView = this.view;
if (newView != null && newView != oldView) {
- for (int i = 0; i < destinations.length; i ++) {
- InternalDistributedMember d = destinations[i];
+ for (InternalDistributedMember d : destinations) {
if (!newView.contains(d)) {
logger.debug("messenger: member has left the view: {} view is now {}", d, newView);
result.add(d);
@@ -829,7 +827,7 @@ public class JGroupsMessenger implements Messenger {
return null;
}
- InternalDistributedMember sender = null;
+ InternalDistributedMember sender;
Exception problem = null;
byte[] buf = jgmsg.getRawBuffer();
@@ -916,12 +914,10 @@ public class JGroupsMessenger implements Messenger {
try {
Digest digest = new Digest();
digest.readFrom(dis);
- if (digest != null) {
- logger.trace("installing JGroups message digest {}", digest);
- this.myChannel.getProtocolStack()
- .getTopProtocol().down(new Event(Event.MERGE_DIGEST, digest));
- jrsp.setMessengerData(null);
- }
+ logger.trace("installing JGroups message digest {}", digest);
+ this.myChannel.getProtocolStack()
+ .getTopProtocol().down(new Event(Event.MERGE_DIGEST, digest));
+ jrsp.setMessengerData(null);
} catch (Exception e) {
logger.fatal("Unable to read JGroups messaging digest", e);
}
@@ -961,12 +957,13 @@ public class JGroupsMessenger implements Messenger {
/**
* returns the member ID for the given GMSMember object
*/
+ @SuppressWarnings("UnusedParameters")
private InternalDistributedMember getMemberFromView(GMSMember jgId, short version) {
NetView v = services.getJoinLeave().getView();
if (v != null) {
for (InternalDistributedMember m: v.getMembers()) {
- if (((GMSMember)m.getNetMember()).equals(jgId)) {
+ if (m.getNetMember().equals(jgId)) {
return m;
}
}
@@ -1090,7 +1087,7 @@ public class JGroupsMessenger implements Messenger {
}
}
if (h == null) {
- h = (MessageHandler)services.getManager();
+ h = services.getManager();
}
return h;
}