In some situations MetadataManager.SingleThreaded.startSchemaRequest could fail to set CompletableFuture arg

private void startSchemaRequest(CompletableFuture<RefreshSchemaResult> refreshFuture) {
  assert adminExecutor.inEventLoop();
  if (closeWasCalled) {
    refreshFuture.complete(new RefreshSchemaResult(metadata));
    return;
  }
  if (currentSchemaRefresh == null) {
    currentSchemaRefresh = refreshFuture;
    LOG.debug("[{}] Starting schema refresh", logPrefix);
    initControlConnectionForSchema()
        .thenCompose(v -> context.getTopologyMonitor().checkSchemaAgreement())
        .whenComplete(
            (schemaInAgreement, agreementError) -> {
              if (agreementError != null) {
                refreshFuture.completeExceptionally(agreementError);
              } else {
                schemaQueriesFactory
                    .newInstance()
                    .execute()
                    .thenApplyAsync(this::parseAndApplySchemaRows, adminExecutor)
                    .whenComplete(
                        (newMetadata, metadataError) -> {
                          if (metadataError != null) {
                            refreshFuture.completeExceptionally(metadataError);
                          } else {
                            refreshFuture.complete(
                                new RefreshSchemaResult(newMetadata, schemaInAgreement));
                          }

...

@Override
public SchemaQueries newInstance() {
  DriverChannel channel = context.getControlConnection().channel();
  if (channel == null || channel.closeFuture().isDone()) {
    throw new IllegalStateException("Control channel not available, aborting schema refresh");
  }
  Node node =
      context
          .getMetadataManager()
          .getMetadata()
          .findNode(channel.getEndPoint())
          .orElseThrow(
              () ->
                  new IllegalStateException(
                      "Could not find control node metadata "
                          + channel.getEndPoint()
                          + ", aborting schema refresh"));
  return newInstance(node, channel);
}