KubeClusterMembershipConnector implements ClusterMembershipConnector { @Override public Mono<ClusterMembershipRevision<ClusterMember>> register(Function<ClusterMember, ClusterMembershipRevision<ClusterMember>> selfUpdate) { return reconciler.apply(current -> KubeRegistrationActions.registerLocal(context, current, selfUpdate).map(f -> f.apply(current))) .map(KubeClusterState::getLocalMemberRevision); } 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 testRegistrationConnectionErrorInRegistrationRequest() throws InterruptedException { kubeExecutors.failOnMembershipUpdate(new RuntimeException("Simulated membership update error"), 1); try { connector.register(ClusterMemberGenerator::clusterMemberRegistrationRevision).block(); fail("Failure expected"); } catch (Exception ignore) { } doRegister(); }