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="/vagrant/keytabs/storm.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/kafka.keytab"
storeKey=true
useTicketCache=true
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 ReadableByteChannel, GatheringByteChannel {
   private UserPrincipal userPrincipal = new UserPrincipal("ANONYMOUS");
   
   public Channel(SocketChannel socketChannel) throws IOException 
  
   /**
     * returns user principal for the session
     * Incase of PLAINTEXT returns ANONYMOUS as the UserPrincipal
     */
    public UserPrincipal getUserPrincipal() {
        return userPrincipal;
    }
    
    @Override
    public int write(ByteBuffer src) throws IOException
 
    @Override
    public int read(ByteBuffer dst) throws IOException
 
    /* returns the socketChannel */
    public SocketChannel getIOChannel()
 
	/* incase of PLAINTEXT channel this always returns true */
    public boolean isHandshakeComplete()
 
     /**
     * Performs SSL or GSS-API 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
    */
    public int handshake(boolean read, boolean write) throws IOException {
        return 0;
    }


}

package org.apache.kafka.common.network;
 
public class GSSChannel extends Channel {


 public GSSChannel(SocketChannel socketChannel, Subject subject) throws IOException 


@Overrride
public UserPrincipal getUserPrincipal()


 @Override
 public int handshake(boolean read, boolean write) throws IOException 


 @Override
 public int read(ByteBuffer dst) throws IOException 


 @Override
 public int write(ByteBuffer src) throws IOException 



 @Override
 public long write(ByteBuffer[] srcs, int offset, int length) throws IOException 


@Override
 public long write(ByteBuffer[] srcs) throws IOException


@Override
public int handshake(boolean write, boolean read) throws IOException


@Override
 public boolean isHandshakeComplete() 


}

public class GSSServerChannel extends GSSChannel {
   public GSSServerChannel(SocketChannel socketChannel, Subject subject) throws IOException 
   
   @Override
    public int handshake(boolean read, boolean write) throws IOException
}
 
 

public class GSSBlockingClientChannel extends GSSChannel {
   public GSSBlockingClientChannel(SocketChannel socketChannel, Subject subject) throws IOException 
   
   @Override
    public int handshake(boolean read, boolean write) throws IOException
}
 
 

public class GSSClientChannel extends GSSChannel {
   public GSSClientChannel(SocketChannel socketChannel, Subject subject) throws IOException 
   
   @Override
    public int handshake(boolean read, boolean write) throws IOException
}
 
 

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()
   
}

package kafka.common
 
/*
 * In case of PLAINTEXT encryptionEnable will be false and for SSL encryptionEnable will always be true
 * For KERBEROS clients can request encryption or not
 */
case class Protocol(securityProtocol: SecurityProtocol, encryptionEnable: Boolean)

class BlockingChannel( val host: String,
                       val port: Int,
                       val readBufferSize: Int,
                       val writeBufferSize: Int,
                       val readTimeoutMs: Int,
                       val protocol: Protocol = Protocol(SecurityProtocol.PLAINTEXT, false))