This is an automated email from the ASF dual-hosted git repository.
bschuchardt pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/geode.git
The following commit(s) were added to refs/heads/develop by this push:
new b08e37f GEODE-5546 auto-reconnecting member reuses old address
including vmViewId
b08e37f is described below
commit b08e37fba1261c118acf9d264f46c048dd519276
Author: Bruce Schuchardt <[email protected]>
AuthorDate: Fri Aug 10 12:25:34 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.
This closes #2286
---
.../apache/geode/cache30/ReconnectDUnitTest.java | 9 ++--
.../gms/membership/GMSJoinLeaveJUnitTest.java | 63 ++++++++++++++++++++++
.../gms/messenger/GMSQuorumCheckerJUnitTest.java | 50 ++++++++---------
.../gms/messenger/JGroupsMessengerJUnitTest.java | 4 +-
.../internal/InternalDistributedSystem.java | 3 +-
.../internal/membership/QuorumChecker.java | 4 +-
.../membership/gms/interfaces/Messenger.java | 6 +++
.../membership/gms/locator/GMSLocator.java | 2 -
.../membership/gms/membership/GMSJoinLeave.java | 27 +++++++---
.../membership/gms/messenger/GMSQuorumChecker.java | 10 ++--
.../membership/gms/messenger/JGroupsMessenger.java | 36 ++++++++++---
.../gms/messenger/MembershipInformation.java | 45 ++++++++++++++++
.../admin/remote/RemoteTransportConfig.java | 7 +--
.../cache/tier/sockets/CacheClientNotifier.java | 3 --
14 files changed, 212 insertions(+), 57 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..ea8a082 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,33 +69,34 @@ 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);
assertSame(view.getMembers().size(), answerer.getPingCount());
assertTrue(qc.checkForQuorum(500));
- assertSame(qc.getMembershipInfo(), channel);
+ assertSame(MembershipInformation.class, qc.getMembershipInfo().getClass());
+ assertSame(((MembershipInformation) qc.getMembershipInfo()).getChannel(),
channel);
}
@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 +106,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 +125,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 +138,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 +152,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 +165,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 +182,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 +196,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 +215,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 +229,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 +246,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 +260,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 +270,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 +284,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 +295,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/integrationTest/java/org/apache/geode/distributed/internal/membership/gms/messenger/JGroupsMessengerJUnitTest.java
b/geode-core/src/integrationTest/java/org/apache/geode/distributed/internal/membership/gms/messenger/JGroupsMessengerJUnitTest.java
index ce7990d..37cbf9e 100755
---
a/geode-core/src/integrationTest/java/org/apache/geode/distributed/internal/membership/gms/messenger/JGroupsMessengerJUnitTest.java
+++
b/geode-core/src/integrationTest/java/org/apache/geode/distributed/internal/membership/gms/messenger/JGroupsMessengerJUnitTest.java
@@ -45,6 +45,7 @@ import java.io.DataInputStream;
import java.io.DataOutput;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -879,7 +880,8 @@ public class JGroupsMessengerJUnitTest {
public void testUseOldJChannel() throws Exception {
initMocks(false);
JChannel channel = messenger.myChannel;
- services.getConfig().getTransport().setOldDSMembershipInfo(channel);
+ services.getConfig().getTransport().setOldDSMembershipInfo(new
MembershipInformation(channel,
+ Collections.singleton(new InternalDistributedMember("localhost",
10000))));
JGroupsMessenger newMessenger = new JGroupsMessenger();
newMessenger.init(services);
newMessenger.start();
diff --git
a/geode-core/src/main/java/org/apache/geode/distributed/internal/InternalDistributedSystem.java
b/geode-core/src/main/java/org/apache/geode/distributed/internal/InternalDistributedSystem.java
index c1e7c11..9d992a1 100644
---
a/geode-core/src/main/java/org/apache/geode/distributed/internal/InternalDistributedSystem.java
+++
b/geode-core/src/main/java/org/apache/geode/distributed/internal/InternalDistributedSystem.java
@@ -72,6 +72,7 @@ import
org.apache.geode.distributed.internal.membership.InternalDistributedMembe
import org.apache.geode.distributed.internal.membership.MembershipManager;
import org.apache.geode.distributed.internal.membership.QuorumChecker;
import org.apache.geode.distributed.internal.membership.gms.Services;
+import
org.apache.geode.distributed.internal.membership.gms.messenger.MembershipInformation;
import
org.apache.geode.distributed.internal.membership.gms.mgr.GMSMembershipManager;
import org.apache.geode.i18n.LogWriterI18n;
import org.apache.geode.internal.Assert;
@@ -2472,7 +2473,7 @@ public class InternalDistributedSystem extends
DistributedSystem
* isReconnectingDS returns true. This is used to connect the new DM to the
distributed system
* through RemoteTransportConfig.
*/
- public Object oldDSMembershipInfo() {
+ public MembershipInformation oldDSMembershipInfo() {
if (this.quorumChecker != null) {
return this.quorumChecker.getMembershipInfo();
}
diff --git
a/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/QuorumChecker.java
b/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/QuorumChecker.java
index c480b8a..2fafbae 100644
---
a/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/QuorumChecker.java
+++
b/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/QuorumChecker.java
@@ -15,6 +15,8 @@
package org.apache.geode.distributed.internal.membership;
+import
org.apache.geode.distributed.internal.membership.gms.messenger.MembershipInformation;
+
/**
* A QuorumChecker is created after a forced-disconnect in order to probe the
network to see if
* there is a quorum of members that can be contacted.
@@ -50,7 +52,7 @@ public interface QuorumChecker {
* Get the membership info from the old system that needs to be passed to
the one that is
* reconnecting.
*/
- Object getMembershipInfo();
+ MembershipInformation getMembershipInfo();
/**
* Returns the membership view that is being used to establish a quorum
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..45b17bf 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);
+ }
+ }
}
}
@@ -999,13 +1007,14 @@ public class GMSJoinLeave implements JoinLeave,
MessageHandler {
return;
}
- boolean viewContainsMyUnjoinedAddress = false;
+ boolean viewContainsMyNewAddress = false;
if (!this.isJoined) {
// 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) {
- viewContainsMyUnjoinedAddress = true;
+ if (localAddress.equals(mbr)
+ && !services.getMessenger().isOldMembershipIdentifier(mbr)) {
+ viewContainsMyNewAddress = true;
break;
}
}
@@ -1017,7 +1026,7 @@ public class GMSJoinLeave implements JoinLeave,
MessageHandler {
.send(new ViewAckMessage(view.getViewId(), m.getSender(),
this.preparedView));
} else {
this.preparedView = view;
- if (viewContainsMyUnjoinedAddress) {
+ if (viewContainsMyNewAddress) {
installView(view); // this will notifyAll the joinResponse
}
ackView(m);
@@ -1029,7 +1038,7 @@ public class GMSJoinLeave implements JoinLeave,
MessageHandler {
localAddress, view);
forceDisconnect("This node is no longer in the membership view");
} else {
- if (isJoined || viewContainsMyUnjoinedAddress) {
+ if (isJoined || viewContainsMyNewAddress) {
installView(view);
}
if (!m.isRebroadcast()) { // no need to ack a rebroadcast view
@@ -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..32c802e 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() {
@@ -120,8 +124,8 @@ public class GMSQuorumChecker implements QuorumChecker {
}
@Override
- public Object getMembershipInfo() {
- return channel;
+ public MembershipInformation getMembershipInfo() {
+ return new MembershipInformation(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..fdb61af 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);
+ MembershipInformation oldInfo = (MembershipInformation)
oldDSMembershipInfo;
+ myChannel = oldInfo.getChannel();
+ usedDistributedMemberIdentifiers = oldInfo.getMembershipIdentifiers();
- 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 != null && 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()) {
@@ -480,8 +498,8 @@ public class JGroupsMessenger implements Messenger {
ipaddr = (IpAddress) getAddress.invoke(udp, new Object[0]);
this.jgAddress = new JGAddress(logicalAddress, ipaddr);
} catch (NoSuchMethodException | InvocationTargetException |
IllegalAccessException e) {
- logger
- .info("Unable to find getPhysicallAddress method in UDP - parsing
its address instead");
+ throw new InternalGemFireError(
+ "Unable to configure JGroups channel for membership
communications", e);
}
}
@@ -513,7 +531,6 @@ public class JGroupsMessenger implements Messenger {
gmsMember.setMemberWeight((byte) (services.getConfig().getMemberWeight() &
0xff));
gmsMember.setNetworkPartitionDetectionEnabled(
services.getConfig().getDistributionConfig().getEnableNetworkPartitionDetection());
-
}
@Override
@@ -1193,6 +1210,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 +1235,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/MembershipInformation.java
b/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/gms/messenger/MembershipInformation.java
new file mode 100644
index 0000000..adcfc43
--- /dev/null
+++
b/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/gms/messenger/MembershipInformation.java
@@ -0,0 +1,45 @@
+/*
+ * 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 MembershipInformation is used to pass membership data from a GMS that
was
+ * kicked out of the cluster to a new one during auto-reconnect operations.
+ */
+public class MembershipInformation {
+ private final JChannel channel;
+ private final Set<DistributedMember> membershipIdentifiers;
+
+ protected MembershipInformation(JChannel channel,
+ Set<DistributedMember> oldMembershipIdentifiers) {
+
+ this.channel = channel;
+ this.membershipIdentifiers = oldMembershipIdentifiers;
+ }
+
+ public JChannel getChannel() {
+ return channel;
+ }
+
+ public Set<DistributedMember> getMembershipIdentifiers() {
+ return membershipIdentifiers;
+ }
+}
diff --git
a/geode-core/src/main/java/org/apache/geode/internal/admin/remote/RemoteTransportConfig.java
b/geode-core/src/main/java/org/apache/geode/internal/admin/remote/RemoteTransportConfig.java
index 4dd70d1..0b5f880 100644
---
a/geode-core/src/main/java/org/apache/geode/internal/admin/remote/RemoteTransportConfig.java
+++
b/geode-core/src/main/java/org/apache/geode/internal/admin/remote/RemoteTransportConfig.java
@@ -34,6 +34,7 @@ import java.util.StringTokenizer;
import org.apache.commons.lang.StringUtils;
import org.apache.geode.distributed.internal.DistributionConfig;
+import
org.apache.geode.distributed.internal.membership.gms.messenger.MembershipInformation;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.admin.SSLConfig;
import org.apache.geode.internal.admin.TransportConfig;
@@ -54,7 +55,7 @@ public class RemoteTransportConfig implements TransportConfig
{
private final String membershipPortRange;
private int tcpPort;
private boolean isReconnectingDS;
- private Object oldDSMembershipInfo;
+ private MembershipInformation oldDSMembershipInfo;
private int vmKind = -1;
// -------------------------------------------------------------------------
@@ -234,11 +235,11 @@ public class RemoteTransportConfig implements
TransportConfig {
this.isReconnectingDS = isReconnectingDS;
}
- public Object getOldDSMembershipInfo() {
+ public MembershipInformation getOldDSMembershipInfo() {
return oldDSMembershipInfo;
}
- public void setOldDSMembershipInfo(Object oldDSMembershipInfo) {
+ public void setOldDSMembershipInfo(MembershipInformation
oldDSMembershipInfo) {
this.oldDSMembershipInfo = oldDSMembershipInfo;
}
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);
