class KeyValueBucketLoader extends BaseBucketLoader {

  @Override
  protected Mono<byte[]> discoverConfig(final NodeIdentifier seed, final String bucket) {
    final CoreContext ctx = core().context();

    return Mono.defer(() -> {
      CarrierBucketConfigRequest request = new CarrierBucketConfigRequest(
        ctx.environment().timeoutConfig().connectTimeout(),
        ctx,
        new CollectionIdentifier(bucket, Optional.empty(), Optional.empty()),
        BestEffortRetryStrategy.INSTANCE,
        seed
      );
      core().send(request);
      return Reactor.wrap(request, request.response(), true);
    })
    .map(response -> {
      if (response.status().success()) {
        return response.content();
      } else if (response.status() == ResponseStatus.UNSUPPORTED) {
        throw new UnsupportedConfigMechanismException();
      } else {
        throw new ConfigException("Received error status from KeyValueBucketLoader: " + response);
      }
    });
  }

  public  KeyValueBucketLoader(final Core core);

}