private void trim0(long newStartOffset, CompletableFuture<Void> cf) {
            stream.trim(newStartOffset).whenCompleteAsync((rst, ex) -> {
                FutureUtil.suppress(() -> {
                    if (ex != null) {
                        if (!maybeHaltAndCompleteWaitingFuture(ex, cf)) {
                            LOGGER.error("Trim stream[{}] (new offset = {}) failed, retry later", streamId(), newStartOffset, ex);
                            generalRetryScheduler.schedule(() -> trim0(newStartOffset, cf), 3, TimeUnit.SECONDS);
                        }
                    } else {
                        cf.complete(rst);
                    }
                }, LOGGER);
            }, generalCallbackExecutors);
        }
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