class LogWatchCallback implements LogWatch, AutoCloseable {

  private static final Logger LOGGER = LoggerFactory.getLogger(LogWatchCallback.class);

  private final OutputStream out;
  private WritableByteChannel outChannel;
  private volatile InputStream output;

  private final AtomicBoolean closed = new AtomicBoolean(false);
  private final CompletableFuture<AsyncBody> asyncBody = new CompletableFuture<>();
  private final SerialExecutor serialExecutor;

  public LogWatchCallback(OutputStream out, OperationContext context) {
    this.out = out;
    if (out != null) {
      outChannel = Channels.newChannel(out);
    }
    this.serialExecutor = new SerialExecutor(context.getExecutor());
  }

  @Override
  public void close() {
    cleanUp();
  }

  private void cleanUp() {
    if (!closed.compareAndSet(false, true)) {
      return;
    }
    asyncBody.thenAccept(AsyncBody::cancel);
    serialExecutor.shutdownNow();
  }

  public LogWatchCallback callAndWait(HttpClient client, URL url) {<FILL_FUNCTION_BODY>}

  @Override
  public InputStream getOutput() {
    return output;
  }

  public void onFailure(Throwable u) {
    //If we have closed the watch ignore everything
    if (closed.get()) {
      return;
    }

    LOGGER.error("Log Callback Failure.", u);
    cleanUp();
  }

}

    HttpRequest request = client.newHttpRequestBuilder().url(url).build();

    if (out == null) {
      // we can pass the input stream directly to the consumer
      client.sendAsync(request, InputStream.class).whenComplete((r, e) -> {
        if (e != null) {
          onFailure(e);
        }
        if (r != null) {
          this.output = r.body();
        }
      }).join();
    } else {
      // we need to write the bytes to the given output
      // we don't know if the write will be blocking, so hand it off to another thread
      client.consumeBytes(request, (buffers, a) -> CompletableFuture.runAsync(() -> {
        for (ByteBuffer byteBuffer : buffers) {
          try {
            outChannel.write(byteBuffer);
          } catch (IOException e1) {
            throw KubernetesClientException.launderThrowable(e1);
          }
        }
      }, serialExecutor).whenComplete((v, t) -> {
        if (t != null) {
          a.cancel();
          onFailure(t);
        } else if (!closed.get()) {
          a.consume();
        } else {
          a.cancel();
        }
      })).whenComplete((a, e) -> {
        if (e != null) {
          onFailure(e);
        }
        if (a != null) {
          asyncBody.complete(a.body());
          a.body().consume();
          a.body().done().whenComplete((v, t) -> CompletableFuture.runAsync(() -> {
            if (t != null) {
              onFailure(t);
            } else {
              cleanUp();
            }
          }, serialExecutor));
        }
      });
    }

    return this;
  
<no_super_class>