docs: public void completeExceptionally(long id, Throwable throwable, int topologyId) {
        BaseAckTarget ackTarget = collectorMap.get(id);
        if (ackTarget != null) {
           ackTarget.completeExceptionally(throwable, topologyId);
        }
     }