public static boolean areKerberosCredentialsValid(
            UserGroupInformation ugi, boolean useTicketCache) {
        Preconditions.checkState(isKerberosSecurityEnabled(ugi));

        // note: UGI::hasKerberosCredentials inaccurately reports false
        // for logins based on a keytab (fixed in Hadoop 2.6.1, see HADOOP-10786),
        // so we check only in ticket cache scenario.
        if (useTicketCache && !ugi.hasKerberosCredentials()) {
            if (hasHDFSDelegationToken(ugi)) {
                LOG.warn(
                        "Hadoop security is enabled but current login user does not have Kerberos credentials, "
                                + "use delegation token instead. Flink application will terminate after token expires.");
                return true;
            } else {
                LOG.error(
                        "Hadoop security is enabled, but current login user has neither Kerberos credentials "
                                + "nor delegation tokens!");
                return false;
            }
        }

        return true;
    }
@Test
    public void testShouldReturnTrueWhenDelegationTokenIsPresent() {
        UserGroupInformation.setConfiguration(
                getHadoopConfigWithAuthMethod(AuthenticationMethod.KERBEROS));
        UserGroupInformation userWithoutCredentialsButHavingToken =
                createTestUser(AuthenticationMethod.KERBEROS);
        userWithoutCredentialsButHavingToken.addToken(getHDFSDelegationToken());
        assumeFalse(userWithoutCredentialsButHavingToken.hasKerberosCredentials());

        boolean result =
                HadoopUtils.areKerberosCredentialsValid(userWithoutCredentialsButHavingToken, true);

        assertTrue(result);
    }