@Override
        public CompletableFuture<FetchResult> fetch(FetchContext context, long startOffset, long endOffset,
            int maxBytesHint) {
            CompletableFuture<FetchResult> cf = new CompletableFuture<>();
            Timeout timeout = fetchTimeout.newTimeout(t -> LOGGER.warn("fetch timeout, stream[{}] [{}, {})", streamId(), startOffset, endOffset), 1, TimeUnit.MINUTES);
            stream.fetch(context, startOffset, endOffset, maxBytesHint).whenComplete((rst, e) -> FutureUtil.suppress(() -> {
                timeout.cancel();
                Throwable ex = FutureUtil.cause(e);
                if (ex != null) {
                    if (!(ex instanceof FastReadFailFastException)) {
                        LOGGER.error("Fetch stream[{}] [{},{}) fail", streamId(), startOffset, endOffset, ex);
                    }
                    cf.completeExceptionally(ex);
                } else {
                    cf.complete(rst);
                }
            }, LOGGER));
            return cf;
        }
Maybe halt and complete the waiting future if the exception is a ElasticStreamClientException with a halt error code.

@param t             the exception
@param waitingFuture the waiting future
@return true if the waiting future is completed, otherwise false
core/src/main/scala/kafka/log/streamaspect/ClientWrapper.java