@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);
        }
    }
@SuppressWarnings("unchecked")
    @Test
    public void validOptionsWithExplicitOptionValues()
            throws IOException, UnsupportedCallbackException {
        String explicitScope1 = "scope1";
        String explicitScope2 = "scope2";
        String explicitScopeClaimName = "putScopeInHere";
        String principalClaimName = "principal";
        final String[] scopeClaimNameOptionValues = {null, explicitScopeClaimName};
        for (String scopeClaimNameOptionValue : scopeClaimNameOptionValues) {
            Map<String, String> options = new HashMap<>();
            String user = "user";
            options.put("unsecuredLoginStringClaim_" + principalClaimName, user);
            options.put("unsecuredLoginListClaim_" + "list", ",1,2,");
            options.put("unsecuredLoginListClaim_" + "emptyList1", "");
            options.put("unsecuredLoginListClaim_" + "emptyList2", ",");
            options.put("unsecuredLoginNumberClaim_" + "number", "1");
            long lifetimeSeconds = 10000;
            options.put("unsecuredLoginLifetimeSeconds", String.valueOf(lifetimeSeconds));
            options.put("unsecuredLoginPrincipalClaimName", principalClaimName);
            if (scopeClaimNameOptionValue != null)
                options.put("unsecuredLoginScopeClaimName", scopeClaimNameOptionValue);
            String actualScopeClaimName = scopeClaimNameOptionValue == null ? "scope" : explicitScopeClaimName;
            options.put("unsecuredLoginListClaim_" + actualScopeClaimName,
                    String.format("|%s|%s", explicitScope1, explicitScope2));
            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, lifetimeSeconds * 1000);
            Map<String, Object> claims = jws.claims();
            assertEquals(new HashSet<>(Arrays.asList(actualScopeClaimName, principalClaimName, "iat", "exp", "number",
                    "list", "emptyList1", "emptyList2")), claims.keySet());
            assertEquals(new HashSet<>(Arrays.asList(explicitScope1, explicitScope2)),
                    new HashSet<>((List<String>) claims.get(actualScopeClaimName)));
            assertEquals(new HashSet<>(Arrays.asList(explicitScope1, explicitScope2)), jws.scope());
            assertEquals(1.0, jws.claim("number", Number.class));
            assertEquals(Arrays.asList("1", "2", ""), jws.claim("list", List.class));
            assertEquals(Collections.emptyList(), jws.claim("emptyList1", List.class));
            assertEquals(Collections.emptyList(), jws.claim("emptyList2", List.class));
        }
    }