class ActorUtils {

    public void broadcast(final Function<Short, Object> messageSupplier, final Class<?> messageClass) {
        for (final String shardName : configuration.getAllShardNames()) {

            Future<PrimaryShardInfo> primaryFuture = findPrimaryShardAsync(shardName);
            primaryFuture.onComplete(new OnComplete<PrimaryShardInfo>() {
                @Override
                public void onComplete(final Throwable failure, final PrimaryShardInfo primaryShardInfo) {
                    if (failure != null) {
                        LOG.warn("broadcast failed to send message {} to shard {}", messageClass.getSimpleName(),
                            shardName, failure);
                    } else {
                        Object message = messageSupplier.apply(primaryShardInfo.getPrimaryShardVersion());
                        primaryShardInfo.getPrimaryShardActor().tell(message, ActorRef.noSender());
                    }
                }
            }, getClientDispatcher());
        }
    }

    public  ActorUtils(final ActorSystem actorSystem, final ActorRef shardManager,
            final ClusterWrapper clusterWrapper, final Configuration configuration);
    public  ActorUtils(final ActorSystem actorSystem, final ActorRef shardManager,
            final ClusterWrapper clusterWrapper, final Configuration configuration,
            final DatastoreContext datastoreContext, final PrimaryShardInfoFutureCache primaryShardInfoCache);

}