class StaticMasterReplicaConnector<K, V> implements MasterReplicaConnector<K, V> {

    private final RedisClient redisClient;

    private final RedisCodec<K, V> codec;

    private final Iterable<RedisURI> redisURIs;

    StaticMasterReplicaConnector(RedisClient redisClient, RedisCodec<K, V> codec, Iterable<RedisURI> redisURIs) {
        this.redisClient = redisClient;
        this.codec = codec;
        this.redisURIs = redisURIs;
    }

    @Override
    public CompletableFuture<StatefulRedisMasterReplicaConnection<K, V>> connectAsync() {

        Map<RedisURI, StatefulRedisConnection<K, V>> initialConnections = new HashMap<>();

        TopologyProvider topologyProvider = new StaticMasterReplicaTopologyProvider(redisClient, redisURIs);

        RedisURI seedNode = redisURIs.iterator().next();

        MasterReplicaTopologyRefresh refresh = new MasterReplicaTopologyRefresh(redisClient, topologyProvider);
        MasterReplicaConnectionProvider<K, V> connectionProvider = new MasterReplicaConnectionProvider<>(redisClient, codec,
                seedNode, initialConnections);

        return refresh.getNodes(seedNode).flatMap(nodes -> {

            EventRecorder.getInstance().record(new MasterReplicaTopologyChangedEvent(seedNode, nodes));

            if (nodes.isEmpty()) {
                return Mono.error(new RedisException(String.format("Cannot determine topology from %s", redisURIs)));
            }

            return initializeConnection(codec, seedNode, connectionProvider, nodes);
        }).onErrorMap(ExecutionException.class, Throwable::getCause).toFuture();
    }

    private Mono<StatefulRedisMasterReplicaConnection<K, V>> initializeConnection(RedisCodec<K, V> codec, RedisURI seedNode,
            MasterReplicaConnectionProvider<K, V> connectionProvider, List<RedisNodeDescription> nodes) {<FILL_FUNCTION_BODY>}

}


        connectionProvider.setKnownNodes(nodes);

        MasterReplicaChannelWriter channelWriter = new MasterReplicaChannelWriter(connectionProvider, redisClient.getResources(), redisClient.getOptions());

        StatefulRedisMasterReplicaConnectionImpl<K, V> connection = new StatefulRedisMasterReplicaConnectionImpl<>(
                channelWriter, codec, seedNode.getTimeout());
        connection.setOptions(redisClient.getOptions());

        return Mono.just(connection);
    
<no_super_class>