public static EventHandlerExceptionInfo fromInternal(
        Throwable internal,
        Supplier<OptionalInt> latestControllerSupplier
    ) {
        if (internal instanceof ApiException) {
            // This exception is a standard API error response from the controller, which can pass
            // through without modification.
            return new EventHandlerExceptionInfo(false, false, internal);
        } else if (internal instanceof NotLeaderException) {
            // The controller has lost leadership.
            return new EventHandlerExceptionInfo(false, true, internal,
                ControllerExceptions.newWrongControllerException(latestControllerSupplier.get()));
        } else if (internal instanceof RejectedExecutionException) {
            // The controller event queue is shutting down.
            return new EventHandlerExceptionInfo(false, false, internal,
                new TimeoutException("The controller is shutting down.", internal));
        } else if (internal instanceof BoundedListTooLongException) {
            // The operation could not be performed because it would have created an overly large
            // batch.
            return new EventHandlerExceptionInfo(false, false, internal,
                new PolicyViolationException("Unable to perform excessively large batch " +
                    "operation."));
        } else if (internal instanceof UnexpectedBaseOffsetException) {
            // The active controller picked the wrong end offset for its next batch. It must now
            // fail over. This should be pretty rare.
            return new EventHandlerExceptionInfo(false, true, internal,
                new NotControllerException("Unexpected end offset. Controller will resign."));
        } else if (internal instanceof InterruptedException) {
            // The controller event queue has been interrupted. This normally only happens during
            // a JUnit test that has hung. The test framework sometimes sends an InterruptException
            // to all threads to try to get them to shut down. This isn't the correct way to shut
            // the test, but it may happen if something hung.
            return new EventHandlerExceptionInfo(true, true, internal,
                new UnknownServerException("The controller was interrupted."));
        } else {
            // This is the catch-all case for things that aren't supposed to happen. Null pointer
            // exceptions, illegal argument exceptions, etc. They get translated into an
            // UnknownServerException and a controller failover.
            return new EventHandlerExceptionInfo(true, true, internal,
                new UnknownServerException(internal));
        }
    }
Create an EventHandlerExceptionInfo object from an internal exception.

@param internal                  The internal exception.
@param latestControllerSupplier  A function we can call to obtain the latest leader id.

@return                          The new immutable info object.
metadata/src/main/java/org/apache/kafka/controller/errors/EventHandlerExceptionInfo.java