@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 addsExtensions() throws IOException, UnsupportedCallbackException {
        Map<String, String> options = new HashMap<>();
        options.put("unsecuredLoginExtension_testId", "1");
        OAuthBearerUnsecuredLoginCallbackHandler callbackHandler = createCallbackHandler(options, new MockTime());
        SaslExtensionsCallback callback = new SaslExtensionsCallback();

        callbackHandler.handle(new Callback[] {callback});

        assertEquals("1", callback.extensions().map().get("testId"));
    }