class DistributedShardedDOMDataTree implements DOMDataTreeService, DOMDataTreeShardingService,
        DistributedShardFactory {

    @Override
    public CompletionStage<DistributedShardRegistration> createDistributedShard(
            final DOMDataTreeIdentifier prefix, final Collection<MemberName> replicaMembers)
            throws DOMDataTreeShardingConflictException {

        synchronized (shards) {
            final DOMDataTreePrefixTableEntry<DOMDataTreeShardRegistration<DOMDataTreeShard>> lookup =
                    shards.lookup(prefix);
            if (lookup != null && lookup.getValue().getPrefix().equals(prefix)) {
                throw new DOMDataTreeShardingConflictException(
                        "Prefix " + prefix + " is already occupied by another shard.");
            }
        }

        final PrefixedShardConfigWriter writer = writerMap.get(prefix.getDatastoreType());

        final ListenableFuture<Void> writeFuture =
                writer.writeConfig(prefix.getRootIdentifier(), replicaMembers);

        final Promise<DistributedShardRegistration> shardRegistrationPromise = akka.dispatch.Futures.promise();
        Futures.addCallback(writeFuture, new FutureCallback<Void>() {
            @Override
            public void onSuccess(final Void result) {

                final Future<Object> ask =
                        Patterns.ask(shardedDataTreeActor, new LookupPrefixShard(prefix), SHARD_FUTURE_TIMEOUT);

                shardRegistrationPromise.completeWith(ask.transform(
                        new Mapper<Object, DistributedShardRegistration>() {
                            @Override
                            public DistributedShardRegistration apply(final Object parameter) {
                                return new DistributedShardRegistrationImpl(
                                        prefix, shardedDataTreeActor, DistributedShardedDOMDataTree.this);
                            }
                        },
                        new Mapper<Throwable, Throwable>() {
                            @Override
                            public Throwable apply(final Throwable throwable) {
                                return new DOMDataTreeShardCreationFailedException(
                                        "Unable to create a cds shard.", throwable);
                            }
                        }, actorSystem.dispatcher()));
            }

            @Override
            public void onFailure(final Throwable throwable) {
                shardRegistrationPromise.failure(
                        new DOMDataTreeShardCreationFailedException("Unable to create a cds shard.", throwable));
            }
        }, MoreExecutors.directExecutor());

        return FutureConverters.toJava(shardRegistrationPromise.future());
    }

    public  DistributedShardedDOMDataTree(final ActorSystemProvider actorSystemProvider,
                                         final DistributedDataStoreInterface distributedOperDatastore,
                                         final DistributedDataStoreInterface distributedConfigDatastore);

}