...

A new configuration property sasl.callback.handlers will be added to enable new callback handlers to be specified for brokers and clients. This will be a list of classes that implement the org.apache.kafka.common.security.auth.AuthCallbackHandler interface. A different handler may be provided for each enabled mechanism.

Callback Handler

Callback handlers will implement AuthCallbackHandler which extends The callback handler interface AuthCallbackHandler will extend the standard javax.security.auth.callback.CallbackHandler interface, enabling the handler to be passed directly to SaslServer/SaslClient implementations. The callback handler configured for a mechanism must include the callbacks as described below:

...

Callback handlers which require additional options at runtime (eg. URL of a credential server) may include arguments in the JAAS configuration using the sasl.jaas.config property (KIP-85). This is similar to the way keytab location is configured for GSSAPI. Client callback handlers can retrieve Subject using Subject.getSubject(AccessController.getContext()) to obtain credentials populated by the login module.

package org.apache.kafka.common.security.auth;
import org.apache.kafka.common.Configurable;
import org.apache.kafka.common.network.Mode;
import java.util.Collection;
import javax.security.auth.callback.CallbackHandler;

public interface AuthCallbackHandler extends Configurable, CallbackHandler {
    /**
     * Returns the connection mode supported by this callback handler
     */
    Mode mode();
    /**
     * Returns the SASL mechanisms supported by this callback handler
     */
    Collection<String> saslMechanisms();
    /**
     * Closes this instance.
     */
    void close();
}

...

package org.apache.kafka.common.security.plain;
import javax.security.auth.callback.Callback;

public class PlainCredentialCallback implements Callback {
    private final String username;
    private final String password;
    private boolean authenticated;
    public PlainCredentialCallback(String username, String password) {
        this.username = username;
        this.password = password;
    }
    public String username() {
        return username;
    }
    public String password() {
        return password;
    }
    public boolean authenticated() {
        return this.authenticated;
    }
    public void authenticated(boolean success) {
        this.authenticated = success;
    }
}

SASL/SCRAM Callback

package org.apache.kafka.common.security.scram;
import javax.security.auth.callback.Callback;
public class ScramCredentialCallback implements Callback {
    private final String username;
    private String salt;
    private byte[] serverKey;
    private byte[] storedKey;
    private int iterationCount;
    public ScramCredentialCallback(String username) {
        this.username = username;
    }
    public String username() {
        return username;
    }
    public String salt() {
        return salt;
    }
    public void salt(String salt) {
        this.salt = salt;
    }
    public byte[] serverKey() {
        return serverKey;
    }
    public void serverKey(byte[] serverKey) {
        this.serverKey = serverKey;
    }
    public byte[] storedKey() {
        return storedKey;
    }
    public void storedKey(byte[] storedKey) {
        this.storedKey = storedKey;
    }
    public int iterationCount() {
        return iterationCount;
    }
    public void iterationCount(int iterationCount) {
        this.iterationCount = iterationCount;
    }
}

...

ChannelBuilder  will create an instance of each configured callback handler using the default constructor. For mechanisms without a callback handler override, the existing default callback handlers (SaslServerCallbackHandler/SaslClientCallbackHandler) will be created once for each mechanism in ChannelBuilder, instead of per-connection. This enables callback handlers using external authentication servers to cache credentials or reuse connections if required. SaslClientCallbackHandler will be modified to obtain subject using Subject using Subject.getSubject(AccessController.getContext()) to avoid the current per-connection state.

...

Configure a new mechanism not included in Kafka using custom SaslServer/SaslClient

Callbacks A handler that handles any callbacks required for these server/client implementations may be specified in sasl.callback.handlers for brokers and clients

...

Callbacks defined for the mechanism in the Java implementation must be handled by custom callback handlers if the behaviour differs from the default callbacks in Kafka.

...

Existing behaviour will be retained as default

Test Plan

Existing integration and system tests will test the default behaviour. Additional unit and integration tests will be added to test the new configuration:

1. Test that PLAIN credential provider can be replaced
2. Test that SCRAM password store can be replaced
3. Test that new mechanisms not included in Kafka can be run with custom callback handlers

Rejected Alternatives

Define a new credential provider interface instead of using CallbackHandler

...