KafkaServer {
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
keyTab="/keytabs/kafka.keytab"
storeKey=true
useTicketCache=false
serviceName="kafka" // this will be used to connect to other brokers for replica management and also controller requests. This should be set to whatever principal that kafka brokers are running.
principal="kafka/_HOST@EXAMPLE.COM";
};
Client {
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
keyTab="/keytabs/kafka.keytab"
storeKey=true
useTicketCache=false
serviceName="zookeeper"
principal="kafka@EXAMPLE.COM";
}


KafkaServer will be used to authenticate Kafka broker against kerberos
and Client section will be used for zkClient to access kerberos enabled zookeeper cluster.

KafkaClient {
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
keyTab="/keytabs/kafkaclient.keytab"
storeKey=true
useTicketCache=false
serviceName="kafka"
principal="kafkaproducer/_HOST@EXAMPLE.COM";
};
 
The above config is for any client ( producer, consumer) connecting to kerberos enabled Kafka cluster.
Here serviceName must match the principal name used under KafkaServer.
 

package org.apache.kafka.common.network;
 
public class Channel implements ScatteringByteChannel, GatheringByteChannel {
   private TransportLayer transportLayer;
   private Authenticator authenticator;

   public Channel(TransportLayer transportLayer, Authenticator authenticator) throws IOException 
  
   /**
    * returns user principal for the session
    * Incase of PLAINTEXT and No Authentication returns ANONYMOUS as the userPrincipal
    * If SSL used without any SASL Authentication returns SSLSession.peerPrincipal
    */
    public UserPrincipal userPrincipal() {
       return authenticator.userPrincipal();
    }

    /**
     * starts transportLayer handshake
     * If transportLayer handshake finishes, than initiates AuthenticationLayer.authenticate
     * @returns 0 if handshake and authentication finishes otherwise returns appropriate SelctionKey.OP
     */
    public int connect(boolean read, boolean write) throws IOException 
    
    /**
     * 
     */
    public void blockingConnect(long timeout) throws IOException
  
    @Override
    public int write(ByteBuffer src) throws IOException
 
    @Override
    public int read(ByteBuffer dst) throws IOException
 
    /* returns the socketChannel */
    public SocketChannel socketChannel();
 
	@Override
    public long write(ByteBuffer[] srcs) throws IOException 
   
    @Override
    public long write(ByteBuffer[] srcs, int offset, int length) throws IOException

    @Override
    public int read(ByteBuffer dst) throws IOException

    @Override
    public long read(ByteBuffer[] dsts) throws IOException 

    @Override
    public long read(ByteBuffer[] dsts, int offset, int length) throws IOException

    public boolean finishConnect() throws IOException
    
    public DataInputStream getInputStream() throws IOException

    public DataOutputStream getOutputStream() throws IOException

    public void close() throws IOException
    
 }

public interface TransportLayer {

    /**
     * Closes this channel
     *
     * @throws IOException If and I/O error occurs
     */
    void close() throws IOException;


    /**
     * Tells wheather or not this channel is open.
     */
    boolean isOpen();

    /**
     * Writes a sequence of bytes to this channel from the given buffer.
     */
    int write(ByteBuffer src) throws IOException;

    long write(ByteBuffer[] srcs) throws IOException;

    long write(ByteBuffer[] srcs, int offset, int length) throws IOException;

    int read(ByteBuffer dst) throws IOException;

    long read(ByteBuffer[] dsts) throws IOException;

    long read(ByteBuffer[] dsts, int offset, int length) throws IOException;

    boolean isReady();

    boolean finishConnect() throws IOException;

    SocketChannel socketChannel();

    /**
     * Performs SSL handshake hence is a no-op for the non-secure
     * implementation
     * @param read Unused in non-secure implementation
     * @param write Unused in non-secure implementation
     * @return Always return 0
     * @throws IOException
    */
    int handshake(boolean read, boolean write) throws IOException;

    DataInputStream inStream() throws IOException;

    DataOutputStream outStream() throws IOException;

    boolean flush(ByteBuffer buffer) throws IOException;

    Principal getPeerPrincipal();
}

public class PlainTextTransportLayer implements TransportLayer {
   public PlainTextTransportLayer(SocketChannel socketChannel) throws IOException
}  
 

public class SSLTransportLayer implements TransportLayer {
	public SSLTransportLayer(SocketChannel socketChannel, SSLEngine sslEngine) throws IOException 
}  
 

public interface Authenticator {

    /**
     * Closes any resources
     *
     * @throws IOException if any I/O error occurs
     */
    void close() throws IOException;

    /**
     *
     * @throws IOException
     */
    void init() throws IOException;

    /**
     * Returns UserPrincipal after authentication is established
     */
    UserPrincipal userPrincipal();


    /**
     * Does authentication in non-blocking way and returns SelectionKey.OP if further communication needed
     */
    int authenticate(boolean read, boolean write) throws IOException;

    /**
     * returns true if authentication is complete otherwise returns false;
     */

    boolean isComplete();

}
 
 

public class DefaultAuthenticator implements Authenticator {

    TransportLayer transportLayer;

    public DefaultAuthenticator(TransportLayer transportLayer) {
        this.transportLayer = transportLayer;
    }

    public void init() {}

    public int authenticate(boolean read, boolean write) throws IOException {
        return 0;
    }

    /** returns peer host incase of SSL */
    public UserPrincipal userPrincipal() {
        return new UserPrincipal(transportLayer.getPeerPrincipal().toString());
    }

    public void close() throws IOException {}

    public boolean isComplete() {
        return true;
    }
}
  
 

public class SaslServerAuthenticator implements Authenticator {

    public SaslServerAuthenticator(final Subject subject, TransportLayer transportLayer) {
    }
}
  
 

public class SaslServerAuthenticator implements Authenticator {
   public SaslServerAuthenticator(final Subject subject, TransportLayer transportLayer) {
   }
}
  
 

public class AuthUtils {
   /**
     * Construct a JAAS configuration object per kafka jaas configuration file
     * @param storm_conf Storm configuration
     * @return JAAS configuration object
     */
    public static Configuration getConfiguration(String jaasConfigFilePath)
 
    public static String getJaasConfig(String loginContextName, String key) throws IOException
 
    public static String getDefaultRealm()
 
}
 
 

package org.apache.kafka.common.security.kerberos;
 
public class Login {
private volatile Subject subject = null;
 
//logs in based on jaas conf and starts a thread to renew it .
public Login(final String loginContextName)
            throws LoginException 
 
// returns generated subject after the login.
public Subject getSubject() 


}
 

package kafka.common
 
object KerberosLoginManager {
   var kerberosLogin: Login 
 
   //does kerberos login and generates the subject
   def init()
 
   // returns kerberosLogin.getSubject()
   def subject()
 
   def shutdown()
   
}

sasl.authentication.mechanism (KERBEROS will be supported for revision1) 
ssl.protocol
ssl.cipher.suites
ssl.enabled.protocols
ssl.keystore.type
ssl.keystore.location
ssl.keystore.password
ssl.key.password
ssl.truststore.type
ssl.truststore.location
ssl.truststore.password
ssl.client.require.cert
ssl.keymanager.algorithm
ssl.trustmanager.algorithm