@Override
    public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException {
        if (!configured())
            throw new IllegalStateException("Callback handler not configured");
        for (Callback callback : callbacks) {
            if (callback instanceof OAuthBearerTokenCallback)
                try {
                    handleTokenCallback((OAuthBearerTokenCallback) callback);
                } catch (KafkaException e) {
                    throw new IOException(e.getMessage(), e);
                }
            else if (callback instanceof SaslExtensionsCallback)
                try {
                    handleExtensionsCallback((SaslExtensionsCallback) callback);
                } catch (KafkaException e) {
                    throw new IOException(e.getMessage(), e);
                }
            else
                throw new UnsupportedCallbackException(callback);
        }
    }
@Test
    public void minimalToken() throws IOException, UnsupportedCallbackException {
        Map<String, String> options = new HashMap<>();
        String user = "user";
        options.put("unsecuredLoginStringClaim_sub", user);
        MockTime mockTime = new MockTime();
        OAuthBearerUnsecuredLoginCallbackHandler callbackHandler = createCallbackHandler(options, mockTime);
        OAuthBearerTokenCallback callback = new OAuthBearerTokenCallback();
        callbackHandler.handle(new Callback[] {callback});
        OAuthBearerUnsecuredJws jws = (OAuthBearerUnsecuredJws) callback.token();
        assertNotNull(jws, "create token failed");
        long startMs = mockTime.milliseconds();
        confirmCorrectValues(jws, user, startMs, 1000 * 60 * 60);
        assertEquals(new HashSet<>(Arrays.asList("sub", "iat", "exp")), jws.claims().keySet());
    }