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 2018/08/08 22:28:31 UTC
[geode] branch feature/GEODE-5546 created (now c8da263)
This is an automated email from the ASF dual-hosted git repository.
bschuchardt pushed a change to branch feature/GEODE-5546
in repository https://gitbox.apache.org/repos/asf/geode.git.
at c8da263 GEODE-5546 auto-reconnecting member reuses old address including vmViewId
This branch includes the following new commits:
new c8da263 GEODE-5546 auto-reconnecting member reuses old address including vmViewId
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[geode] 01/01: GEODE-5546 auto-reconnecting member reuses old
address including vmViewId
Posted by bs...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
bschuchardt pushed a commit to branch feature/GEODE-5546
in repository https://gitbox.apache.org/repos/asf/geode.git
commit c8da2631cb5d51222d3e301fb628838fff140d0f
Author: Bruce Schuchardt <bs...@pivotal.io>
AuthorDate: Wed Aug 8 15:24:54 2018 -0700
GEODE-5546 auto-reconnecting member reuses old address including vmViewId
Old membership IDs are now retained in JGroupsMessenger and GMSJoinLeave
uses a new method, Messenger.isOldMembershipIdentifier(), to avoid accepting
a prepared view that contains an old identity.
GMSJoinLeave is also modified to send an immediate removal message to
servers that are no longer members of the cluster but are attempting to interact
with the cluster.
---
.../apache/geode/cache30/ReconnectDUnitTest.java | 9 ++--
.../gms/membership/GMSJoinLeaveJUnitTest.java | 63 ++++++++++++++++++++++
.../gms/messenger/GMSQuorumCheckerJUnitTest.java | 47 ++++++++--------
.../membership/gms/interfaces/Messenger.java | 6 +++
.../membership/gms/locator/GMSLocator.java | 2 -
.../membership/gms/membership/GMSJoinLeave.java | 19 +++++--
.../membership/gms/messenger/GMSQuorumChecker.java | 8 ++-
.../membership/gms/messenger/JGroupsMessenger.java | 36 ++++++++++---
.../gms/messenger/OldMembershipInformation.java | 41 ++++++++++++++
.../cache/tier/sockets/CacheClientNotifier.java | 3 --
10 files changed, 190 insertions(+), 44 deletions(-)
diff --git a/geode-core/src/distributedTest/java/org/apache/geode/cache30/ReconnectDUnitTest.java b/geode-core/src/distributedTest/java/org/apache/geode/cache30/ReconnectDUnitTest.java
index 65c41ad..4ec20a5 100755
--- a/geode-core/src/distributedTest/java/org/apache/geode/cache30/ReconnectDUnitTest.java
+++ b/geode-core/src/distributedTest/java/org/apache/geode/cache30/ReconnectDUnitTest.java
@@ -357,7 +357,7 @@ public class ReconnectDUnitTest extends JUnit4CacheTestCase {
};
vm0.invoke(create1);
- DistributedMember dm = (DistributedMember) vm1.invoke(create2);
+ final DistributedMember dm = (DistributedMember) vm1.invoke(create2);
IgnoredException.addIgnoredException("ForcedDisconnectException");
forceDisconnect(vm1);
@@ -391,6 +391,9 @@ public class ReconnectDUnitTest extends JUnit4CacheTestCase {
failure = false;
cache = ((InternalLocator) locator).getCache();
system = cache.getInternalDistributedSystem();
+ assertTrue(
+ ((GMSMembershipManager) MembershipManagerHelper.getMembershipManager(system))
+ .getServices().getMessenger().isOldMembershipIdentifier(dm));
return ds.getReconnectedSystem().getDistributedMember();
} catch (InterruptedException e) {
LogWriterUtils.getLogWriter().warning("interrupted while waiting for reconnect");
@@ -434,10 +437,10 @@ public class ReconnectDUnitTest extends JUnit4CacheTestCase {
assertTrue("expected DistributedSystem to disconnect", stopped);
// recreate the system in vm1 without a locator and crash it
- dm = (DistributedMember) vm1.invoke(create1);
+ DistributedMember evenNewerdm = (DistributedMember) vm1.invoke(create1);
forceDisconnect(vm1);
newdm = waitForReconnect(vm1);
- assertNotSame("expected a reconnect to occur in member", dm, newdm);
+ assertNotSame("expected a reconnect to occur in member", evenNewerdm, newdm);
DistributedTestUtils.deleteLocatorStateFile(locPort);
DistributedTestUtils.deleteLocatorStateFile(secondLocPort);
}
diff --git a/geode-core/src/integrationTest/java/org/apache/geode/distributed/internal/membership/gms/membership/GMSJoinLeaveJUnitTest.java b/geode-core/src/integrationTest/java/org/apache/geode/distributed/internal/membership/gms/membership/GMSJoinLeaveJUnitTest.java
index 2697051..0ff678c 100644
--- a/geode-core/src/integrationTest/java/org/apache/geode/distributed/internal/membership/gms/membership/GMSJoinLeaveJUnitTest.java
+++ b/geode-core/src/integrationTest/java/org/apache/geode/distributed/internal/membership/gms/membership/GMSJoinLeaveJUnitTest.java
@@ -21,9 +21,11 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Matchers.isA;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.reset;
import static org.mockito.Mockito.timeout;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
@@ -563,6 +565,67 @@ public class GMSJoinLeaveJUnitTest {
}
@Test
+ public void testRemoveMessageForRogueCausesImmediateRemovalMessageToRogue() throws Exception {
+ initMocks();
+ synchronized (gmsJoinLeave.getViewInstallationLock()) {
+ gmsJoinLeave.becomeCoordinator();
+ }
+ prepareAndInstallView(gmsJoinLeaveMemberId,
+ createMemberList(gmsJoinLeaveMemberId, mockMembers[0], mockMembers[1]));
+ reset(messenger);
+ RemoveMemberMessage msg = new RemoveMemberMessage(gmsJoinLeaveMemberId,
+ new InternalDistributedMember("localhost", 10000), "removing for test");
+ msg.setSender(mockMembers[0]);
+ gmsJoinLeave.processMessage(msg);
+ verify(messenger).send(isA(RemoveMemberMessage.class));
+ }
+
+ @Test
+ public void testRemoveRequestCausesForcedDisconnectInRogue() throws Exception {
+ initMocks();
+ // gmsJoinLeave mistakenly uses an old viewID when joining, making it a rogue member
+ gmsJoinLeaveMemberId.setVmViewId(-1);
+ InternalDistributedMember previousMemberId =
+ new InternalDistributedMember(gmsJoinLeaveMemberId.getId(), gmsJoinLeaveMemberId.getPort());
+ previousMemberId.setVmViewId(0);
+ NetView view = new NetView(mockMembers[0], 1,
+ createMemberList(mockMembers[0], previousMemberId, mockMembers[1]));
+ InstallViewMessage viewMessage = new InstallViewMessage(view, 0, true);
+ viewMessage.setSender(mockMembers[0]);
+ gmsJoinLeave.processMessage(viewMessage);
+ assertEquals(0, gmsJoinLeaveMemberId.getVmViewId());
+ // a RemoveMember message should cause it to force-disconnect
+ RemoveMemberMessage msg =
+ new RemoveMemberMessage(gmsJoinLeaveMemberId, gmsJoinLeaveMemberId, "removing for test");
+ msg.setSender(mockMembers[0]);
+ gmsJoinLeave.processMessage(msg);
+ verify(manager).forceDisconnect("removing for test");
+ }
+
+ @Test
+ public void testViewWithOldIDNotAcceptedAsJoinResponse() throws Exception {
+ initMocks();
+ when(messenger.isOldMembershipIdentifier(any(DistributedMember.class)))
+ .thenReturn(Boolean.TRUE);
+ List<InternalDistributedMember> mbrs = new LinkedList<>();
+ Set<InternalDistributedMember> shutdowns = new HashSet<>();
+ Set<InternalDistributedMember> crashes = new HashSet<>();
+ mbrs.add(mockMembers[0]);
+ mbrs.add(mockMembers[1]);
+ mbrs.add(mockMembers[2]);
+ InternalDistributedMember oldId = new InternalDistributedMember(
+ gmsJoinLeaveMemberId.getInetAddress(), gmsJoinLeaveMemberId.getPort());
+ oldId.setVmViewId(0);
+ mbrs.add(oldId);
+
+ // prepare the view
+ NetView netView = new NetView(mockMembers[0], 1, mbrs, shutdowns, crashes);
+ gmsJoinLeave.processMessage(new InstallViewMessage(netView, null, true));
+ assertEquals(-1, gmsJoinLeaveMemberId.getVmViewId());
+ verify(messenger).isOldMembershipIdentifier(isA(DistributedMember.class));
+ }
+
+ @Test
public void testRemoveCausesForcedDisconnect() throws Exception {
String reason = "testing";
initMocks();
diff --git a/geode-core/src/integrationTest/java/org/apache/geode/distributed/internal/membership/gms/messenger/GMSQuorumCheckerJUnitTest.java b/geode-core/src/integrationTest/java/org/apache/geode/distributed/internal/membership/gms/messenger/GMSQuorumCheckerJUnitTest.java
index c1f5178..e479553 100644
--- a/geode-core/src/integrationTest/java/org/apache/geode/distributed/internal/membership/gms/messenger/GMSQuorumCheckerJUnitTest.java
+++ b/geode-core/src/integrationTest/java/org/apache/geode/distributed/internal/membership/gms/messenger/GMSQuorumCheckerJUnitTest.java
@@ -21,7 +21,6 @@ import static org.mockito.Matchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
-import java.io.IOException;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
@@ -53,7 +52,7 @@ public class GMSQuorumCheckerJUnitTest {
private JGAddress address;
@Before
- public void initMocks() throws Exception {
+ public void initMocks() {
mockMembers = new InternalDistributedMember[12];
for (int i = 0; i < mockMembers.length; i++) {
mockMembers[i] = new InternalDistributedMember("localhost", 8888 + i);
@@ -70,14 +69,14 @@ public class GMSQuorumCheckerJUnitTest {
@Test
public void testQuorumCheckerAllRespond() throws Exception {
NetView view = prepareView();
- Set<Integer> pongResponders = new HashSet<Integer>();
+ Set<Integer> pongResponders = new HashSet<>();
for (int i = 0; i < mockMembers.length; i++) {
pongResponders.add(mockMembers[i].getPort());
}
PingMessageAnswer answerer = new PingMessageAnswer(channel, pongResponders);
Mockito.doAnswer(answerer).when(channel).send(any(Message.class));
- GMSQuorumChecker qc = new GMSQuorumChecker(view, 51, channel);
+ GMSQuorumChecker qc = new GMSQuorumChecker(view, 51, channel, null);
qc.initialize();
boolean quorum = qc.checkForQuorum(500);
assertTrue(quorum);
@@ -89,14 +88,14 @@ public class GMSQuorumCheckerJUnitTest {
@Test
public void testQuorumCheckerMajorityRespond() throws Exception {
NetView view = prepareView();
- Set<Integer> pongResponders = new HashSet<Integer>();
+ Set<Integer> pongResponders = new HashSet<>();
for (int i = 0; i < mockMembers.length - 1; i++) {
pongResponders.add(mockMembers[i].getPort());
}
PingMessageAnswer answerer = new PingMessageAnswer(channel, pongResponders);
Mockito.doAnswer(answerer).when(channel).send(any(Message.class));
- GMSQuorumChecker qc = new GMSQuorumChecker(view, 51, channel);
+ GMSQuorumChecker qc = new GMSQuorumChecker(view, 51, channel, null);
qc.initialize();
boolean quorum = qc.checkForQuorum(500);
assertTrue(quorum);
@@ -106,12 +105,12 @@ public class GMSQuorumCheckerJUnitTest {
@Test
public void testQuorumCheckerNotEnoughWeightForQuorum() throws Exception {
NetView view = prepareView();
- Set<Integer> pongResponders = new HashSet<Integer>();
+ Set<Integer> pongResponders = new HashSet<>();
pongResponders.add(mockMembers[0].getPort());
PingMessageAnswer answerer = new PingMessageAnswer(channel, pongResponders);
Mockito.doAnswer(answerer).when(channel).send(any(Message.class));
- GMSQuorumChecker qc = new GMSQuorumChecker(view, 51, channel);
+ GMSQuorumChecker qc = new GMSQuorumChecker(view, 51, channel, null);
qc.initialize();
boolean quorum = qc.checkForQuorum(500);
assertFalse(quorum);
@@ -125,7 +124,7 @@ public class GMSQuorumCheckerJUnitTest {
PingMessageAnswer answerer = new PingMessageAnswer(channel, pongResponders);
Mockito.doAnswer(answerer).when(channel).send(any(Message.class));
- GMSQuorumChecker qc = new GMSQuorumChecker(view, 51, channel);
+ GMSQuorumChecker qc = new GMSQuorumChecker(view, 51, channel, null);
qc.initialize();
boolean quorum = qc.checkForQuorum(500);
assertFalse(quorum);
@@ -138,7 +137,7 @@ public class GMSQuorumCheckerJUnitTest {
mockMembers[0].setVmKind(ClusterDistributionManager.LOCATOR_DM_TYPE);
mockMembers[1].setVmKind(ClusterDistributionManager.LOCATOR_DM_TYPE);
- Set<Integer> pongResponders = new HashSet<Integer>();
+ Set<Integer> pongResponders = new HashSet<>();
for (int i = 0; i < mockMembers.length; i++) {
pongResponders.add(mockMembers[i].getPort());
}
@@ -152,7 +151,7 @@ public class GMSQuorumCheckerJUnitTest {
PingMessageAnswer answerer = new PingMessageAnswer(channel, pongResponders);
Mockito.doAnswer(answerer).when(channel).send(any(Message.class));
- GMSQuorumChecker qc = new GMSQuorumChecker(view, 51, channel);
+ GMSQuorumChecker qc = new GMSQuorumChecker(view, 51, channel, null);
qc.initialize();
boolean quorum = qc.checkForQuorum(500);
assertTrue(quorum);
@@ -165,7 +164,7 @@ public class GMSQuorumCheckerJUnitTest {
mockMembers[0].setVmKind(ClusterDistributionManager.LOCATOR_DM_TYPE);
mockMembers[1].setVmKind(ClusterDistributionManager.LOCATOR_DM_TYPE);
- Set<Integer> pongResponders = new HashSet<Integer>();
+ Set<Integer> pongResponders = new HashSet<>();
for (int i = 0; i < mockMembers.length; i++) {
pongResponders.add(mockMembers[i].getPort());
}
@@ -182,7 +181,7 @@ public class GMSQuorumCheckerJUnitTest {
PingMessageAnswer answerer = new PingMessageAnswer(channel, pongResponders);
Mockito.doAnswer(answerer).when(channel).send(any(Message.class));
- GMSQuorumChecker qc = new GMSQuorumChecker(view, 51, channel);
+ GMSQuorumChecker qc = new GMSQuorumChecker(view, 51, channel, null);
qc.initialize();
boolean quorum = qc.checkForQuorum(500);
assertTrue(quorum);
@@ -196,7 +195,7 @@ public class GMSQuorumCheckerJUnitTest {
mockMembers[0].setVmKind(ClusterDistributionManager.LOCATOR_DM_TYPE);
mockMembers[1].setVmKind(ClusterDistributionManager.LOCATOR_DM_TYPE);
- Set<Integer> pongResponders = new HashSet<Integer>();
+ Set<Integer> pongResponders = new HashSet<>();
for (int i = 0; i < mockMembers.length; i++) {
pongResponders.add(mockMembers[i].getPort());
}
@@ -215,7 +214,7 @@ public class GMSQuorumCheckerJUnitTest {
PingMessageAnswer answerer = new PingMessageAnswer(channel, pongResponders);
Mockito.doAnswer(answerer).when(channel).send(any(Message.class));
- GMSQuorumChecker qc = new GMSQuorumChecker(view, 51, channel);
+ GMSQuorumChecker qc = new GMSQuorumChecker(view, 51, channel, null);
qc.initialize();
boolean quorum = qc.checkForQuorum(500);
assertFalse(quorum);
@@ -229,7 +228,7 @@ public class GMSQuorumCheckerJUnitTest {
mockMembers[0].setVmKind(ClusterDistributionManager.LOCATOR_DM_TYPE);
mockMembers[1].setVmKind(ClusterDistributionManager.LOCATOR_DM_TYPE);
- Set<Integer> pongResponders = new HashSet<Integer>();
+ Set<Integer> pongResponders = new HashSet<>();
for (int i = 0; i < mockMembers.length; i++) {
pongResponders.add(mockMembers[i].getPort());
}
@@ -246,7 +245,7 @@ public class GMSQuorumCheckerJUnitTest {
PingMessageAnswer answerer = new PingMessageAnswer(channel, pongResponders);
Mockito.doAnswer(answerer).when(channel).send(any(Message.class));
- GMSQuorumChecker qc = new GMSQuorumChecker(view, 51, channel);
+ GMSQuorumChecker qc = new GMSQuorumChecker(view, 51, channel, null);
qc.initialize();
boolean quorum = qc.checkForQuorum(500);
assertFalse(quorum);
@@ -260,7 +259,7 @@ public class GMSQuorumCheckerJUnitTest {
mockMembers[0].setVmKind(ClusterDistributionManager.LOCATOR_DM_TYPE);
mockMembers[1].setVmKind(ClusterDistributionManager.LOCATOR_DM_TYPE);
- Set<Integer> pongResponders = new HashSet<Integer>();
+ Set<Integer> pongResponders = new HashSet<>();
for (int i = 0; i < numMembers; i++) {
pongResponders.add(mockMembers[i].getPort());
}
@@ -270,7 +269,7 @@ public class GMSQuorumCheckerJUnitTest {
PingMessageAnswer answerer = new PingMessageAnswer(channel, pongResponders);
Mockito.doAnswer(answerer).when(channel).send(any(Message.class));
- GMSQuorumChecker qc = new GMSQuorumChecker(view, 51, channel);
+ GMSQuorumChecker qc = new GMSQuorumChecker(view, 51, channel, null);
qc.initialize();
boolean quorum = qc.checkForQuorum(500);
assertTrue(quorum);
@@ -284,7 +283,7 @@ public class GMSQuorumCheckerJUnitTest {
mockMembers[0].setVmKind(ClusterDistributionManager.LOCATOR_DM_TYPE);
mockMembers[1].setVmKind(ClusterDistributionManager.LOCATOR_DM_TYPE);
- Set<Integer> pongResponders = new HashSet<Integer>();
+ Set<Integer> pongResponders = new HashSet<>();
for (int i = 0; i < numMembers; i++) {
pongResponders.add(mockMembers[i].getPort());
}
@@ -295,20 +294,20 @@ public class GMSQuorumCheckerJUnitTest {
PingMessageAnswer answerer = new PingMessageAnswer(channel, pongResponders);
Mockito.doAnswer(answerer).when(channel).send(any(Message.class));
- GMSQuorumChecker qc = new GMSQuorumChecker(view, 51, channel);
+ GMSQuorumChecker qc = new GMSQuorumChecker(view, 51, channel, null);
qc.initialize();
boolean quorum = qc.checkForQuorum(500);
assertFalse(quorum);
assertSame(view.getMembers().size(), answerer.getPingCount());
}
- private NetView prepareView() throws IOException {
+ private NetView prepareView() {
return prepareView(mockMembers.length);
}
- private NetView prepareView(int numMembers) throws IOException {
+ private NetView prepareView(int numMembers) {
int viewId = 1;
- List<InternalDistributedMember> mbrs = new LinkedList<InternalDistributedMember>();
+ List<InternalDistributedMember> mbrs = new LinkedList<>();
for (int i = 0; i < numMembers; i++) {
mbrs.add(mockMembers[i]);
}
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/gms/interfaces/Messenger.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/gms/interfaces/Messenger.java
index 080f0da..c0b3c01 100755
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/gms/interfaces/Messenger.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/gms/interfaces/Messenger.java
@@ -17,6 +17,7 @@ package org.apache.geode.distributed.internal.membership.gms.interfaces;
import java.util.Map;
import java.util.Set;
+import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.internal.DistributionMessage;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.distributed.internal.membership.NetView;
@@ -52,6 +53,11 @@ public interface Messenger extends Service {
InternalDistributedMember getMemberID();
/**
+ * check to see if a member ID has already been used
+ */
+ boolean isOldMembershipIdentifier(DistributedMember id);
+
+ /**
* retrieves the quorum checker that is used during auto-reconnect attempts
*/
QuorumChecker getQuorumChecker();
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/gms/locator/GMSLocator.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/gms/locator/GMSLocator.java
index e6822c4..f9a8ddb 100644
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/gms/locator/GMSLocator.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/gms/locator/GMSLocator.java
@@ -299,10 +299,8 @@ public class GMSLocator implements Locator, NetLocator {
}
fromView = viewCoordinator != null && !viewCoordinator.equals(localAddress);
if (!fromView) {
- logger.info("This member is becoming coordinator");
v = null;
}
- logger.debug("this member is becoming coordinator from view {} ", fromView);
}
byte[] coordPk = null;
if (v != null) {
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/gms/membership/GMSJoinLeave.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/gms/membership/GMSJoinLeave.java
index 770996a..498a85b 100644
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/gms/membership/GMSJoinLeave.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/gms/membership/GMSJoinLeave.java
@@ -690,6 +690,14 @@ public class GMSJoinLeave implements JoinLeave, MessageHandler {
this.prepareProcessor.processRemoveRequest(mbr);
}
}
+ if (isCoordinator) {
+ if (!v.contains(mbr)) {
+ // removing a rogue process
+ RemoveMemberMessage removeMemberMessage = new RemoveMemberMessage(mbr, mbr,
+ incomingRequest.getReason());
+ services.getMessenger().send(removeMemberMessage);
+ }
+ }
}
}
@@ -1004,7 +1012,8 @@ public class GMSJoinLeave implements JoinLeave, MessageHandler {
// if we're still waiting for a join response and we're in this view we
// should install the view so join() can finish its work
for (InternalDistributedMember mbr : view.getMembers()) {
- if (localAddress.compareTo(mbr) == 0) {
+ if (localAddress.compareTo(mbr) == 0
+ && !services.getMessenger().isOldMembershipIdentifier(mbr)) {
viewContainsMyUnjoinedAddress = true;
break;
}
@@ -2219,10 +2228,11 @@ public class GMSJoinLeave implements JoinLeave, MessageHandler {
continue;
}
} else {
- if (System.currentTimeMillis() < okayToCreateView) {
+ long timeRemaining = okayToCreateView - System.currentTimeMillis();
+ if (timeRemaining > 0) {
// sleep to let more requests arrive
try {
- viewRequests.wait(100);
+ viewRequests.wait(Math.min(100, timeRemaining));
continue;
} catch (InterruptedException e) {
return;
@@ -2389,9 +2399,10 @@ public class GMSJoinLeave implements JoinLeave, MessageHandler {
removalReqs.add(mbr);
removalReasons.add(((RemoveMemberMessage) msg).getReason());
} else {
+ // unknown, probably rogue, process - send it a removal message
sendRemoveMessages(Collections.singletonList(mbr),
Collections.singletonList(((RemoveMemberMessage) msg).getReason()),
- new HashSet<InternalDistributedMember>());
+ new HashSet<>());
}
}
break;
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/gms/messenger/GMSQuorumChecker.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/gms/messenger/GMSQuorumChecker.java
index ceea47b..212d789 100644
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/gms/messenger/GMSQuorumChecker.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/gms/messenger/GMSQuorumChecker.java
@@ -32,6 +32,7 @@ import org.jgroups.Message;
import org.jgroups.Receiver;
import org.jgroups.View;
+import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.distributed.internal.membership.NetView;
import org.apache.geode.distributed.internal.membership.QuorumChecker;
@@ -53,11 +54,14 @@ public class GMSQuorumChecker implements QuorumChecker {
private final JChannel channel;
private JGAddress myAddress;
private final int partitionThreshold;
+ private Set<DistributedMember> oldDistributedMemberIdentifiers;
- public GMSQuorumChecker(NetView jgView, int partitionThreshold, JChannel channel) {
+ public GMSQuorumChecker(NetView jgView, int partitionThreshold, JChannel channel,
+ Set<DistributedMember> oldDistributedMemberIdentifiers) {
this.lastView = jgView;
this.partitionThreshold = partitionThreshold;
this.channel = channel;
+ this.oldDistributedMemberIdentifiers = oldDistributedMemberIdentifiers;
}
public void initialize() {
@@ -121,7 +125,7 @@ public class GMSQuorumChecker implements QuorumChecker {
@Override
public Object getMembershipInfo() {
- return channel;
+ return new OldMembershipInformation(channel, oldDistributedMemberIdentifiers);
}
private boolean calculateQuorum() {
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/gms/messenger/JGroupsMessenger.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/gms/messenger/JGroupsMessenger.java
index 71ca91e..7af9b90 100644
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/gms/messenger/JGroupsMessenger.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/gms/messenger/JGroupsMessenger.java
@@ -68,6 +68,7 @@ import org.apache.geode.DataSerializer;
import org.apache.geode.ForcedDisconnectException;
import org.apache.geode.GemFireConfigException;
import org.apache.geode.GemFireIOException;
+import org.apache.geode.InternalGemFireError;
import org.apache.geode.SystemConnectException;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.DistributedSystemDisconnectedException;
@@ -167,6 +168,12 @@ public class JGroupsMessenger implements Messenger {
private GMSEncrypt encrypt;
+ /**
+ * DistributedMember identifiers already used, either in this JGroupsMessenger instance
+ * or in a past one & retained through an auto-reconnect.
+ */
+ private Set<DistributedMember> usedDistributedMemberIdentifiers = new HashSet<>();
+
@Override
@edu.umd.cs.findbugs.annotations.SuppressWarnings(
value = "ST_WRITE_TO_STATIC_FROM_INSTANCE_METHOD")
@@ -298,11 +305,13 @@ public class JGroupsMessenger implements Messenger {
// start the jgroups channel and establish the membership ID
boolean reconnecting = false;
try {
- Object oldChannel = services.getConfig().getTransport().getOldDSMembershipInfo();
- if (oldChannel != null) {
+ Object oldDSMembershipInfo = services.getConfig().getTransport().getOldDSMembershipInfo();
+ if (oldDSMembershipInfo != null) {
logger.debug("Reusing JGroups channel from previous system", properties);
+ OldMembershipInformation oldInfo = (OldMembershipInformation) oldDSMembershipInfo;
+ myChannel = oldInfo.getChannel();
+ usedDistributedMemberIdentifiers = oldInfo.getOldMembershipIdentifiers();
- myChannel = (JChannel) oldChannel;
// scrub the old channel
ViewId vid = new ViewId(new JGAddress(), 0);
List<Address> members = new ArrayList<>();
@@ -359,6 +368,11 @@ public class JGroupsMessenger implements Messenger {
}
+ @Override
+ public boolean isOldMembershipIdentifier(DistributedMember id) {
+ return usedDistributedMemberIdentifiers.contains(id);
+ }
+
/**
* JGroups picks an IPv6 address if preferIPv4Stack is false or not set and preferIPv6Addresses is
* not set or is true. We want it to use an IPv4 address for a dual-IP stack so that both IPv4 and
@@ -379,6 +393,10 @@ public class JGroupsMessenger implements Messenger {
@Override
public void stop() {
+ if (localAddress.getVmViewId() >= 0) {
+ // keep track of old addresses that were used to successfully join the cluster
+ usedDistributedMemberIdentifiers.add(localAddress);
+ }
if (this.myChannel != null) {
if ((services.isShutdownDueToForcedDisconnect() && services.isAutoReconnectEnabled())
|| services.getManager().isReconnectingDS()) {
@@ -481,7 +499,9 @@ public class JGroupsMessenger implements Messenger {
this.jgAddress = new JGAddress(logicalAddress, ipaddr);
} catch (NoSuchMethodException | InvocationTargetException | IllegalAccessException e) {
logger
- .info("Unable to find getPhysicallAddress method in UDP - parsing its address instead");
+ .info("Unable to find getPhysicallAddress method in UDP");
+ throw new InternalGemFireError(
+ "unable to configure JGroups channel for membership communications", e);
}
}
@@ -513,7 +533,6 @@ public class JGroupsMessenger implements Messenger {
gmsMember.setMemberWeight((byte) (services.getConfig().getMemberWeight() & 0xff));
gmsMember.setNetworkPartitionDetectionEnabled(
services.getConfig().getDistributionConfig().getEnableNetworkPartitionDetection());
-
}
@Override
@@ -1193,6 +1212,10 @@ public class JGroupsMessenger implements Messenger {
@Override
public void emergencyClose() {
this.view = null;
+ if (localAddress.getVmViewId() >= 0) {
+ // keep track of old addresses that were used to successfully join the cluster
+ usedDistributedMemberIdentifiers.add(localAddress);
+ }
if (this.myChannel != null) {
if ((services.isShutdownDueToForcedDisconnect() && services.isAutoReconnectEnabled())
|| services.getManager().isReconnectingDS()) {
@@ -1214,7 +1237,8 @@ public class JGroupsMessenger implements Messenger {
}
}
GMSQuorumChecker qc =
- new GMSQuorumChecker(view, services.getConfig().getLossThreshold(), this.myChannel);
+ new GMSQuorumChecker(view, services.getConfig().getLossThreshold(), this.myChannel,
+ usedDistributedMemberIdentifiers);
qc.initialize();
return qc;
}
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/gms/messenger/OldMembershipInformation.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/gms/messenger/OldMembershipInformation.java
new file mode 100644
index 0000000..4ad8108
--- /dev/null
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/gms/messenger/OldMembershipInformation.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package org.apache.geode.distributed.internal.membership.gms.messenger;
+
+import java.util.Set;
+
+import org.jgroups.JChannel;
+
+import org.apache.geode.distributed.DistributedMember;
+
+class OldMembershipInformation {
+ private final JChannel channel;
+ private final Set<DistributedMember> oldMembershipIdentifiers;
+
+ protected OldMembershipInformation(JChannel channel,
+ Set<DistributedMember> oldMembershipIdentifiers) {
+
+ this.channel = channel;
+ this.oldMembershipIdentifiers = oldMembershipIdentifiers;
+ }
+
+ public JChannel getChannel() {
+ return channel;
+ }
+
+ public Set<DistributedMember> getOldMembershipIdentifiers() {
+ return oldMembershipIdentifiers;
+ }
+}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientNotifier.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientNotifier.java
index 743cac8..bf90825 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientNotifier.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientNotifier.java
@@ -1034,11 +1034,8 @@ public class CacheClientNotifier {
// try to canonicalize the ID.
CacheClientProxy proxy = getClientProxy((ClientProxyMembershipID) id, true);
if (proxy != null) {
- // this._logger.info(LocalizedStrings.DEBUG, "BRUCE: found match for " + id + ": " +
- // proxy.getProxyID());
result.add(proxy.getProxyID());
} else {
- // this._logger.info(LocalizedStrings.DEBUG, "BRUCE: did not find match for " + id);
// this was causing OOMEs in HARegion initial image processing because
// messages had routing for clients unknown to this server
// result.add((ClientProxyMembershipID)id);