KubeClusterMembershipConnector implements ClusterMembershipConnector { @Override public Map<String, ClusterMembershipRevision<ClusterMember>> getClusterMemberSiblings() { return reconciler.getCurrent().getNotStaleClusterMemberSiblings(); } KubeClusterMembershipConnector(ClusterMember initial,
                                          KubeMembershipExecutor kubeMembershipExecutor,
                                          KubeLeaderElectionExecutor kubeLeaderElectionExecutor,
                                          KubeClusterMembershipConfiguration configuration,
                                          TitusRuntime titusRuntime); @PreDestroy void shutdown(); @Override ClusterMembershipRevision<ClusterMember> getLocalClusterMemberRevision(); @Override Map<String, ClusterMembershipRevision<ClusterMember>> getClusterMemberSiblings(); @Override ClusterMembershipRevision<ClusterMemberLeadership> getLocalLeadershipRevision(); @Override Optional<ClusterMembershipRevision<ClusterMemberLeadership>> findCurrentLeader(); @Override Mono<ClusterMembershipRevision<ClusterMember>> register(Function<ClusterMember, ClusterMembershipRevision<ClusterMember>> selfUpdate); @Override Mono<ClusterMembershipRevision<ClusterMember>> unregister(Function<ClusterMember, ClusterMembershipRevision<ClusterMember>> selfUpdate); @Override Mono<Void> joinLeadershipGroup(); @Override Mono<Boolean> leaveLeadershipGroup(boolean onlyNonLeader); @Override Flux<ClusterMembershipEvent> membershipChangeEvents();  }
@Test public void testStaleMemberRemoval() throws InterruptedException { ClusterMembershipRevision<ClusterMember> stale = clusterMemberRegistrationRevision(activeClusterMember("staleSibling")).toBuilder() .withTimestamp(1) .build(); kubeExecutors.addOrUpdateSibling(stale); doAddSibling("sibling1"); assertThat(connector.getClusterMemberSiblings()).containsKey("sibling1"); await().until(() -> kubeExecutors.getMemberById("staleSibling").map(r -> "staleSibling").onErrorReturn("null").block().equals("null")); }