rdev/gossip-revisited
1
1
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Fixed build

Browse Source
This commit is contained in:
Jaime Freire
2023-11-25 08:18:02 +01:00
parent db9bb4ccdf
commit 789a5f6abc
141 changed files with 3259 additions and 2957 deletions
Download Patch File Download Diff File Expand all files Collapse all files

101
gossip-itest/pom.xml
View File

@ -16,73 +16,44 @@
See the License for the specific language governing permissions and
limitations under the License.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.apache.gossip</groupId>
<artifactId>gossip-parent</artifactId>
<parent>
<groupId>org.apache.gossip</groupId>
<artifactId>gossip-parent</artifactId>
<version>0.1.3-incubating-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
<name>Gossip Itest</name>
<artifactId>gossip-itest</artifactId>
<version>0.1.3-incubating-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
<name>Gossip itest</name>
<artifactId>gossip-itest</artifactId>
<version>0.1.3-incubating-SNAPSHOT</version>
<dependencies>
<dependency>
<groupId>org.apache.gossip</groupId>
<artifactId>gossip-base</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.gossip</groupId>
<artifactId>gossip-base</artifactId>
<version>${project.version}</version>
<type>test-jar</type>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.gossip</groupId>
<artifactId>gossip-protocol-jackson</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.gossip</groupId>
<artifactId>gossip-transport-udp</artifactId>
<version>${project.version}</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.1</version>
<configuration>
<source>${java.version}</source>
<target>${java.version}</target>
</configuration>
</plugin>
<plugin>
<artifactId>maven-surefire-plugin</artifactId>
<version>2.19.1</version>
<configuration>
<systemPropertyVariables>
<java.io.tmpdir>${project.build.directory}</java.io.tmpdir>
</systemPropertyVariables>
</configuration>
<dependencies>
<dependency>
<groupId>org.junit.platform</groupId>
<artifactId>junit-platform-surefire-provider</artifactId>
<version>${junit.platform.version}</version>
</dependency>
</dependencies>
</plugin>
</plugins>
</build>
<dependencies>
<dependency>
<groupId>org.apache.gossip</groupId>
<artifactId>gossip-base</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.gossip</groupId>
<artifactId>gossip-base</artifactId>
<version>${project.version}</version>
<type>test-jar</type>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.gossip</groupId>
<artifactId>gossip-protocol-jackson</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.gossip</groupId>
<artifactId>gossip-transport-udp</artifactId>
<version>${project.version}</version>
</dependency>
</dependencies>
</project>

28
gossip-itest/src/test/java/org/apache/gossip/DataTest.java
View File

@ -18,6 +18,18 @@
package org.apache.gossip;
import io.teknek.tunit.TUnit;
import java.net.URI;
import java.net.URISyntaxException;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.function.Function;
import org.apache.gossip.crdt.CrdtAddRemoveSet;
import org.apache.gossip.crdt.GrowOnlyCounter;
import org.apache.gossip.crdt.GrowOnlySet;
@ -34,25 +46,11 @@ import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import java.net.URI;
import java.net.URISyntaxException;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.function.Function;
public class DataTest {
private static final List<GossipManager> clients = new ArrayList<>();
private final String gCounterKey = "crdtgc";
private final String pnCounterKey = "crdtpn";
private static final List<GossipManager> clients = new ArrayList<>();
@BeforeClass
public static void initializeMembers() throws InterruptedException, UnknownHostException, URISyntaxException{
final int clusterMembers = 2;

71
gossip-itest/src/test/java/org/apache/gossip/IdAndPropertyTest.java
View File

@ -25,22 +25,20 @@ import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeUnit;
import org.apache.gossip.manager.DatacenterRackAwareActiveGossiper;
import org.apache.gossip.manager.GossipManager;
import org.apache.gossip.manager.GossipManagerBuilder;
import org.junit.jupiter.api.Test;
import org.junit.platform.runner.JUnitPlatform;
import org.junit.runner.RunWith;
import io.teknek.tunit.TUnit;
@RunWith(JUnitPlatform.class)
public class IdAndPropertyTest extends AbstractIntegrationBase {
@Test
public void testDatacenterRackGossiper() throws URISyntaxException, UnknownHostException, InterruptedException {
public void testDatacenterRackGossiper()
throws URISyntaxException, UnknownHostException, InterruptedException {
GossipSettings settings = new GossipSettings();
settings.setActiveGossipClass(DatacenterRackAwareActiveGossiper.class.getName());
List<Member> startupMembers = new ArrayList<>();
@ -48,44 +46,59 @@ public class IdAndPropertyTest extends AbstractIntegrationBase {
x.put("a", "b");
x.put("datacenter", "dc1");
x.put("rack", "rack1");
GossipManager gossipService1 = GossipManagerBuilder.newBuilder()
GossipManager gossipService1 =
GossipManagerBuilder.newBuilder()
.cluster("a")
.uri(new URI("udp://" + "127.0.0.1" + ":" + (29000 + 0)))
.id("0")
.properties(x)
.gossipMembers(startupMembers)
.gossipSettings(settings).build();
.gossipSettings(settings)
.build();
gossipService1.init();
register(gossipService1);
Map<String, String> y = new HashMap<>();
y.put("a", "c");
y.put("datacenter", "dc2");
y.put("rack", "rack2");
GossipManager gossipService2 = GossipManagerBuilder.newBuilder().cluster("a")
.uri( new URI("udp://" + "127.0.0.1" + ":" + (29000 + 10)))
GossipManager gossipService2 =
GossipManagerBuilder.newBuilder()
.cluster("a")
.uri(new URI("udp://" + "127.0.0.1" + ":" + (29000 + 10)))
.id("1")
.properties(y)
.gossipMembers(Arrays.asList(new RemoteMember("a",
new URI("udp://" + "127.0.0.1" + ":" + (29000 + 0)), "0")))
.gossipSettings(settings).build();
.gossipMembers(
Arrays.asList(
new RemoteMember(
"a", new URI("udp://" + "127.0.0.1" + ":" + (29000 + 0)), "0")))
.gossipSettings(settings)
.build();
gossipService2.init();
register(gossipService2);
TUnit.assertThat(() -> {
String value = "";
try {
value = gossipService1.getLiveMembers().get(0).getProperties().get("a");
} catch (RuntimeException e){ }
return value;
}).afterWaitingAtMost(10, TimeUnit.SECONDS).isEqualTo("c");
TUnit.assertThat(() -> {
String value = "";
try {
value = gossipService2.getLiveMembers().get(0).getProperties().get("a");
} catch (RuntimeException e){ }
return value;
}).afterWaitingAtMost(10, TimeUnit.SECONDS).isEqualTo("b");
TUnit.assertThat(
() -> {
String value = "";
try {
value = gossipService1.getLiveMembers().get(0).getProperties().get("a");
} catch (RuntimeException e) {
}
return value;
})
.afterWaitingAtMost(10, TimeUnit.SECONDS)
.isEqualTo("c");
TUnit.assertThat(
() -> {
String value = "";
try {
value = gossipService2.getLiveMembers().get(0).getProperties().get("a");
} catch (RuntimeException e) {
}
return value;
})
.afterWaitingAtMost(10, TimeUnit.SECONDS)
.isEqualTo("b");
}
}

70
gossip-itest/src/test/java/org/apache/gossip/PerNodeDataEventTest.java
View File

@ -35,7 +35,7 @@ import java.util.concurrent.TimeUnit;
@RunWith(Parameterized.class)
public class PerNodeDataEventTest extends AbstractIntegrationBase {
private String receivedKey = "";
private String receivingNodeId = "";
private Object receivingNodeDataNewValue = "";
@ -51,14 +51,12 @@ public class PerNodeDataEventTest extends AbstractIntegrationBase {
@Parameterized.Parameters(name = "{index} bulkTransfer={1}")
public static Collection<Object[]> data() {
return Arrays.asList(new Object[][]{
{50000, false}, {55000, true}
});
return Arrays.asList(new Object[][] {{50000, false}, {55000, true}});
}
@Test
public void perNodeDataEventTest()
throws InterruptedException, UnknownHostException, URISyntaxException {
throws InterruptedException, UnknownHostException, URISyntaxException {
GossipSettings settings = new GossipSettings();
settings.setPersistRingState(false);
settings.setPersistDataState(false);
@ -74,42 +72,54 @@ public class PerNodeDataEventTest extends AbstractIntegrationBase {
final int clusterMembers = 2;
for (int i = 1; i < clusterMembers + 1; ++i) {
URI uri = new URI("udp://" + "127.0.0.1" + ":" + (base + i));
GossipManager gossipService = GossipManagerBuilder.newBuilder().cluster(cluster).uri(uri)
.id(i + "").gossipMembers(startupMembers).gossipSettings(settings).build();
GossipManager gossipService =
GossipManagerBuilder.newBuilder()
.cluster(cluster)
.uri(uri)
.id(i + "")
.gossipMembers(startupMembers)
.gossipSettings(settings)
.build();
clients.add(gossipService);
gossipService.init();
register(gossipService);
}
// check whether the members are discovered
TUnit.assertThat(() -> {
int total = 0;
for (int i = 0; i < clusterMembers; ++i) {
total += clients.get(i).getLiveMembers().size();
}
return total;
}).afterWaitingAtMost(20, TimeUnit.SECONDS).isEqualTo(2);
TUnit.assertThat(
() -> {
int total = 0;
for (int i = 0; i < clusterMembers; ++i) {
total += clients.get(i).getLiveMembers().size();
}
return total;
})
.afterWaitingAtMost(20, TimeUnit.SECONDS)
.isEqualTo(2);
// Adding new data to Node 1
clients.get(0).gossipPerNodeData(getPerNodeData("category", "distributed"));
// Node 2 is interested in data changes for the key "organization" and "category"
clients.get(1).registerPerNodeDataSubscriber((nodeId, key, oldValue, newValue) -> {
if (!key.equals("organization") && !key.equals("category")) return;
receivingNodeId = nodeId;
receivedKey = key;
receivingNodeDataOldValue = oldValue;
receivingNodeDataNewValue = newValue;
lock.release();
});
clients
.get(1)
.registerPerNodeDataSubscriber(
(nodeId, key, oldValue, newValue) -> {
if (!key.equals("organization") && !key.equals("category")) return;
receivingNodeId = nodeId;
receivedKey = key;
receivingNodeDataOldValue = oldValue;
receivingNodeDataNewValue = newValue;
lock.release();
});
// Node 2 first time adds Node 1 data
lock.tryAcquire(10, TimeUnit.SECONDS);
Assert.assertEquals("1", receivingNodeId);
Assert.assertEquals("category", receivedKey);
Assert.assertEquals(null, receivingNodeDataOldValue);
Assert.assertEquals("distributed", receivingNodeDataNewValue);
// Node 1 adds new per node data
clients.get(0).gossipPerNodeData(getPerNodeData("organization", "apache"));
// Node 2 adds new data key from Node 1
@ -118,7 +128,7 @@ public class PerNodeDataEventTest extends AbstractIntegrationBase {
Assert.assertEquals("organization", receivedKey);
Assert.assertEquals(null, receivingNodeDataOldValue);
Assert.assertEquals("apache", receivingNodeDataNewValue);
// Node 1 updates its value
clients.get(0).gossipPerNodeData(getPerNodeData("organization", "apache-gossip"));
// Node 2 updates existing value
@ -127,9 +137,8 @@ public class PerNodeDataEventTest extends AbstractIntegrationBase {
Assert.assertEquals("organization", receivedKey);
Assert.assertEquals("apache", receivingNodeDataOldValue);
Assert.assertEquals("apache-gossip", receivingNodeDataNewValue);
}
private PerNodeDataMessage getPerNodeData(String key, String value) {
PerNodeDataMessage g = new PerNodeDataMessage();
g.setExpireAt(Long.MAX_VALUE);
@ -138,5 +147,4 @@ public class PerNodeDataEventTest extends AbstractIntegrationBase {
g.setTimestamp(System.currentTimeMillis());
return g;
}
}

236
gossip-itest/src/test/java/org/apache/gossip/PerNodeDataReplicationControlTest.java
View File

@ -22,12 +22,8 @@ import org.apache.gossip.manager.DatacenterRackAwareActiveGossiper;
import org.apache.gossip.manager.GossipManager;
import org.apache.gossip.manager.GossipManagerBuilder;
import org.apache.gossip.model.PerNodeDataMessage;
import org.apache.gossip.model.SharedDataMessage;
import org.apache.gossip.replication.*;
import org.junit.Assert;
import org.junit.Test;
import org.junit.platform.runner.JUnitPlatform;
import org.junit.runner.RunWith;
import java.net.URI;
import java.net.URISyntaxException;
@ -35,87 +31,105 @@ import java.net.UnknownHostException;
import java.util.*;
import java.util.concurrent.TimeUnit;
@RunWith(JUnitPlatform.class)
public class PerNodeDataReplicationControlTest extends AbstractIntegrationBase {
@Test
public void perNodeDataReplicationTest()
throws InterruptedException, UnknownHostException, URISyntaxException {
public void perNodeDataReplicationTest() throws URISyntaxException {
generateStandardNodes(3);
// check whether the members are discovered
TUnit.assertThat(() -> {
int total = 0;
for (GossipManager node : nodes) {
total += node.getLiveMembers().size();
}
return total;
}).afterWaitingAtMost(20, TimeUnit.SECONDS).isEqualTo(2);
TUnit.assertThat(
() -> {
int total = 0;
for (GossipManager node : nodes) {
total += node.getLiveMembers().size();
}
return total;
})
.afterWaitingAtMost(20, TimeUnit.SECONDS)
.isEqualTo(2);
// Adding new per node data to Node 1 with default replication (replicate all)
nodes.get(0).gossipPerNodeData(getPerNodeData("public", "I am visible to all",
new AllReplicable<>()));
nodes
.get(0)
.gossipPerNodeData(getPerNodeData("public", "I am visible to all", new AllReplicable<>()));
// Adding new per node data to Node 1 with no replication (replicate none)
nodes.get(0).gossipPerNodeData(getPerNodeData("private", "I am private",
new NotReplicable<>()));
nodes
.get(0)
.gossipPerNodeData(getPerNodeData("private", "I am private", new NotReplicable<>()));
List<LocalMember> whiteList = new ArrayList<>();
whiteList.add(nodes.get(1).getMyself());
// Adding new per node data to Node 1 with white list Node 2
nodes.get(0).gossipPerNodeData(getPerNodeData("wl", "white list",
new WhiteListReplicable<>(whiteList)));
nodes
.get(0)
.gossipPerNodeData(
getPerNodeData("wl", "white list", new WhiteListReplicable<>(whiteList)));
List<LocalMember> blackList = new ArrayList<>();
blackList.add(nodes.get(1).getMyself());
// Adding new per node data to Node 1 with black list Node 2
nodes.get(0).gossipPerNodeData(getPerNodeData("bl", "black list",
new BlackListReplicable<>(blackList)));
nodes
.get(0)
.gossipPerNodeData(
getPerNodeData("bl", "black list", new BlackListReplicable<>(blackList)));
// Node 2 and 3 must have the shared data with key 'public'
TUnit.assertThat(() -> {
PerNodeDataMessage message = nodes.get(1).findPerNodeGossipData("1", "public");
if(message == null){
return "";
}else {
return message.getPayload();
}
}).afterWaitingAtMost(20, TimeUnit.SECONDS).isEqualTo("I am visible to all");
TUnit.assertThat(
() -> {
PerNodeDataMessage message = nodes.get(1).findPerNodeGossipData("1", "public");
if (message == null) {
return "";
} else {
return message.getPayload();
}
})
.afterWaitingAtMost(20, TimeUnit.SECONDS)
.isEqualTo("I am visible to all");
TUnit.assertThat(() -> {
PerNodeDataMessage message = nodes.get(2).findPerNodeGossipData("1", "public");
if(message == null){
return "";
}else {
return message.getPayload();
}
}).afterWaitingAtMost(20, TimeUnit.SECONDS).isEqualTo("I am visible to all");
TUnit.assertThat(
() -> {
PerNodeDataMessage message = nodes.get(2).findPerNodeGossipData("1", "public");
if (message == null) {
return "";
} else {
return message.getPayload();
}
})
.afterWaitingAtMost(20, TimeUnit.SECONDS)
.isEqualTo("I am visible to all");
// Node 2 must have shared data with key wl
TUnit.assertThat(() -> {
PerNodeDataMessage message = nodes.get(1).findPerNodeGossipData("1", "wl");
if(message == null){
return "";
}else {
return message.getPayload();
}
}).afterWaitingAtMost(20, TimeUnit.SECONDS).isEqualTo("white list");
TUnit.assertThat(
() -> {
PerNodeDataMessage message = nodes.get(1).findPerNodeGossipData("1", "wl");
if (message == null) {
return "";
} else {
return message.getPayload();
}
})
.afterWaitingAtMost(20, TimeUnit.SECONDS)
.isEqualTo("white list");
// Node 3 must have shared data with key bl
TUnit.assertThat(() -> {
PerNodeDataMessage message = nodes.get(2).findPerNodeGossipData("1", "bl");
if(message == null){
return "";
}else {
return message.getPayload();
}
}).afterWaitingAtMost(20, TimeUnit.SECONDS).isEqualTo("black list");
TUnit.assertThat(
() -> {
PerNodeDataMessage message = nodes.get(2).findPerNodeGossipData("1", "bl");
if (message == null) {
return "";
} else {
return message.getPayload();
}
})
.afterWaitingAtMost(20, TimeUnit.SECONDS)
.isEqualTo("black list");
}
@Test
public void perNodeDataDcReplicationTest()
throws InterruptedException, UnknownHostException, URISyntaxException {
throws InterruptedException, UnknownHostException, URISyntaxException {
GossipSettings settings = new GossipSettings();
settings.setPersistRingState(false);
@ -131,57 +145,68 @@ public class PerNodeDataReplicationControlTest extends AbstractIntegrationBase {
RemoteMember seeder = new RemoteMember(cluster, URI.create("udp://127.0.0.1:5001"), "1");
// initialize 2 data centers with each having two racks
createDcNode(URI.create("udp://127.0.0.1:5001"), "1", settings, seeder, cluster,
"DataCenter1", "Rack1");
createDcNode(URI.create("udp://127.0.0.1:5002"), "2", settings, seeder, cluster,
"DataCenter1", "Rack2");
createDcNode(
URI.create("udp://127.0.0.1:5001"), "1", settings, seeder, cluster, "DataCenter1", "Rack1");
createDcNode(
URI.create("udp://127.0.0.1:5002"), "2", settings, seeder, cluster, "DataCenter1", "Rack2");
createDcNode(URI.create("udp://127.0.0.1:5006"), "6", settings, seeder, cluster,
"DataCenter2", "Rack1");
createDcNode(URI.create("udp://127.0.0.1:5007"), "7", settings, seeder, cluster,
"DataCenter2", "Rack1");
createDcNode(
URI.create("udp://127.0.0.1:5006"), "6", settings, seeder, cluster, "DataCenter2", "Rack1");
createDcNode(
URI.create("udp://127.0.0.1:5007"), "7", settings, seeder, cluster, "DataCenter2", "Rack1");
// check whether the members are discovered
TUnit.assertThat(() -> {
int total = 0;
for (int i = 0; i < 4; ++i) {
total += nodes.get(i).getLiveMembers().size();
}
return total;
}).afterWaitingAtMost(20, TimeUnit.SECONDS).isEqualTo(12);
TUnit.assertThat(
() -> {
int total = 0;
for (int i = 0; i < 4; ++i) {
total += nodes.get(i).getLiveMembers().size();
}
return total;
})
.afterWaitingAtMost(20, TimeUnit.SECONDS)
.isEqualTo(12);
// Node 1 has a shared key with 'Dc1Rack1'
nodes.get(0).gossipPerNodeData(getPerNodeData("Dc1Rack1", "I am belong to Dc1",
new DataCenterReplicable<>()));
nodes
.get(0)
.gossipPerNodeData(
getPerNodeData("Dc1Rack1", "I am belong to Dc1", new DataCenterReplicable<>()));
// Node 6 has a shared key with 'Dc2Rack1'
nodes.get(2).gossipPerNodeData(getPerNodeData("Dc2Rack1", "I am belong to Dc2",
new DataCenterReplicable<>()));
nodes
.get(2)
.gossipPerNodeData(
getPerNodeData("Dc2Rack1", "I am belong to Dc2", new DataCenterReplicable<>()));
// Node 2 must have the shared data with key 'Dc1Rack1'
TUnit.assertThat(() -> {
PerNodeDataMessage message = nodes.get(1).findPerNodeGossipData("1", "Dc1Rack1");
if(message == null){
return "";
}else {
return message.getPayload();
}
}).afterWaitingAtMost(20, TimeUnit.SECONDS).isEqualTo("I am belong to Dc1");
TUnit.assertThat(
() -> {
PerNodeDataMessage message = nodes.get(1).findPerNodeGossipData("1", "Dc1Rack1");
if (message == null) {
return "";
} else {
return message.getPayload();
}
})
.afterWaitingAtMost(20, TimeUnit.SECONDS)
.isEqualTo("I am belong to Dc1");
// Node 7 must have the shared data with key 'Dc2Rack1'
TUnit.assertThat(() -> {
PerNodeDataMessage message = nodes.get(3).findPerNodeGossipData("6", "Dc2Rack1");
if(message == null){
return "";
}else {
return message.getPayload();
}
}).afterWaitingAtMost(20, TimeUnit.SECONDS).isEqualTo("I am belong to Dc2");
TUnit.assertThat(
() -> {
PerNodeDataMessage message = nodes.get(3).findPerNodeGossipData("6", "Dc2Rack1");
if (message == null) {
return "";
} else {
return message.getPayload();
}
})
.afterWaitingAtMost(20, TimeUnit.SECONDS)
.isEqualTo("I am belong to Dc2");
}
private PerNodeDataMessage getPerNodeData(String key, String value,
Replicable<PerNodeDataMessage> replicable) {
private PerNodeDataMessage getPerNodeData(
String key, String value, Replicable<PerNodeDataMessage> replicable) {
PerNodeDataMessage g = new PerNodeDataMessage();
g.setExpireAt(Long.MAX_VALUE);
g.setKey(key);
@ -191,17 +216,28 @@ public class PerNodeDataReplicationControlTest extends AbstractIntegrationBase {
return g;
}
private void createDcNode(URI uri, String id, GossipSettings settings, RemoteMember seeder,
String cluster, String dataCenter, String rack){
private void createDcNode(
URI uri,
String id,
GossipSettings settings,
RemoteMember seeder,
String cluster,
String dataCenter,
String rack) {
Map<String, String> props = new HashMap<>();
props.put(DatacenterRackAwareActiveGossiper.DATACENTER, dataCenter);
props.put(DatacenterRackAwareActiveGossiper.RACK, rack);
GossipManager dcNode = GossipManagerBuilder.newBuilder().cluster(cluster).uri(uri).id(id)
.gossipSettings(settings).gossipMembers(Arrays.asList(seeder)).properties(props)
GossipManager dcNode =
GossipManagerBuilder.newBuilder()
.cluster(cluster)
.uri(uri)
.id(id)
.gossipSettings(settings)
.gossipMembers(Arrays.asList(seeder))
.properties(props)
.build();
dcNode.init();
register(dcNode);
}
}

13
gossip-itest/src/test/java/org/apache/gossip/SharedDataEventTest.java
View File

@ -18,6 +18,12 @@
package org.apache.gossip;
import io.teknek.tunit.TUnit;
import java.net.URI;
import java.net.URISyntaxException;
import java.net.UnknownHostException;
import java.util.*;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import org.apache.gossip.crdt.GrowOnlyCounter;
import org.apache.gossip.manager.GossipManager;
import org.apache.gossip.manager.GossipManagerBuilder;
@ -27,13 +33,6 @@ import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import java.net.URI;
import java.net.URISyntaxException;
import java.net.UnknownHostException;
import java.util.*;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
@RunWith(Parameterized.class)
public class SharedDataEventTest extends AbstractIntegrationBase {

71
gossip-itest/src/test/java/org/apache/gossip/SharedDataLockTest.java
View File

@ -23,8 +23,6 @@ import org.apache.gossip.manager.GossipManagerBuilder;
import org.apache.gossip.model.SharedDataMessage;
import org.junit.Assert;
import org.junit.Test;
import org.junit.platform.runner.JUnitPlatform;
import org.junit.runner.RunWith;
import java.net.URI;
import java.net.URISyntaxException;
@ -34,12 +32,11 @@ import java.util.List;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;
@RunWith(JUnitPlatform.class)
public class SharedDataLockTest extends AbstractIntegrationBase {
@Test
public void sharedDataLockRandomVoteTest()
throws InterruptedException, UnknownHostException, URISyntaxException {
throws InterruptedException, UnknownHostException, URISyntaxException {
GossipSettings settings = new GossipSettings();
settings.setPersistRingState(false);
settings.setPersistDataState(false);
@ -54,8 +51,14 @@ public class SharedDataLockTest extends AbstractIntegrationBase {
final int clusterMembers = 10;
for (int i = 1; i < clusterMembers + 1; ++i) {
URI uri = new URI("udp://" + "127.0.0.1" + ":" + (50000 + i));
GossipManager gossipService = GossipManagerBuilder.newBuilder().cluster(cluster).uri(uri)
.id(i + "").gossipMembers(startupMembers).gossipSettings(settings).build();
GossipManager gossipService =
GossipManagerBuilder.newBuilder()
.cluster(cluster)
.uri(uri)
.id(i + "")
.gossipMembers(startupMembers)
.gossipSettings(settings)
.build();
clients.add(gossipService);
gossipService.getLockManager().setNumberOfNodes(clusterMembers);
gossipService.init();
@ -69,34 +72,40 @@ public class SharedDataLockTest extends AbstractIntegrationBase {
final AtomicInteger lockFailedCount = new AtomicInteger(0);
// Node 1 try to lock on key category
Thread Node1LockingThread = new Thread(() -> {
try {
clients.get(0).acquireSharedDataLock("category");
lockSuccessCount.incrementAndGet();
} catch (VoteFailedException ignore) {
lockFailedCount.incrementAndGet();
}
});
Thread Node1LockingThread =
new Thread(
() -> {
try {
clients.get(0).acquireSharedDataLock("category");
lockSuccessCount.incrementAndGet();
} catch (VoteFailedException ignore) {
lockFailedCount.incrementAndGet();
}
});
// Node 3 try to lock on key category
Thread Node3LockingThread = new Thread(() -> {
try {
clients.get(2).acquireSharedDataLock("category");
lockSuccessCount.incrementAndGet();
} catch (VoteFailedException ignore) {
lockFailedCount.incrementAndGet();
}
});
Thread Node3LockingThread =
new Thread(
() -> {
try {
clients.get(2).acquireSharedDataLock("category");
lockSuccessCount.incrementAndGet();
} catch (VoteFailedException ignore) {
lockFailedCount.incrementAndGet();
}
});
// Node 6 try to lock on key category
Thread Node5LockingThread = new Thread(() -> {
try {
clients.get(5).acquireSharedDataLock("category");
lockSuccessCount.incrementAndGet();
} catch (VoteFailedException ignore) {
lockFailedCount.incrementAndGet();
}
});
Thread Node5LockingThread =
new Thread(
() -> {
try {
clients.get(5).acquireSharedDataLock("category");
lockSuccessCount.incrementAndGet();
} catch (VoteFailedException ignore) {
lockFailedCount.incrementAndGet();
}
});
Node1LockingThread.start();
Node3LockingThread.start();
@ -110,7 +119,6 @@ public class SharedDataLockTest extends AbstractIntegrationBase {
Assert.assertEquals(1, lockSuccessCount.get());
// Other nodes should fail
Assert.assertEquals(2, lockFailedCount.get());
}
private SharedDataMessage sharedNodeData(String key, String value) {
@ -121,5 +129,4 @@ public class SharedDataLockTest extends AbstractIntegrationBase {
g.setTimestamp(System.currentTimeMillis());
return g;
}
}

228
gossip-itest/src/test/java/org/apache/gossip/SharedDataReplicationControlTest.java
View File

@ -28,10 +28,7 @@ import org.apache.gossip.replication.DataCenterReplicable;
import org.apache.gossip.replication.NotReplicable;
import org.apache.gossip.replication.Replicable;
import org.apache.gossip.replication.WhiteListReplicable;
import org.junit.Assert;
import org.junit.Test;
import org.junit.platform.runner.JUnitPlatform;
import org.junit.runner.RunWith;
import java.net.URI;
import java.net.URISyntaxException;
@ -44,84 +41,100 @@ import java.util.Map;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
@RunWith(JUnitPlatform.class)
public class SharedDataReplicationControlTest extends AbstractIntegrationBase {
@Test
public void sharedDataReplicationTest()
throws InterruptedException, UnknownHostException, URISyntaxException {
throws URISyntaxException {
generateStandardNodes(3);
// check whether the members are discovered
TUnit.assertThat(() -> {
int total = 0;
for (GossipManager node : nodes) {
total += node.getLiveMembers().size();
}
return total;
}).afterWaitingAtMost(20, TimeUnit.SECONDS).isEqualTo(2);
TUnit.assertThat(
() -> {
int total = 0;
for (GossipManager node : nodes) {
total += node.getLiveMembers().size();
}
return total;
})
.afterWaitingAtMost(20, TimeUnit.SECONDS)
.isEqualTo(2);
// Adding new shared data to Node 1 with default replication (replicate all)
nodes.get(0).gossipSharedData(sharedNodeData("public", "I am visible to all",
new AllReplicable<>()));
nodes
.get(0)
.gossipSharedData(sharedNodeData("public", "I am visible to all", new AllReplicable<>()));
// Adding new shared data to Node 1 with no replication (replicate none)
nodes.get(0).gossipSharedData(sharedNodeData("private", "I am private",
new NotReplicable<>()));
nodes.get(0).gossipSharedData(sharedNodeData("private", "I am private", new NotReplicable<>()));
List<LocalMember> whiteList = new ArrayList<>();
whiteList.add(nodes.get(1).getMyself());
// Adding new shared data to Node 1 with white list Node 2
nodes.get(0).gossipSharedData(sharedNodeData("wl", "white list",
new WhiteListReplicable<>(whiteList)));
nodes
.get(0)
.gossipSharedData(sharedNodeData("wl", "white list", new WhiteListReplicable<>(whiteList)));
List<LocalMember> blackList = new ArrayList<>();
blackList.add(nodes.get(1).getMyself());
// Adding new shared data to Node 1 with black list Node 2
nodes.get(0).gossipSharedData(sharedNodeData("bl", "black list",
new BlackListReplicable<>(blackList)));
nodes
.get(0)
.gossipSharedData(sharedNodeData("bl", "black list", new BlackListReplicable<>(blackList)));
TUnit.assertThat(() -> {
SharedDataMessage message = nodes.get(1).findSharedGossipData("public");
if(message == null){
return "";
}else {
return message.getPayload();
}
}).afterWaitingAtMost(20, TimeUnit.SECONDS).isEqualTo("I am visible to all");
TUnit.assertThat(
() -> {
SharedDataMessage message = nodes.get(1).findSharedGossipData("public");
if (message == null) {
return "";
} else {
return message.getPayload();
}
})
.afterWaitingAtMost(20, TimeUnit.SECONDS)
.isEqualTo("I am visible to all");
TUnit.assertThat(() -> {
SharedDataMessage message = nodes.get(2).findSharedGossipData("public");
if(message == null){
return "";
}else {
return message.getPayload();
}
}).afterWaitingAtMost(20, TimeUnit.SECONDS).isEqualTo("I am visible to all");
TUnit.assertThat(
() -> {
SharedDataMessage message = nodes.get(2).findSharedGossipData("public");
if (message == null) {
return "";
} else {
return message.getPayload();
}
})
.afterWaitingAtMost(20, TimeUnit.SECONDS)
.isEqualTo("I am visible to all");
// Node 2 must have shared data with key wl
TUnit.assertThat(() -> {
SharedDataMessage message = nodes.get(1).findSharedGossipData("wl");
if(message == null){
return "";
}else {
return message.getPayload();
}
}).afterWaitingAtMost(20, TimeUnit.SECONDS).isEqualTo("white list");
TUnit.assertThat(
() -> {
SharedDataMessage message = nodes.get(1).findSharedGossipData("wl");
if (message == null) {
return "";
} else {
return message.getPayload();
}
})
.afterWaitingAtMost(20, TimeUnit.SECONDS)
.isEqualTo("white list");
// Node 3 must have shared data with key bl
TUnit.assertThat(() -> {
SharedDataMessage message = nodes.get(2).findSharedGossipData("bl");
if(message == null){
return "";
}else {
return message.getPayload();
}
}).afterWaitingAtMost(20, TimeUnit.SECONDS).isEqualTo("black list");
TUnit.assertThat(
() -> {
SharedDataMessage message = nodes.get(2).findSharedGossipData("bl");
if (message == null) {
return "";
} else {
return message.getPayload();
}
})
.afterWaitingAtMost(20, TimeUnit.SECONDS)
.isEqualTo("black list");
}
@Test
public void sharedDataDcReplicationTest()
throws InterruptedException, UnknownHostException, URISyntaxException {
throws InterruptedException, UnknownHostException, URISyntaxException {
GossipSettings settings = new GossipSettings();
settings.setPersistRingState(false);
@ -137,56 +150,68 @@ public class SharedDataReplicationControlTest extends AbstractIntegrationBase {
RemoteMember seeder = new RemoteMember(cluster, URI.create("udp://127.0.0.1:5001"), "1");
// initialize 2 data centers with each having two racks
createDcNode(URI.create("udp://127.0.0.1:5001"), "1", settings, seeder, cluster,
"DataCenter1", "Rack1");
createDcNode(URI.create("udp://127.0.0.1:5002"), "2", settings, seeder, cluster,
"DataCenter1", "Rack2");
createDcNode(
URI.create("udp://127.0.0.1:5001"), "1", settings, seeder, cluster, "DataCenter1", "Rack1");
createDcNode(
URI.create("udp://127.0.0.1:5002"), "2", settings, seeder, cluster, "DataCenter1", "Rack2");
createDcNode(URI.create("udp://127.0.0.1:5006"), "6", settings, seeder, cluster,
"DataCenter2", "Rack1");
createDcNode(URI.create("udp://127.0.0.1:5007"), "7", settings, seeder, cluster,
"DataCenter2", "Rack1");
createDcNode(
URI.create("udp://127.0.0.1:5006"), "6", settings, seeder, cluster, "DataCenter2", "Rack1");
createDcNode(
URI.create("udp://127.0.0.1:5007"), "7", settings, seeder, cluster, "DataCenter2", "Rack1");
// check whether the members are discovered
TUnit.assertThat(() -> {
int total = 0;
for (int i = 0; i < 4; ++i) {
total += nodes.get(i).getLiveMembers().size();
}
return total;
}).afterWaitingAtMost(20, TimeUnit.SECONDS).isEqualTo(12);
TUnit.assertThat(
() -> {
int total = 0;
for (int i = 0; i < 4; ++i) {
total += nodes.get(i).getLiveMembers().size();
}
return total;
})
.afterWaitingAtMost(20, TimeUnit.SECONDS)
.isEqualTo(12);
// Node 1 has a shared key with 'Dc1Rack1'
nodes.get(0).gossipSharedData(sharedNodeData("Dc1Rack1", "I am belong to Dc1",
new DataCenterReplicable<>()));
nodes
.get(0)
.gossipSharedData(
sharedNodeData("Dc1Rack1", "I am belong to Dc1", new DataCenterReplicable<>()));
// Node 6 has a shared key with 'Dc2Rack1'
nodes.get(2).gossipSharedData(sharedNodeData("Dc2Rack1", "I am belong to Dc2",
new DataCenterReplicable<>()));
nodes
.get(2)
.gossipSharedData(
sharedNodeData("Dc2Rack1", "I am belong to Dc2", new DataCenterReplicable<>()));
// Node 2 must have the shared data with key 'Dc1Rack1'
TUnit.assertThat(() -> {
SharedDataMessage message = nodes.get(1).findSharedGossipData("Dc1Rack1");
if(message == null){
return "";
}else {
return message.getPayload();
}
}).afterWaitingAtMost(20, TimeUnit.SECONDS).isEqualTo("I am belong to Dc1");
TUnit.assertThat(
() -> {
SharedDataMessage message = nodes.get(1).findSharedGossipData("Dc1Rack1");
if (message == null) {
return "";
} else {
return message.getPayload();
}
})
.afterWaitingAtMost(20, TimeUnit.SECONDS)
.isEqualTo("I am belong to Dc1");
// Node 7 must have the shared data with key 'Dc2Rack1'
TUnit.assertThat(() -> {
SharedDataMessage message = nodes.get(3).findSharedGossipData("Dc2Rack1");
if(message == null){
return "";
}else {
return message.getPayload();
}
}).afterWaitingAtMost(20, TimeUnit.SECONDS).isEqualTo("I am belong to Dc2");
TUnit.assertThat(
() -> {
SharedDataMessage message = nodes.get(3).findSharedGossipData("Dc2Rack1");
if (message == null) {
return "";
} else {
return message.getPayload();
}
})
.afterWaitingAtMost(20, TimeUnit.SECONDS)
.isEqualTo("I am belong to Dc2");
}
private SharedDataMessage sharedNodeData(String key, String value,
Replicable<SharedDataMessage> replicable) {
private SharedDataMessage sharedNodeData(
String key, String value, Replicable<SharedDataMessage> replicable) {
SharedDataMessage g = new SharedDataMessage();
g.setExpireAt(Long.MAX_VALUE);
g.setKey(key);
@ -196,17 +221,28 @@ public class SharedDataReplicationControlTest extends AbstractIntegrationBase {
return g;
}
private void createDcNode(URI uri, String id, GossipSettings settings, RemoteMember seeder,
String cluster, String dataCenter, String rack){
private void createDcNode(
URI uri,
String id,
GossipSettings settings,
RemoteMember seeder,
String cluster,
String dataCenter,
String rack) {
Map<String, String> props = new HashMap<>();
props.put(DatacenterRackAwareActiveGossiper.DATACENTER, dataCenter);
props.put(DatacenterRackAwareActiveGossiper.RACK, rack);
GossipManager dcNode = GossipManagerBuilder.newBuilder().cluster(cluster).uri(uri).id(id)
.gossipSettings(settings).gossipMembers(Arrays.asList(seeder)).properties(props)
GossipManager dcNode =
GossipManagerBuilder.newBuilder()
.cluster(cluster)
.uri(uri)
.id(id)
.gossipSettings(settings)
.gossipMembers(Arrays.asList(seeder))
.properties(props)
.build();
dcNode.init();
register(dcNode);
}
}

117
gossip-itest/src/test/java/org/apache/gossip/ShutdownDeadtimeTest.java
View File

@ -18,37 +18,28 @@
package org.apache.gossip;
import io.teknek.tunit.TUnit;
import java.net.URI;
import java.net.URISyntaxException;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import lombok.extern.slf4j.Slf4j;
import org.apache.gossip.manager.GossipManager;
import org.apache.gossip.manager.GossipManagerBuilder;
import org.apache.log4j.Logger;
import org.junit.platform.runner.JUnitPlatform;
import org.junit.jupiter.api.Test;
import org.junit.runner.RunWith;
@RunWith(JUnitPlatform.class)
@Slf4j
public class ShutdownDeadtimeTest {
private static final Logger log = Logger.getLogger(ShutdownDeadtimeTest.class);
// Note: this test is floppy depending on the values in GossipSettings (smaller values seem to do harm), and the
// Note: this test is floppy depending on the values in GossipSettings (smaller values seem to do
// harm), and the
// sleep that happens after startup.
@Test
public void DeadNodesDoNotComeAliveAgain()
throws InterruptedException, UnknownHostException, URISyntaxException {
throws InterruptedException, URISyntaxException {
GossipSettings settings = new GossipSettings(100, 10000, 1000, 1, 10.0, "normal", false);
settings.setPersistRingState(false);
settings.setPersistDataState(false);
@ -59,11 +50,13 @@ public class ShutdownDeadtimeTest {
URI uri = new URI("udp://" + "127.0.0.1" + ":" + (30300 + i));
startupMembers.add(new RemoteMember(cluster, uri, i + ""));
}
final List<GossipManager> clients = Collections.synchronizedList(new ArrayList<GossipManager>());
final List<GossipManager> clients =
Collections.synchronizedList(new ArrayList<GossipManager>());
final int clusterMembers = 5;
for (int i = 1; i < clusterMembers + 1; ++i) {
URI uri = new URI("udp://" + "127.0.0.1" + ":" + (30300 + i));
GossipManager gossipService = GossipManagerBuilder.newBuilder()
GossipManager gossipService =
GossipManagerBuilder.newBuilder()
.cluster(cluster)
.uri(uri)
.id(i + "")
@ -72,77 +65,77 @@ public class ShutdownDeadtimeTest {
.build();
clients.add(gossipService);
gossipService.init();
Thread.sleep(1000);
Thread.sleep(1000);
}
TUnit.assertThat(new Callable<Integer>() {
public Integer call() throws Exception {
int total = 0;
for (int i = 0; i < clusterMembers; ++i) {
total += clients.get(i).getLiveMembers().size();
}
return total;
}
}).afterWaitingAtMost(40, TimeUnit.SECONDS).isEqualTo(20);
TUnit.assertThat(
() -> {
int total = 0;
for (int i = 0; i < clusterMembers; ++i) {
total += clients.get(i).getLiveMembers().size();
}
return total;
})
.afterWaitingAtMost(40, TimeUnit.SECONDS)
.isEqualTo(20);
// shutdown one client and verify that one client is lost.
Random r = new Random();
int randomClientId = r.nextInt(clusterMembers);
log.info("shutting down " + randomClientId);
final int shutdownPort = clients.get(randomClientId).getMyself().getUri()
.getPort();
final int shutdownPort = clients.get(randomClientId).getMyself().getUri().getPort();
final String shutdownId = clients.get(randomClientId).getMyself().getId();
clients.get(randomClientId).shutdown();
TUnit.assertThat(new Callable<Integer>() {
public Integer call() throws Exception {
int total = 0;
for (int i = 0; i < clusterMembers; ++i) {
total += clients.get(i).getLiveMembers().size();
}
return total;
}
}).afterWaitingAtMost(40, TimeUnit.SECONDS).isEqualTo(16);
TUnit.assertThat(
() -> {
int total = 0;
for (int i = 0; i < clusterMembers; ++i) {
total += clients.get(i).getLiveMembers().size();
}
return total;
})
.afterWaitingAtMost(40, TimeUnit.SECONDS)
.isEqualTo(16);
clients.remove(randomClientId);
TUnit.assertThat(new Callable<Integer>() {
public Integer call() throws Exception {
int total = 0;
for (int i = 0; i < clusterMembers - 1; ++i) {
total += clients.get(i).getDeadMembers().size();
}
return total;
}
}).afterWaitingAtMost(50, TimeUnit.SECONDS).isEqualTo(4);
TUnit.assertThat(
() -> {
int total = 0;
for (int i = 0; i < clusterMembers - 1; ++i) {
total += clients.get(i).getDeadMembers().size();
}
return total;
})
.afterWaitingAtMost(50, TimeUnit.SECONDS)
.isEqualTo(4);
URI uri = new URI("udp://" + "127.0.0.1" + ":" + shutdownPort);
// start client again
GossipManager gossipService = GossipManagerBuilder.newBuilder()
GossipManager gossipService =
GossipManagerBuilder.newBuilder()
.gossipSettings(settings)
.cluster(cluster)
.uri(uri)
.id(shutdownId+"")
.id(shutdownId)
.gossipMembers(startupMembers)
.build();
clients.add(gossipService);
gossipService.init();
// verify that the client is alive again for every node
TUnit.assertThat(new Callable<Integer>() {
public Integer call() throws Exception {
int total = 0;
for (int i = 0; i < clusterMembers; ++i) {
total += clients.get(i).getLiveMembers().size();
}
return total;
}
}).afterWaitingAtMost(60, TimeUnit.SECONDS).isEqualTo(20);
TUnit.assertThat(
() -> {
int total = 0;
for (int i = 0; i < clusterMembers; ++i) {
total += clients.get(i).getLiveMembers().size();
}
return total;
})
.afterWaitingAtMost(60, TimeUnit.SECONDS)
.isEqualTo(20);
for (int i = 0; i < clusterMembers; ++i) {
final int j = i;
new Thread() {
public void run(){
clients.get(j).shutdown();
}
}.start();
new Thread(() -> clients.get(j).shutdown()).start();
}
}
}

60
gossip-itest/src/test/java/org/apache/gossip/SignedMessageTest.java
View File

@ -40,22 +40,27 @@ import io.teknek.tunit.TUnit;
public class SignedMessageTest extends AbstractIntegrationBase {
private GossipSettings gossiperThatSigns(){
private GossipSettings gossiperThatSigns() {
GossipSettings settings = new GossipSettings();
settings.setPersistRingState(false);
settings.setPersistDataState(false);
settings.setSignMessages(true);
return settings;
}
private GossipSettings gossiperThatSigns(String keysDir){
private GossipSettings gossiperThatSigns(String keysDir) {
GossipSettings settings = gossiperThatSigns();
settings.setPathToKeyStore(Objects.requireNonNull(keysDir));
return settings;
}
@Test
public void dataTest() throws InterruptedException, URISyntaxException, NoSuchAlgorithmException, NoSuchProviderException, IOException {
public void dataTest()
throws InterruptedException,
URISyntaxException,
NoSuchAlgorithmException,
NoSuchProviderException,
IOException {
final String keys = System.getProperty("java.io.tmpdir") + "/keys";
GossipSettings settings = gossiperThatSigns(keys);
setup(keys);
@ -68,7 +73,8 @@ public class SignedMessageTest extends AbstractIntegrationBase {
final List<GossipManager> clients = new ArrayList<>();
for (int i = 1; i < 3; ++i) {
URI uri = new URI("udp://" + "127.0.0.1" + ":" + (30000 + i));
GossipManager gossipService = GossipManagerBuilder.newBuilder()
GossipManager gossipService =
GossipManagerBuilder.newBuilder()
.cluster(cluster)
.uri(uri)
.id(i + "")
@ -82,25 +88,28 @@ public class SignedMessageTest extends AbstractIntegrationBase {
assertOnlySignedMessages(clients);
cleanup(keys, clients);
}
private void assertTwoAlive(List<GossipManager> clients){
TUnit.assertThat(() -> {
int total = 0;
for (int i = 0; i < clients.size(); ++i) {
total += clients.get(i).getLiveMembers().size();
}
return total;
}).afterWaitingAtMost(20, TimeUnit.SECONDS).isEqualTo(2);
private void assertTwoAlive(List<GossipManager> clients) {
TUnit.assertThat(
() -> {
int total = 0;
for (int i = 0; i < clients.size(); ++i) {
total += clients.get(i).getLiveMembers().size();
}
return total;
})
.afterWaitingAtMost(20, TimeUnit.SECONDS)
.isEqualTo(2);
}
private void assertOnlySignedMessages(List<GossipManager> clients){
Assert.assertEquals(0, clients.get(0).getRegistry()
.meter(PassiveGossipConstants.UNSIGNED_MESSAGE).getCount());
Assert.assertTrue(clients.get(0).getRegistry()
.meter(PassiveGossipConstants.SIGNED_MESSAGE).getCount() > 0);
private void assertOnlySignedMessages(List<GossipManager> clients) {
Assert.assertEquals(
0, clients.get(0).getRegistry().meter(PassiveGossipConstants.UNSIGNED_MESSAGE).getCount());
Assert.assertTrue(
clients.get(0).getRegistry().meter(PassiveGossipConstants.SIGNED_MESSAGE).getCount() > 0);
}
private void cleanup(String keys, List<GossipManager> clients){
private void cleanup(String keys, List<GossipManager> clients) {
new File(keys, "1").delete();
new File(keys, "2").delete();
new File(keys).delete();
@ -108,8 +117,9 @@ public class SignedMessageTest extends AbstractIntegrationBase {
clients.get(i).shutdown();
}
}
private void setup(String keys) throws NoSuchAlgorithmException, NoSuchProviderException, IOException {
private void setup(String keys)
throws NoSuchAlgorithmException, NoSuchProviderException, IOException {
new File(keys).mkdir();
KeyTool.generatePubandPrivateKeyFiles(keys, "1");
KeyTool.generatePubandPrivateKeyFiles(keys, "2");

79
gossip-itest/src/test/java/org/apache/gossip/StartupSettingsTest.java
View File

@ -17,9 +17,9 @@
*/
package org.apache.gossip;
import lombok.extern.slf4j.Slf4j;
import org.apache.gossip.manager.GossipManager;
import org.apache.gossip.manager.GossipManagerBuilder;
import org.apache.log4j.Logger;
import org.junit.jupiter.api.Test;
@ -29,61 +29,68 @@ import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.UUID;
import org.junit.platform.runner.JUnitPlatform;
import org.junit.runner.RunWith;
/**
* Tests support of using {@code StartupSettings} and thereby reading
* setup config from file.
*/
@RunWith(JUnitPlatform.class)
/** Tests support of using {@code StartupSettings} and thereby reading setup config from file. */
@Slf4j
public class StartupSettingsTest {
private static final Logger log = Logger.getLogger(StartupSettingsTest.class);
private static final String CLUSTER = UUID.randomUUID().toString();
@Test
public void testUsingSettingsFile() throws IOException, InterruptedException, URISyntaxException {
File settingsFile = File.createTempFile("gossipTest",".json");
public void testUsingSettingsFile() throws IOException, URISyntaxException {
File settingsFile = File.createTempFile("gossipTest", ".json");
settingsFile.deleteOnExit();
writeSettingsFile(settingsFile);
URI uri = new URI("udp://" + "127.0.0.1" + ":" + 50000);
GossipSettings firstGossipSettings = new GossipSettings();
firstGossipSettings.setTransportManagerClass("org.apache.gossip.transport.UnitTestTransportManager");
firstGossipSettings.setProtocolManagerClass("org.apache.gossip.protocol.UnitTestProtocolManager");
GossipManager firstService = GossipManagerBuilder.newBuilder()
firstGossipSettings.setTransportManagerClass(
"org.apache.gossip.transport.UnitTestTransportManager");
firstGossipSettings.setProtocolManagerClass(
"org.apache.gossip.protocol.UnitTestProtocolManager");
GossipManager firstService =
GossipManagerBuilder.newBuilder()
.cluster(CLUSTER)
.uri(uri)
.id("1")
.gossipSettings(firstGossipSettings).build();
.gossipSettings(firstGossipSettings)
.build();
firstService.init();
GossipManager manager = GossipManagerBuilder.newBuilder()
.startupSettings(StartupSettings.fromJSONFile(settingsFile)).build();
GossipManager manager =
GossipManagerBuilder.newBuilder()
.startupSettings(StartupSettings.fromJSONFile(settingsFile))
.build();
manager.init();
firstService.shutdown();
manager.shutdown();
}
private void writeSettingsFile( File target ) throws IOException {
private void writeSettingsFile(File target) throws IOException {
String settings =
"[{\n" + // It is odd that this is meant to be in an array, but oh well.
" \"cluster\":\"" + CLUSTER + "\",\n" +
" \"id\":\"" + "2" + "\",\n" +
" \"uri\":\"udp://127.0.0.1:50001\",\n" +
" \"gossip_interval\":1000,\n" +
" \"window_size\":1000,\n" +
" \"minimum_samples\":5,\n" +
" \"cleanup_interval\":10000,\n" +
" \"convict_threshold\":2.6,\n" +
" \"distribution\":\"exponential\",\n" +
" \"transport_manager_class\":\"org.apache.gossip.transport.UnitTestTransportManager\",\n" +
" \"protocol_manager_class\":\"org.apache.gossip.protocol.UnitTestProtocolManager\",\n" +
" \"properties\":{},\n" +
" \"members\":[\n" +
" {\"cluster\": \"" + CLUSTER + "\",\"uri\":\"udp://127.0.0.1:5000\"}\n" +
" ]\n" +
"}]";
"[{\n"
+ // It is odd that this is meant to be in an array, but oh well.
" \"cluster\":\""
+ CLUSTER
+ "\",\n"
+ " \"id\":\""
+ "2"
+ "\",\n"
+ " \"uri\":\"udp://127.0.0.1:50001\",\n"
+ " \"gossip_interval\":1000,\n"
+ " \"window_size\":1000,\n"
+ " \"minimum_samples\":5,\n"
+ " \"cleanup_interval\":10000,\n"
+ " \"convict_threshold\":2.6,\n"
+ " \"distribution\":\"exponential\",\n"
+ " \"transport_manager_class\":\"org.apache.gossip.transport.UnitTestTransportManager\",\n"
+ " \"protocol_manager_class\":\"org.apache.gossip.protocol.UnitTestProtocolManager\",\n"
+ " \"properties\":{},\n"
+ " \"members\":[\n"
+ " {\"cluster\": \""
+ CLUSTER
+ "\",\"uri\":\"udp://127.0.0.1:5000\"}\n"
+ " ]\n"
+ "}]";
log.info( "Using settings file with contents of:\n---\n" + settings + "\n---" );
log.info("Using settings file with contents of:\n---\n" + settings + "\n---");
FileOutputStream output = new FileOutputStream(target);
output.write(settings.getBytes());
output.close();

40
gossip-itest/src/test/java/org/apache/gossip/TenNodeThreeSeedTest.java
View File

@ -15,7 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.gossip;
package org.apache.gossip;
import io.teknek.tunit.TUnit;
@ -27,13 +27,10 @@ import java.util.List;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import org.junit.platform.runner.JUnitPlatform;
import org.junit.runner.RunWith;
import org.apache.gossip.manager.GossipManager;
import org.apache.gossip.manager.GossipManagerBuilder;
import org.junit.jupiter.api.Test;
@RunWith(JUnitPlatform.class)
public class TenNodeThreeSeedTest {
@Test
@ -53,15 +50,16 @@ public class TenNodeThreeSeedTest {
String cluster = UUID.randomUUID().toString();
int seedNodes = 3;
List<Member> startupMembers = new ArrayList<>();
for (int i = 1; i < seedNodes+1; ++i) {
for (int i = 1; i < seedNodes + 1; ++i) {
URI uri = new URI("udp://" + "127.0.0.1" + ":" + (base + i));
startupMembers.add(new RemoteMember(cluster, uri, i + ""));
}
final List<GossipManager> clients = new ArrayList<>();
final int clusterMembers = 5;
for (int i = 1; i < clusterMembers+1; ++i) {
for (int i = 1; i < clusterMembers + 1; ++i) {
URI uri = new URI("udp://" + "127.0.0.1" + ":" + (base + i));
GossipManager gossipService = GossipManagerBuilder.newBuilder()
GossipManager gossipService =
GossipManagerBuilder.newBuilder()
.cluster(cluster)
.uri(uri)
.id(i + "")
@ -70,20 +68,24 @@ public class TenNodeThreeSeedTest {
.build();
gossipService.init();
clients.add(gossipService);
}
TUnit.assertThat(new Callable<Integer> (){
public Integer call() throws Exception {
int total = 0;
for (int i = 0; i < clusterMembers; ++i) {
total += clients.get(i).getLiveMembers().size();
}
return total;
}}).afterWaitingAtMost(40, TimeUnit.SECONDS).isEqualTo(20);
}
TUnit.assertThat(
new Callable<Integer>() {
public Integer call() throws Exception {
int total = 0;
for (int i = 0; i < clusterMembers; ++i) {
total += clients.get(i).getLiveMembers().size();
}
return total;
}
})
.afterWaitingAtMost(40, TimeUnit.SECONDS)
.isEqualTo(20);
for (int i = 0; i < clusterMembers; ++i) {
int j = i;
new Thread(){
public void run(){
new Thread() {
public void run() {
clients.get(j).shutdown();
}
}.start();