Skip to content

Commit 450a41a

Browse files
belabanbelaban
committed
Backport commits for JGRP-3033 from master
1 parent ea17870 commit 450a41a

4 files changed

Lines changed: 283 additions & 6 deletions

File tree

‎src/org/jgroups/protocols/pbcast/JoinRsp.java‎

Lines changed: 16 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44

55
import org.jgroups.Constructable;
66
import org.jgroups.Global;
7+
import org.jgroups.MergeView;
78
import org.jgroups.View;
89
import org.jgroups.util.Digest;
910
import org.jgroups.util.SizeStreamable;
@@ -25,6 +26,7 @@ public class JoinRsp implements SizeStreamable, Constructable<JoinRsp> {
2526
protected static final byte VIEW_PRESENT = 1 << 0;
2627
protected static final byte DIGEST_PRESENT = 1 << 1;
2728
protected static final byte FAIL_REASON_PRESENT = 1 << 2;
29+
protected static final byte MERGE_VIEW_PRESENT = 1 << 3;
2830

2931

3032
public JoinRsp() {
@@ -49,8 +51,12 @@ public Supplier<? extends JoinRsp> create() {
4951
@Override
5052
public void writeTo(DataOutput out) throws IOException {
5153
byte flags=0;
52-
if(view != null)
53-
flags|=VIEW_PRESENT;
54+
if(view != null) {
55+
if(view instanceof MergeView)
56+
flags|=MERGE_VIEW_PRESENT;
57+
else
58+
flags|=VIEW_PRESENT;
59+
}
5460
if(digest != null)
5561
flags|=DIGEST_PRESENT;
5662
if(fail_reason != null)
@@ -75,10 +81,16 @@ public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
7581
byte flags=in.readByte();
7682

7783
// 1. view
78-
if((flags & VIEW_PRESENT) == VIEW_PRESENT) {
79-
view=new View();
84+
if((flags & MERGE_VIEW_PRESENT) == MERGE_VIEW_PRESENT) {
85+
view=new MergeView();
8086
view.readFrom(in);
8187
}
88+
else {
89+
if((flags & VIEW_PRESENT) == VIEW_PRESENT) {
90+
view=new View();
91+
view.readFrom(in);
92+
}
93+
}
8294

8395
// 2. digest
8496
if((flags & DIGEST_PRESENT) == DIGEST_PRESENT) {

‎src/org/jgroups/util/Digest.java‎

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -168,8 +168,16 @@ public void readFrom(DataInput in, boolean read_addrs) throws IOException, Class
168168
members=Util.readAddresses(in);
169169
seqnos=new long[capacity() * 2];
170170
}
171-
else
172-
seqnos=new long[in.readShort() *2];
171+
else {
172+
int num_mbrs=in.readShort();
173+
// 'members' was set by the caller (e.g. to a view's membership, to avoid shipping it twice). If the
174+
// stream disagrees, it isn't the digest we expect, and reading on would create a Digest whose seqnos
175+
// don't match its members - failing much later with an ArrayIndexOutOfBoundsException on iteration
176+
if(members != null && num_mbrs != members.length)
177+
throw new IOException(String.format("digest in stream has %d members, but %d were expected",
178+
num_mbrs, members.length));
179+
seqnos=new long[num_mbrs * 2];
180+
}
173181

174182
for(int i=0; i < seqnos.length/2; i++)
175183
Bits.readLongSequence(in, seqnos, i*2);

‎tests/junit-functional/org/jgroups/tests/JoinRspMergeViewTest.java‎

Lines changed: 138 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,138 @@
1+
package org.jgroups.tests;
2+
3+
import org.jgroups.Address;
4+
import org.jgroups.Global;
5+
import org.jgroups.MergeView;
6+
import org.jgroups.View;
7+
import org.jgroups.ViewId;
8+
import org.jgroups.protocols.pbcast.JoinRsp;
9+
import org.jgroups.util.*;
10+
import org.testng.annotations.BeforeClass;
11+
import org.testng.annotations.Test;
12+
13+
import java.io.IOException;
14+
import java.util.Arrays;
15+
16+
import static org.testng.Assert.assertEquals;
17+
import static org.testng.Assert.assertTrue;
18+
19+
/**
20+
* Tests that a {@link JoinRsp} can round-trip a {@link MergeView}.
21+
* <p/>
22+
* {@code JoinRsp.writeTo()} calls {@code view.writeTo()} polymorphically, so a MergeView also writes
23+
* its subgroups. {@code JoinRsp.readFrom()} however always creates a plain {@code new View()}, which
24+
* doesn't read the subgroups back. The leftover subgroup bytes desync the stream, and the following
25+
* {@code digest.readFrom(in, false)} reads the *number of subgroups* as the digest's member count.
26+
* Depending on how that count compares to the view size, iterating the digest (as
27+
* {@code NAKACK2.setDigest()} does) either blows up with an ArrayIndexOutOfBoundsException or
28+
* silently yields garbage seqnos.
29+
* <p/>
30+
* A coordinator sends a MergeView in a JOIN-RSP whenever it gets a JOIN-REQ from a member that is
31+
* already in its view (CoordGmsImpl.handleMembershipChange() -> GMS.getViewAndDigest()) while its
32+
* current view happens to be a MergeView.
33+
*
34+
* @author Claude
35+
*/
36+
@Test(groups=Global.FUNCTIONAL)
37+
public class JoinRspMergeViewTest {
38+
protected Address a, b, c;
39+
40+
@BeforeClass
41+
protected void setup() {
42+
a=Util.createRandomAddress("A");
43+
b=Util.createRandomAddress("B");
44+
c=Util.createRandomAddress("C");
45+
}
46+
47+
/** Baseline: a regular view round-trips correctly */
48+
public void testRegularView() throws Exception {
49+
View view=View.create(a, 5, a, b, c);
50+
_testRoundTrip(view);
51+
}
52+
53+
/** 3 members, 2 subgroups: digest ends up with 2 seqno pairs but 3 members -> AIOOBE on iteration */
54+
public void testMergeViewWithFewerSubgroupsThanMembers() throws Exception {
55+
View view=new MergeView(new ViewId(a, 5), Arrays.asList(a, b, c),
56+
Arrays.asList(View.create(a, 1, a, b), View.create(c, 1, c)));
57+
_testRoundTrip(view);
58+
}
59+
60+
/** 2 members, 2 subgroups: no exception, but the seqnos are silently wrong */
61+
public void testMergeViewWithSameSubgroupsAsMembers() throws Exception {
62+
View view=new MergeView(new ViewId(a, 5), Arrays.asList(a, b),
63+
Arrays.asList(View.create(a, 1, a), View.create(b, 1, b)));
64+
_testRoundTrip(view);
65+
}
66+
67+
/** 2 members, 3 subgroups: digest ends up with more seqno pairs than members */
68+
public void testMergeViewWithMoreSubgroupsThanMembers() throws Exception {
69+
View view=new MergeView(new ViewId(a, 5), Arrays.asList(a, b),
70+
Arrays.asList(View.create(a, 1, a), View.create(b, 1, b),
71+
View.create(a, 2, a, b)));
72+
_testRoundTrip(view);
73+
}
74+
75+
/**
76+
* A JOIN-RSP from an unpatched coordinator (which writes the MergeView's subgroups) must be rejected
77+
* with an IOException, so that GMS.readJoinRsp() logs it and the joiner retries the JOIN, rather than
78+
* an ArrayIndexOutOfBoundsException propagating out of JChannel.connect().
79+
*/
80+
public void testJoinRspFromUnpatchedCoord() throws Exception {
81+
View view=new MergeView(new ViewId(a, 5), Arrays.asList(a, b, c),
82+
Arrays.asList(View.create(a, 1, a, b), View.create(c, 1, c)));
83+
ByteArray buf=marshalLegacy(view, createDigest(view));
84+
try {
85+
JoinRsp rsp=Util.streamableFromBuffer(JoinRsp::new, buf.array(), buf.getOffset(), buf.getLength());
86+
for(Digest.Entry ignored: rsp.getDigest()) // what NAKACK2.setDigest() does
87+
;
88+
throw new AssertionError("should have thrown an IOException, but got " + rsp);
89+
}
90+
catch(IOException ex) {
91+
System.out.printf("received expected exception: %s\n", ex);
92+
}
93+
}
94+
95+
/** Marshals a JoinRsp the way 4.2.30 and earlier do: the view is written polymorphically, subgroups and all */
96+
protected static ByteArray marshalLegacy(View view, Digest digest) throws Exception {
97+
ByteArrayDataOutputStream out=new ByteArrayDataOutputStream(512);
98+
out.writeByte(1); // Util.writeStreamable(): non-null marker
99+
out.writeByte(1 | 2); // JoinRsp: VIEW_PRESENT | DIGEST_PRESENT
100+
view.writeTo(out);
101+
digest.writeTo(out, false);
102+
return out.getBuffer();
103+
}
104+
105+
/**
106+
* Marshals a JoinRsp exactly as GMS.marshal(JoinRsp)/GMS.readJoinRsp() do, then asserts that the
107+
* digest survived the round trip and can be iterated (as NAKACK2.setDigest() does).
108+
*/
109+
protected static void _testRoundTrip(View view) throws Exception {
110+
Digest digest=createDigest(view);
111+
assertEquals(digest.capacity(), view.size());
112+
113+
ByteArray buf=Util.streamableToBuffer(new JoinRsp(view, digest));
114+
JoinRsp rsp=Util.streamableFromBuffer(JoinRsp::new, buf.array(), buf.getOffset(), buf.getLength());
115+
116+
View rv=rsp.getView();
117+
assert view.equals(rv) : String.format("view changed: view=%s, deserialized view=%s", view, rv);
118+
119+
Digest new_digest=rsp.getDigest();
120+
int count=0;
121+
for(Digest.Entry ignored: new_digest) // this is what NAKACK2.setDigest() does
122+
count++;
123+
assertEquals(count, view.size(), "digest has a different number of entries than the view has members");
124+
assert new_digest.equals(digest) :
125+
String.format("digest changed: expected %s, but got %s", digest, new_digest);
126+
}
127+
128+
/** Creates a digest matching the view, as CoordGmsImpl does before sending a JOIN-RSP */
129+
protected static Digest createDigest(View view) {
130+
MutableDigest digest=new MutableDigest(view.getMembersRaw());
131+
long seqno=10;
132+
for(Address mbr: view.getMembersRaw())
133+
digest.set(mbr, seqno, seqno+=10);
134+
assertTrue(digest.allSet());
135+
return digest;
136+
}
137+
}
138+

‎tests/junit-functional/org/jgroups/tests/JoinRspMergeViewTest2.java‎

Lines changed: 119 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,119 @@
1+
package org.jgroups.tests;
2+
3+
import org.jgroups.*;
4+
import org.jgroups.conf.ClassConfigurator;
5+
import org.jgroups.protocols.pbcast.GMS;
6+
import org.jgroups.stack.Protocol;
7+
import org.jgroups.stack.ProtocolStack;
8+
import org.jgroups.util.MessageBatch;
9+
import org.jgroups.util.Util;
10+
import org.testng.annotations.AfterMethod;
11+
import org.testng.annotations.BeforeMethod;
12+
import org.testng.annotations.Test;
13+
14+
import java.util.Iterator;
15+
import java.util.List;
16+
import java.util.stream.Collectors;
17+
import java.util.stream.Stream;
18+
19+
/**
20+
* Tests the case where we have 2 members A,B with a MergeView. Member C joins but misses the first JoinRsp. The second
21+
* JoinRsp should lead to a deserialization issue, see https://redhat.atlassian.net/browse/JGRP-3033
22+
* @author Bela Ban
23+
* @since 5.6.0
24+
*/
25+
@Test(groups=Global.FUNCTIONAL)
26+
public class JoinRspMergeViewTest2 {
27+
protected JChannel a,b,c;
28+
protected static short GMS_ID=ClassConfigurator.getProtocolId(GMS.class);
29+
30+
@BeforeMethod
31+
protected void init() throws Exception {
32+
a=new JChannel(Util.getTestStack()).name("A").connect(JoinRspMergeViewTest2.class.getSimpleName());
33+
b=new JChannel(Util.getTestStack()).name("B").connect(JoinRspMergeViewTest2.class.getSimpleName());
34+
c=new JChannel(Util.getTestStack()).name("C");
35+
Util.waitUntilAllChannelsHaveSameView(3000, 100, a,b);
36+
}
37+
38+
@AfterMethod
39+
protected void destroy() {
40+
Util.close(c,b,a);
41+
}
42+
43+
/**
44+
* Cluster A and B. C joins, but discards its JoinRsps. A MergeView is installed in A and B. Now, C doesn't drop the
45+
* JoinRsps and gets a JoinRsp with a MergeView, which should result in an exception with the unpatched code.
46+
*/
47+
public void testIncorrectDeserialization() throws Exception {
48+
DropJoinResponses drop=new DropJoinResponses().enable(true);
49+
GMS gms_c=c.stack().findProtocol(GMS.class);
50+
gms_c.setMaxJoinAttempts(100).setJoinTimeout(2000);
51+
c.stack().insertProtocol(drop, ProtocolStack.Position.BELOW, GMS.class);
52+
53+
Runnable r=() -> {
54+
injectMergeView(); // injects MV={A,B,C} into A and B
55+
drop.enable(false);
56+
};
57+
new Thread(r).start();
58+
59+
c.connect(JoinRspMergeViewTest2.class.getSimpleName()); // fails without patch for JGRP-3033
60+
Util.waitUntilAllChannelsHaveSameView(10000, 500, a,b,c);
61+
System.out.printf("\nviews:\n%s\n", Stream.of(a,b,c).map(ch -> String.format("%s -> %s", ch.address(), ch.view()))
62+
.collect(Collectors.joining("\n")));
63+
64+
65+
}
66+
67+
// Injects a MergeView {A,B,C} into A and B (*not* C!)
68+
protected void injectMergeView() {
69+
// Inject a MergeView
70+
View v1=View.create(a.address(), 5, a.address());
71+
View v2=View.create(b.address(), 6, b.address());
72+
View v3=View.create(c.address(), 3, c.address());
73+
MergeView mv=new MergeView(a.address(), 7, List.of(a.address(), b.address(), c.address()), List.of(v1,v2,v3));
74+
Stream.of(a,b).forEach(ch -> {
75+
GMS gms=ch.stack().findProtocol(GMS.class);
76+
gms.installView(mv);
77+
});
78+
System.out.printf("\nviews:\n%s\n", Stream.of(a,b).map(ch -> String.format("%s -> %s", ch.address(), ch.view()))
79+
.collect(Collectors.joining("\n")));
80+
}
81+
82+
// used by C, drops received JoinRsps until disabled
83+
protected static class DropJoinResponses extends Protocol {
84+
protected boolean enabled=true;
85+
86+
protected DropJoinResponses enable(boolean f) {this.enabled=f; return this;}
87+
88+
@Override
89+
public void up(MessageBatch batch) {
90+
if(enabled) {
91+
for(Iterator<Message> it=batch.iterator(); it.hasNext();) {
92+
Message msg=it.next();
93+
GMS.GmsHeader hdr=msg.getHeader(GMS_ID);
94+
if(hdr != null) {
95+
if(hdr.getType() == GMS.GmsHeader.JOIN_RSP) {
96+
System.out.printf("-- dropped JOIN-RSP from %s -> %s: %s\n", msg.src(), msg.dest(), msg.printHeaders());
97+
it.remove(); // drop JoinRsp
98+
}
99+
}
100+
}
101+
}
102+
up_prot.up(batch);
103+
}
104+
105+
@Override
106+
public Object up(Message msg) {
107+
if(enabled) {
108+
GMS.GmsHeader hdr=msg.getHeader(GMS_ID);
109+
if(hdr != null) {
110+
if(hdr.getType() == GMS.GmsHeader.JOIN_RSP) {
111+
System.out.printf("-- dropped JOIN-RSP from %s -> %s: %s\n", msg.src(), msg.dest(), msg.printHeaders());
112+
return null; // drop the first JOIN-RSP from A -> C
113+
}
114+
}
115+
}
116+
return up_prot.up(msg);
117+
}
118+
}
119+
}

0 commit comments

Comments
 (0)