public static Optional<Throwable> findThrowableOfThrowableType(
            Throwable throwable, ThrowableType throwableType) {
        if (throwable == null || throwableType == null) {
            return Optional.empty();
        }

        Throwable t = throwable;
        while (t != null) {
            final ThrowableAnnotation annotation =
                    t.getClass().getAnnotation(ThrowableAnnotation.class);
            if (annotation != null && annotation.value() == throwableType) {
                return Optional.of(t);
            } else {
                t = t.getCause();
            }
        }

        return Optional.empty();
    }
@Test
    void testFindThrowableOfThrowableType() {
        // no throwable type
        assertThat(
                        ThrowableClassifier.findThrowableOfThrowableType(
                                new Exception(), ThrowableType.RecoverableError))
                .isNotPresent();

        // no recoverable throwable type
        assertThat(
                        ThrowableClassifier.findThrowableOfThrowableType(
                                new TestPartitionDataMissingErrorException(),
                                ThrowableType.RecoverableError))
                .isNotPresent();

        // direct recoverable throwable
        assertThat(
                        ThrowableClassifier.findThrowableOfThrowableType(
                                new TestRecoverableErrorException(),
                                ThrowableType.RecoverableError))
                .isPresent();

        // nested recoverable throwable
        assertThat(
                        ThrowableClassifier.findThrowableOfThrowableType(
                                new Exception(new TestRecoverableErrorException()),
                                ThrowableType.RecoverableError))
                .isPresent();

        // inherit recoverable throwable
        assertThat(
                        ThrowableClassifier.findThrowableOfThrowableType(
                                new TestRecoverableFailureSubException(),
                                ThrowableType.RecoverableError))
                .isPresent();
    }