kafka 连接配置

kafka kerberos

kafka适配kerberos认证, 重要的是在jaas的链接信息里把keytab, princial, krb conf加上, 然后加上其他常规的配置, 就可以按照正常的方式去连接了.

参数里还配置了需要连接的service服务是kafka, 毕竟获取ts/tgt密钥需要服务认可, 另外登陆的身份用户也是kafka.

keytab和principal需要从本地或者远程获取, KerberosUtil只是一个本地工具类.

public class KafkaConfig extends Configuration {
    
    private static final String BOOTSTRAP_SERVER_CONFIG    = "bootstrap.servers";
    private static final String CONSUMER_GROUP_ID_PROPERTY = "group.id";
    private static final String SECURITY_PROTOCOL_CONFIG   = "security.protocol";
    private static final String SASL_MECHANISM_CONFIG      = "sasl.mechanism";
    private static final String SASL_KERBEROS_SERVICE_NAME       = "sasl.kerberos.service.name";
    private static final String SASL_JAAS_CONFIG       = "sasl.jaas.config";
    private static final String SASL_MECHANISM_GSSAPI       = "GSSAPI";

    public void setKerberosProperties(Properties properties) {
        // kerberos
        System.setProperty("java.security.krb5.conf", "/etc/krb5.conf");
        String keytab = null;
        String principal = null;
        try {
            keytab = KerberosUtil.getKeytabRemote("kafka", kdcProxyFeignApi);
            principal = KerberosUtil.getPrincipalByUsername("kafka", kdcProxyFeignApi);
        } catch (Exception e) {
            log.error("failed to get kerberos credential info", e);
        }
        String jaasConfig = "com.sun.security.auth.module.Krb5LoginModule required " +
                "useKeyTab=true " +
                "storeKey=true " +
                "keyTab=\"" + keytab + "\" " +
                "principal=\"" + principal + "\";";
        String kerberosServiceName = "kafka";
        properties.put(SASL_KERBEROS_SERVICE_NAME, kerberosServiceName);
        properties.put(SASL_JAAS_CONFIG, jaasConfig);
    }
}