public static void abort(Logger log, String what, Throwable exception,
            KafkaFutureImpl<String> doneFuture) throws KafkaException {
        log.warn("{} caught an exception", what, exception);
        if (exception.getMessage() == null || exception.getMessage().isEmpty()) {
            doneFuture.complete(exception.getClass().getCanonicalName());
        } else {
            doneFuture.complete(exception.getMessage());
        }
        throw new KafkaException(exception);
    }
Handle an exception in a TaskWorker.

@param log               The logger to use.
@param what              The component that had the exception.
@param exception         The exception.
@param doneFuture        The TaskWorker's doneFuture
@throws KafkaException   A wrapped version of the exception.
trogdor/src/main/java/org/apache/kafka/trogdor/common/WorkerUtils.java