Skip to content

Commit 71fe298

Browse files
artem-vartem-v
authored
Merge pull request #389 from scalecube/remove_removedMembersHistory
Tests cleanup
2 parents 036fd61 + d9f720f commit 71fe298

File tree

8 files changed

+85
-182
lines changed

8 files changed

+85
-182
lines changed

‎cluster-api/src/main/java/io/scalecube/cluster/membership/MembershipConfig.java‎

Lines changed: 0 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,6 @@ public final class MembershipConfig implements Cloneable {
2828
private int syncInterval = DEFAULT_SYNC_INTERVAL;
2929
private int syncTimeout = DEFAULT_SYNC_TIMEOUT;
3030
private int suspicionMult = DEFAULT_SUSPICION_MULT;
31-
private int removedMembersHistorySize = 42;
3231
private String namespace = "default";
3332

3433
public MembershipConfig() {}
@@ -158,22 +157,6 @@ public MembershipConfig namespace(String namespace) {
158157
return m;
159158
}
160159

161-
public int removedMembersHistorySize() {
162-
return removedMembersHistorySize;
163-
}
164-
165-
/**
166-
* Setter for {@code removedMembersHistorySize}.
167-
*
168-
* @param removedMembersHistorySize history size for remove members
169-
* @return new {@code MembershipConfig} instance
170-
*/
171-
public MembershipConfig removedMembersHistorySize(int removedMembersHistorySize) {
172-
MembershipConfig m = clone();
173-
m.removedMembersHistorySize = removedMembersHistorySize;
174-
return m;
175-
}
176-
177160
@Override
178161
public MembershipConfig clone() {
179162
try {
@@ -191,7 +174,6 @@ public String toString() {
191174
.add("syncTimeout=" + syncTimeout)
192175
.add("suspicionMult=" + suspicionMult)
193176
.add("namespace='" + namespace + "'")
194-
.add("removedMembersHistorySize=" + removedMembersHistorySize)
195177
.toString();
196178
}
197179
}

‎cluster/src/main/java/io/scalecube/cluster/fdetector/FailureDetectorImpl.java‎

Lines changed: 0 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -417,13 +417,4 @@ private boolean isTransitPingAck(Message message) {
417417
return PING_ACK.equals(message.qualifier())
418418
&& message.<PingData>data().getOriginalIssuer() != null;
419419
}
420-
421-
/**
422-
* <b>NOTE:</b> this method is for testing purpose only.
423-
*
424-
* @return transport
425-
*/
426-
Transport getTransport() {
427-
return transport;
428-
}
429420
}

‎cluster/src/main/java/io/scalecube/cluster/gossip/GossipProtocolImpl.java‎

Lines changed: 0 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -365,22 +365,4 @@ private Set<String> getGossipsThatMostLikelyDisseminated(long period) {
365365
.map(gossipState -> gossipState.gossip().gossipId())
366366
.collect(Collectors.toSet());
367367
}
368-
369-
/**
370-
* <b>NOTE:</b> this method is for testing purpose only.
371-
*
372-
* @return transport
373-
*/
374-
Transport getTransport() {
375-
return transport;
376-
}
377-
378-
/**
379-
* <b>NOTE:</b> this method is for testing purpose only.
380-
*
381-
* @return local member
382-
*/
383-
Member getMember() {
384-
return localMember;
385-
}
386368
}

‎cluster/src/main/java/io/scalecube/cluster/membership/MembershipProtocolImpl.java‎

Lines changed: 3 additions & 90 deletions
Original file line numberDiff line numberDiff line change
@@ -34,10 +34,8 @@
3434
import java.util.Optional;
3535
import java.util.Set;
3636
import java.util.UUID;
37-
import java.util.concurrent.CopyOnWriteArrayList;
3837
import java.util.concurrent.ThreadLocalRandom;
3938
import java.util.concurrent.TimeUnit;
40-
import java.util.function.Predicate;
4139
import java.util.stream.Collectors;
4240
import java.util.stream.Stream;
4341
import org.slf4j.Logger;
@@ -50,6 +48,7 @@
5048
import reactor.core.publisher.Sinks;
5149
import reactor.core.scheduler.Scheduler;
5250

51+
@SuppressWarnings({"FieldCanBeLocal", "unused"})
5352
public final class MembershipProtocolImpl implements MembershipProtocol {
5453

5554
private static final Logger LOGGER = LoggerFactory.getLogger(MembershipProtocol.class);
@@ -84,7 +83,6 @@ private enum MembershipUpdateReason {
8483

8584
private final Map<String, MembershipRecord> membershipTable = new HashMap<>();
8685
private final Map<String, Member> members = new HashMap<>();
87-
private final List<MembershipEvent> removedMembersHistory = new CopyOnWriteArrayList<>();
8886
private final Set<String> aliveEmittedSet = new HashSet<>();
8987

9088
// Subject
@@ -157,10 +155,8 @@ public MembershipProtocolImpl(
157155
.publishOn(scheduler)
158156
.subscribe(
159157
this::onMembershipGossip,
160-
ex -> LOGGER.error("[{}][onMembershipGossip][error] cause:", localMember, ex)),
161-
listen() // Listen removed members for monitoring
162-
.filter(MembershipEvent::isRemoved)
163-
.subscribe(this::onMemberRemoved)));
158+
ex ->
159+
LOGGER.error("[{}][onMembershipGossip][error] cause:", localMember, ex))));
164160
}
165161

166162
// Remove duplicates and local address(es)
@@ -863,87 +859,4 @@ private Mono<Void> spreadMembershipGossip(MembershipRecord r) {
863859
.then();
864860
});
865861
}
866-
867-
/**
868-
* <b>NOTE:</b> this method is for testing purpose only.
869-
*
870-
* @return failure detector
871-
*/
872-
FailureDetector getFailureDetector() {
873-
return failureDetector;
874-
}
875-
876-
/**
877-
* <b>NOTE:</b> this method is for testing purpose only.
878-
*
879-
* @return gossip
880-
*/
881-
GossipProtocol getGossipProtocol() {
882-
return gossipProtocol;
883-
}
884-
885-
/**
886-
* <b>NOTE:</b> this method is for testing purpose only.
887-
*
888-
* @return transport
889-
*/
890-
Transport getTransport() {
891-
return transport;
892-
}
893-
894-
/**
895-
* <b>NOTE:</b> this method is for testing purpose only.
896-
*
897-
* @return metadataStore
898-
*/
899-
MetadataStore getMetadataStore() {
900-
return metadataStore;
901-
}
902-
903-
/**
904-
* <b>NOTE:</b> this method is for testing purpose only.
905-
*
906-
* @return transport
907-
*/
908-
List<MembershipRecord> getMembershipRecords() {
909-
return Collections.unmodifiableList(new ArrayList<>(membershipTable.values()));
910-
}
911-
912-
// ===============================================================
913-
// ============== Helper Methods for Monitoring ==================
914-
// ===============================================================
915-
916-
private int getIncarnation() {
917-
return membershipTable.get(localMember.id()).incarnation();
918-
}
919-
920-
private List<Member> getAliveMembers() {
921-
return findRecordsByCondition(MembershipRecord::isAlive);
922-
}
923-
924-
private List<Member> getSuspectedMembers() {
925-
return findRecordsByCondition(MembershipRecord::isSuspect);
926-
}
927-
928-
private List<Member> getRemovedMembers() {
929-
return removedMembersHistory.stream().map(MembershipEvent::member).collect(Collectors.toList());
930-
}
931-
932-
private List<Member> findRecordsByCondition(Predicate<MembershipRecord> condition) {
933-
return getMembershipRecords().stream()
934-
.filter(condition)
935-
.map(MembershipRecord::member)
936-
.collect(Collectors.toList());
937-
}
938-
939-
private void onMemberRemoved(MembershipEvent event) {
940-
int s = membershipConfig.removedMembersHistorySize();
941-
if (s <= 0) {
942-
return;
943-
}
944-
removedMembersHistory.add(event);
945-
if (removedMembersHistory.size() > s) {
946-
removedMembersHistory.remove(0);
947-
}
948-
}
949862
}

‎cluster/src/test/java/io/scalecube/cluster/BaseTest.java‎

Lines changed: 17 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
import io.scalecube.cluster.transport.api.TransportConfig;
77
import io.scalecube.cluster.utils.NetworkEmulatorTransport;
88
import io.scalecube.transport.netty.tcp.TcpTransportFactory;
9+
import java.lang.reflect.Field;
910
import java.time.Duration;
1011
import java.util.concurrent.TimeUnit;
1112
import org.junit.jupiter.api.AfterEach;
@@ -30,32 +31,43 @@ public final void baseTearDown(TestInfo testInfo) {
3031
LOGGER.info("***** Test finished : " + testInfo.getDisplayName() + " *****");
3132
}
3233

33-
protected void awaitSeconds(long seconds) {
34+
public static <T> T getField(Object obj, String fieldName) {
35+
try {
36+
final Field field = obj.getClass().getDeclaredField(fieldName);
37+
field.setAccessible(true);
38+
//noinspection unchecked
39+
return (T) field.get(obj);
40+
} catch (Exception ex) {
41+
throw new RuntimeException(ex);
42+
}
43+
}
44+
45+
public static void awaitSeconds(long seconds) {
3446
try {
3547
TimeUnit.SECONDS.sleep(seconds);
3648
} catch (InterruptedException e) {
3749
throw Exceptions.propagate(e);
3850
}
3951
}
4052

41-
protected void awaitSuspicion(int clusterSize) {
53+
public static void awaitSuspicion(int clusterSize) {
4254
int defaultSuspicionMult = MembershipConfig.DEFAULT_SUSPICION_MULT;
4355
int pingInterval = MembershipProtocolTest.PING_INTERVAL;
4456
long suspicionTimeoutSec =
4557
ClusterMath.suspicionTimeout(defaultSuspicionMult, clusterSize, pingInterval) / 1000;
4658
awaitSeconds(suspicionTimeoutSec + 2);
4759
}
4860

49-
protected NetworkEmulatorTransport createTransport() {
61+
public static NetworkEmulatorTransport createTransport() {
5062
return createTransport(TransportConfig.defaultConfig());
5163
}
5264

53-
protected NetworkEmulatorTransport createTransport(TransportConfig transportConfig) {
65+
public static NetworkEmulatorTransport createTransport(TransportConfig transportConfig) {
5466
return new NetworkEmulatorTransport(
5567
Transport.bindAwait(transportConfig.transportFactory(new TcpTransportFactory())));
5668
}
5769

58-
protected void destroyTransport(Transport transport) {
70+
public static void destroyTransport(Transport transport) {
5971
if (transport == null || transport.isStopped()) {
6072
return;
6173
}

‎cluster/src/test/java/io/scalecube/cluster/fdetector/FailureDetectorTest.java‎

Lines changed: 10 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -421,22 +421,22 @@ private FailureDetectorImpl createFd(
421421
return new FailureDetectorImpl(localMember, transport, membershipFlux, config, scheduler);
422422
}
423423

424-
private void start(List<FailureDetectorImpl> fdetectors) {
424+
private static void start(List<FailureDetectorImpl> fdetectors) {
425425
for (FailureDetectorImpl fd : fdetectors) {
426426
fd.start();
427427
}
428428
}
429429

430-
private void stop(List<FailureDetectorImpl> fdetectors) {
430+
private static void stop(List<FailureDetectorImpl> fdetectors) {
431431
for (FailureDetectorImpl fd : fdetectors) {
432432
fd.stop();
433433
}
434434
for (FailureDetectorImpl fd : fdetectors) {
435-
destroyTransport(fd.getTransport());
435+
destroyTransport(BaseTest.getField(fd, "transport"));
436436
}
437437
}
438438

439-
private void assertStatus(
439+
private static void assertStatus(
440440
Address address,
441441
MemberStatus status,
442442
Collection<FailureDetectorEvent> events,
@@ -461,10 +461,11 @@ private void assertStatus(
461461
}
462462
}
463463

464-
private Future<List<FailureDetectorEvent>> listenNextEventFor(
464+
private static Future<List<FailureDetectorEvent>> listenNextEventFor(
465465
FailureDetectorImpl fd, List<Address> addresses) {
466+
final Transport transport = BaseTest.getField(fd, "transport");
466467
addresses = new ArrayList<>(addresses);
467-
addresses.remove(fd.getTransport().address()); // exclude self
468+
addresses.remove(transport.address()); // exclude self
468469
if (addresses.isEmpty()) {
469470
throw new IllegalArgumentException();
470471
}
@@ -479,15 +480,16 @@ private Future<List<FailureDetectorEvent>> listenNextEventFor(
479480
return allOf(resultFuture);
480481
}
481482

482-
private Collection<FailureDetectorEvent> awaitEvents(Future<List<FailureDetectorEvent>> events) {
483+
private static Collection<FailureDetectorEvent> awaitEvents(
484+
Future<List<FailureDetectorEvent>> events) {
483485
try {
484486
return events.get(10, TimeUnit.SECONDS);
485487
} catch (Exception e) {
486488
throw new RuntimeException(e);
487489
}
488490
}
489491

490-
private <T> CompletableFuture<List<T>> allOf(List<CompletableFuture<T>> futuresList) {
492+
private static <T> CompletableFuture<List<T>> allOf(List<CompletableFuture<T>> futuresList) {
491493
CompletableFuture<Void> allFuturesResult =
492494
CompletableFuture.allOf(futuresList.toArray(new CompletableFuture[0]));
493495
return allFuturesResult.thenApply(

‎cluster/src/test/java/io/scalecube/cluster/gossip/GossipProtocolTest.java‎

Lines changed: 11 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -125,19 +125,20 @@ void testGossipProtocol(int membersNum, int lossPercent, int meanDelay) throws E
125125
final CountDownLatch latch = new CountDownLatch(membersNum - 1);
126126
final Map<Member, Member> receivers = new ConcurrentHashMap<>();
127127
final AtomicBoolean doubleDelivery = new AtomicBoolean(false);
128-
for (final GossipProtocolImpl protocol : gossipProtocols) {
129-
protocol
128+
for (final GossipProtocolImpl gossipProtocol : gossipProtocols) {
129+
gossipProtocol
130130
.listen()
131131
.subscribe(
132132
gossip -> {
133+
final Member localMember = BaseTest.getField(gossipProtocol, "localMember");
134+
final Transport transport = BaseTest.getField(gossipProtocol, "transport");
135+
133136
if (gossipData.equals(gossip.data())) {
134-
boolean firstTimeAdded =
135-
receivers.put(protocol.getMember(), protocol.getMember()) == null;
137+
boolean firstTimeAdded = receivers.put(localMember, localMember) == null;
136138
if (firstTimeAdded) {
137139
latch.countDown();
138140
} else {
139-
LOGGER.error(
140-
"Delivered gossip twice to: {}", protocol.getTransport().address());
141+
LOGGER.error("Delivered gossip twice to: {}", transport.address());
141142
doubleDelivery.set(true);
142143
}
143144
}
@@ -212,7 +213,7 @@ void testGossipProtocol(int membersNum, int lossPercent, int meanDelay) throws E
212213
private LongSummaryStatistics computeMessageSentStats(List<GossipProtocolImpl> gossipProtocols) {
213214
List<Long> messageSentPerNode = new ArrayList<>(gossipProtocols.size());
214215
for (GossipProtocolImpl gossipProtocol : gossipProtocols) {
215-
NetworkEmulatorTransport transport = (NetworkEmulatorTransport) gossipProtocol.getTransport();
216+
final NetworkEmulatorTransport transport = BaseTest.getField(gossipProtocol, "transport");
216217
messageSentPerNode.add(transport.networkEmulator().totalMessageSentCount());
217218
}
218219
return messageSentPerNode.stream().mapToLong(v -> v).summaryStatistics();
@@ -221,7 +222,7 @@ private LongSummaryStatistics computeMessageSentStats(List<GossipProtocolImpl> g
221222
private LongSummaryStatistics computeMessageLostStats(List<GossipProtocolImpl> gossipProtocols) {
222223
List<Long> messageLostPerNode = new ArrayList<>(gossipProtocols.size());
223224
for (GossipProtocolImpl gossipProtocol : gossipProtocols) {
224-
NetworkEmulatorTransport transport = (NetworkEmulatorTransport) gossipProtocol.getTransport();
225+
final NetworkEmulatorTransport transport = BaseTest.getField(gossipProtocol, "transport");
225226
messageLostPerNode.add(transport.networkEmulator().totalOutboundMessageLostCount());
226227
}
227228
return messageLostPerNode.stream().mapToLong(v -> v).summaryStatistics();
@@ -272,7 +273,7 @@ private GossipProtocolImpl initGossipProtocol(Transport transport, List<Address>
272273
return gossipProtocol;
273274
}
274275

275-
private void destroyGossipProtocols(List<GossipProtocolImpl> gossipProtocols) {
276+
private static void destroyGossipProtocols(List<GossipProtocolImpl> gossipProtocols) {
276277
// Stop all gossip protocols
277278
for (GossipProtocolImpl gossipProtocol : gossipProtocols) {
278279
gossipProtocol.stop();
@@ -281,7 +282,7 @@ private void destroyGossipProtocols(List<GossipProtocolImpl> gossipProtocols) {
281282
// Stop all transports
282283
List<Mono<Void>> futures = new ArrayList<>();
283284
for (GossipProtocolImpl gossipProtocol : gossipProtocols) {
284-
futures.add(gossipProtocol.getTransport().stop());
285+
futures.add(BaseTest.<Transport>getField(gossipProtocol, "transport").stop());
285286
}
286287

287288
try {

0 commit comments

Comments
 (0)