/**
 * Builds a cluster service.
 */
protected static ManagedClusterMembershipService buildClusterMembershipService(
    ClusterConfig config,
    BootstrapService bootstrapService,
    NodeDiscoveryProvider discoveryProvider,
    GroupMembershipProtocol membershipProtocol,
    Version version) {
  // If the local node has not be configured, create a default node.
  Member localMember = Member.builder()
      .withId(config.getNodeConfig().getId())
      .withAddress(config.getNodeConfig().getAddress())
      .withHostId(config.getNodeConfig().getHostId())
      .withRackId(config.getNodeConfig().getRackId())
      .withZoneId(config.getNodeConfig().getZoneId())
      .withProperties(config.getNodeConfig().getProperties())
      .build();
  return new DefaultClusterMembershipService(
      localMember,
      version,
      new DefaultNodeDiscoveryService(bootstrapService, localMember, discoveryProvider),
      bootstrapService,
      membershipProtocol);
}
 

private SwimMembershipProtocol startProtocol(Member member) {
  SwimMembershipProtocol protocol = new SwimMembershipProtocol(new SwimMembershipProtocolConfig()
      .setFailureTimeout(Duration.ofSeconds(2)));
  TestGroupMembershipEventListener listener = new TestGroupMembershipEventListener();
  listeners.put(member.id(), listener);
  protocol.addListener(listener);
  BootstrapService bootstrap = new TestBootstrapService(
      messagingServiceFactory.newMessagingService(member.address()).start().join(),
      unicastServiceFactory.newUnicastService(member.address()).start().join(),
      broadcastServiceFactory.newBroadcastService().start().join());
  NodeDiscoveryProvider provider = new BootstrapDiscoveryProvider(nodes);
  provider.join(bootstrap, member).join();
  NodeDiscoveryService discovery = new DefaultNodeDiscoveryService(bootstrap, member, provider).start().join();
  protocol.join(bootstrap, discovery, member).join();
  protocols.put(member.id(), protocol);
  return protocol;
}
 

/**
 * Creates a Raft server.
 */
private RaftServer createServer(Member member, List<Node> members) {
  RaftServerProtocol protocol;
  ManagedMessagingService messagingService;
  if (USE_NETTY) {
    messagingService = (ManagedMessagingService) new NettyMessagingService("test", member.address(), new MessagingConfig())
        .start()
        .join();
    messagingServices.add(messagingService);
    protocol = new RaftServerMessagingProtocol(messagingService, PROTOCOL_SERIALIZER, addressMap::get);
  } else {
    protocol = protocolFactory.newServerProtocol(member.id());
  }

  BootstrapService bootstrapService = new BootstrapService() {
    @Override
    public MessagingService getMessagingService() {
      return messagingService;
    }

    @Override
    public UnicastService getUnicastService() {
      return new UnicastServiceAdapter();
    }

    @Override
    public BroadcastService getBroadcastService() {
      return new BroadcastServiceAdapter();
    }
  };

  RaftServer.Builder builder = RaftServer.builder(member.id())
      .withProtocol(protocol)
      .withThreadModel(ThreadModel.SHARED_THREAD_POOL)
      .withMembershipService(new DefaultClusterMembershipService(
          member,
          Version.from("1.0.0"),
          new DefaultNodeDiscoveryService(bootstrapService, member, new BootstrapDiscoveryProvider(members)),
          bootstrapService,
          new HeartbeatMembershipProtocol(new HeartbeatMembershipProtocolConfig())))
      .withStorage(RaftStorage.builder()
          .withStorageLevel(StorageLevel.DISK)
          .withDirectory(new File(String.format("target/perf-logs/%s", member.id())))
          .withNamespace(STORAGE_NAMESPACE)
          .withMaxSegmentSize(1024 * 1024 * 64)
          .withDynamicCompaction()
          .withFlushOnCommit(false)
          .build());

  RaftServer server = builder.build();
  servers.add(server);
  return server;
}