背景
套件kafka基于社区的0.10.0.1版本开发。主要扩展了认证授权。
认证的修改包括：
1）增加自定义TBDS认证
2）修改社区认证SASL_PLAIN的校验逻辑：增加白名单，非集群内用户或白名单配置项之外的用户不能访问kafka
使用
1.无认证访问（低版本kafka访问方式）
大数据套件已经关闭无认证端。不支持无认证的访问
2.SASL_PLAIN认证访问
命令行访问（console-consumer，console-producer）
生产示例：
$KAFKA_HOME/bin/kafka-console-producer.sh --broker-list broker_ip:6668 --topic first_topic --producer.config ./config/client_sasl.properties
消费示例：
$KAFKA_HOME/bin/kafka-console-consumer.sh --bootstrap-server broker_ip:6668 --topic first_topic --new-consumer --consumer.config ./config/client_sasl.properties --from-beginning
其中，一定要用新版的消费API即加参数--new-consumer ，client_sasl.properties文件位置随意，内容必须包含如下配置：
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
java客户端
步骤1：用新消费者API去访问，在consumer的参数中多加两个认证相关参数，代码片段：
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
props.put(ConsumerConfig.GROUP_ID_CONFIG, group);
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
props.put(security.protocol,"SASL_PLAIN");
props.put(sasl.mechanism, "PLAIN");
KafkaConsumer<string, string="string"> consumer = new KafkaConsumer<string,string>(props);
producer代码类似，也只需要给producer的参数加上红色部分的配置
步骤2：启动client java程序，加jvm参数：
-Djava.security.auth.login.config=/data/home/tbds/kafka/kafka010/config/kafka_client_jaas.conf
kafka_client_jaas.conf内容：
KafkaClient {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="kafka"
password="kafka@Tbds.com";
};
username和password的具体值需要参考broker安装路径下的config/kafka_jaas.conf文件中的值，从JAAS的KafkaServer段中选择一个用户
python客户端
python只支持pykafka访问，API很简单，示例片段：
consumer =KafkaConsumer('monitor_metrics_topic',
group_id='metrics-group',
sasl_plain_username='{{sasl_plain_username}}',
sasl_plain_password='{{sasl_plain_password}}',
sasl_mechanism='{{sasl_mechanism}}',
security_protocol='{{security_protocol}}',
bootstrap_servers=[{{kafka_broker_list}}])
其中各个参数含义很直观。
3.TBDS认证访问
命令行访问
生产示例：
$KAFKA_HOME/bin/kafka-console-producer.sh --broker-list broker_ip:6667 --topic first_topic --producer.config ./config/client_sasl.properties
消费示例：
$KAFKA_HOME/bin/kafka-console-consumer.sh --bootstrap-server broker_ip:6667 --topic first_topic --new-consumer --consumer.config ./config/client_sasl.properties --from-beginning
需要注意的是端口，端口号跟SASL_PLAIN认证的端口是不一样的，即每种认证都有独立的端口。client_sasl.properties文件中的内容：
security.protocol=SASL_TBDS
sasl.mechanism=TBDS
sasl.tbds.secure.id=xxxx
sasl.tbds.secure.key=xxx
其中id和key是用户对应kafka模块的accesskey相关信息，记住一定是kafka模块的，然后一定是在portal里处于enabled状态（默认申请的是disabled，管理员控制使能）
java客户端
跟SASL_PLAIN类似，只是把client_sasl.properties中的内容作为key-val加到客户端类(KafkaConsumer,KafkaProducer)的构造函数properties参数中中.示例代码片段：
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
props.put(ConsumerConfig.GROUP_ID_CONFIG, group);
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
props.put("security.protocol","SASL_TBDS");
props.put("sasl.mechanism", "TBDS");
props.put("sasl.tbds.secure.id","U6hXsHKVGwwBc0dZRBGPjaby1a3VCf0NfMnV");
props.put("sasl.tbds.secure.key", "riglLTEAk2fBCWZdmjvhskKpsvXAn7rj");
KafkaConsumer<string, string="string"> consumer = new KafkaConsumer<string,string>(props);
python客户端
不支持
4.集群外客户端部署
方案1：
1）通过yum源或rpm安装套件版本的kafka到集群外机器：
yum install kafka_2_2_0_0_2041
或
rpm -ivh kafka_2_2_0_0_2041-0.10.0.1-xxxx.xxxx.x86_64.rpm
2）拷贝集群内任一kafka客户端机器/etc/kafka/config目录的内容，替换集群外客户端 对应路径:
集群内客户端节点：
cd /etc/kafka/conf
tar zcvf kafka.tar.gz /etc/kafka/conf/*
集群外客户端节点：
cd /etc/kafka/conf
tar zxvf kafka.tar.gz
方案2：
1.打包集群内任一kafka客户端所在节点的/usr/hdp/2.2.0.0-2041/kafka/，并拷贝到集群外目标客户端节点
2.同方案1的步骤2
5.客户端代码编译打包
1）套件版kafka jar命名规则
套件的kafka是基于社区二次开发命名规则采用"社区版本号-TBDS-套件版本号"的方式命名.例：我们现在基于社区0.10.0.1版本的kafka进行开发，套件版本是4.0.3.3，则我们打出的kafka jar版本为0.10.0.1-TBDS-4.0.3.3，完整的kafka client maven jar文件名为：kafka-clients-0.10.0.1-TBDS-4.0.3.3.jar
2）基于套件提供的maven库开发
（1）拷贝或部署套件提供的maven库到开发者可访问的本地仓库或远程仓库
（2）在客户端maven工程pom引入对应的套件版kafka依赖，以套件4.0.3.3版本为例， 需要在pom中加入的依赖片段（其他版本依次类推）：
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.11</artifactId>
<version>0.10.0.1-TBDS-4.0.3.3</version>
</dependency>
<dependency> <groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>0.10.0.1-TBDS-4.0.3.3</version>
</dependency>
3）基于开源maven库开发 （强烈不建议使用）
这种方式建议不使用。
（1）在客户端maven工程引入对应的社区版本kafka依赖。
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.11</artifactId>
<version>0.10.0.1</version>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>0.10.0.1</version>
</dependency>
（2）编译完成之后，在运行客户端程序运行之前，把套件版的kafka jar加入classpath 中：
kafka_2.11-0.10.0.1-TBDS-4.0.3.3.jar,
kafka-clients-0.10.0.1-TBDS-4.0.3.3.jar
6.运行
运行客户端代码与社区方式无区别