public String failureMessage(
        int epoch,
        OptionalLong deltaUs,
        boolean isActiveController,
        long lastCommittedOffset
    ) {
        StringBuilder bld = new StringBuilder();
        if (deltaUs.isPresent()) {
            bld.append("event failed with ");
        } else {
            bld.append("event unable to start processing because of ");
        }
        bld.append(internalException.getClass().getSimpleName());
        if (externalException.isPresent()) {
            bld.append(" (treated as ").
                append(externalException.get().getClass().getSimpleName()).append(")");
        }
        if (causesFailover()) {
            bld.append(" at epoch ").append(epoch);
        }
        if (deltaUs.isPresent()) {
            bld.append(" in ").append(deltaUs.getAsLong()).append(" microseconds");
        }
        if (causesFailover()) {
            if (isActiveController) {
                bld.append(". Renouncing leadership and reverting to the last committed offset ");
                bld.append(lastCommittedOffset);
            } else {
                bld.append(". The controller is already in standby mode");
            }
        }
        bld.append(".");
        if (!isFault && internalException.getMessage() != null) {
            bld.append(" Exception message: ");
            bld.append(internalException.getMessage());
        }
        return bld.toString();
    }
Returns true if the class and message fields match for two exceptions. Handles nulls.
metadata/src/main/java/org/apache/kafka/controller/errors/EventHandlerExceptionInfo.java