Kafka 2.8新特性
可以不使用zk。使用自我管理的仲裁机制。
增加describe集群的API。
支持SASL、SSL双向TLS身份认证。
根据IP，限制broker connection的创建速率。
JDK 1.8
zookeeper（可以使用DP的zk，但需指定zk的namespace，避免跟DP的kafka冲突。）
kafka自带的zookeeper的启动：nohup ./bin/zookeeper-server-start.sh -daemon config/zookeeper.properties 2>&1 & @XY
一、kafka安装部署
tar -xzf kafka_2.13-2.8.0.tgz
$ cd kafka_2.13-2.8.0/config
1、配置 server.properties
broker.id=0 ##依次递增
listeners=PLAINTEXT://demo2:9092 ##访问端口
advertised.listeners=PLAINTEXT://demo2:9092
zookeeper.connect=localhost:2181
zookeeper.connect=demo2:2181/kafka-2.8 ##设置zk Chroot path。一个zk可以管理多个kafka集群。也称kafka的namespace
log.dirs=/data/kafka/kafka_2.12-2.8.0/kafka-logs
2、启动kafka
./bin/kafka-server-start.sh -daemon ./config/server.properties
3、测试kafka
./bin/kafka-topics.sh --create --zookeeper demo2:2181/kafka-2.8 --replication-factor 1 --partitions 1 --topic test
./bin/kafka-topics.sh --list --zookeeper demo2:2181/kafka-2.8
./bin/kafka-console-producer.sh --broker-list demo2:9092 --topic test
./bin/kafka-console-consumer.sh --bootstrap-server demo2:9092 --topic test --from-beginning
二、Kafka配置kerberos认证 https://blog.csdn.net/man_earth/article/details/87256449
1、准备工作：
(1)、安装无限制的jce
(2)、手动创建每个broker server的pricipals，并导出keytab，然后分发到每个broker server节点。
addprinc -randkey kafka/kafkahost1@EXAMPLE.COM
xst -norandkey -k /opt/kafkahost1/kafka.keytab kafka/kafkahost1@EXAMPLE.COM
xst -norandkey -k /data/kafka/kafka_2.12-2.8.0-krb/config/kafka.keytab kafka/demo2@HADOOP.COM
2、部署zk的kerberos认证（单独下载的zk）
下载apache-zookeeper-3.5.9-bin.tar.gz ,并解压
cp zoo_sample.cfg zoo.cfg
修改zoo.cfg
dataDir=/data/kafka/apache-zookeeper-3.5.9-bin/zk-data
clientPort=12181
admin.serverPort=38080
authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
jaasLoginRenew=3600000
创建zk_server_jaas.conf文件
在zookeeper/conf目录下，新建zk_server_jaas.conf文件（vi zk_server_jaas.conf）
Server {
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
storeKey=true
useTicketCache=false
keyTab=“/data/kafka/apache-zookeeper-3.5.9-bin/conf/zk.keytab”
principal=“zookeeper/demo2@HADOOP.COM”;
};
Client {
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
storeKey=true
useTicketCache=false
keyTab=“/data/kafka/apache-zookeeper-3.5.9-bin/conf/zk.keytab”
principal=“zookeeper/demo2@HADOOP.COM”;
};
修改启动文件zkEnv.sh
启动文件添加如下内容：
export JVMFLAGS=" -Djava.security.auth.login.config=$ZOOCFGDIR/zk_server_jaas.conf "
启动zk: ./zkServer.sh start
3、配置kafka/config/server.properties
listeners=PLAINTEXT://demo2:9093,SASL_PLAINTEXT://demo2:9092
advertised.listeners=SASL_PLAINTEXT://demo2:9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=GSSAPI
sasl.enabled.mechanisms=GSSAPI
sasl.kerberos.service.name=kafka //这里的service.name要跟上面建立的principal相对应，kafka/kafkahost3@EXAMPLE.COM的principal服务名就是kafka
log.dirs=/data/kafka/kafka_2.12-2.8.0-krb/kafka-logs
zookeeper.connect=demo2:12181
3、ACL配置
kafka自带了一个开箱即用的ACL授权工具。它基于zk来存储所有的acl授权信息。可通过authorizer.class.name属性来配置：authorizer.class.name=kafka.security.authorizer.AclAuthorizer
allow.everyone.if.no.acl.found=true
super.users=User:kafka;User:@@@ ##跟kafka.keytab的pricipals的用户名一致。
#ACL ： topic 的用户权限配置，比如org.apache.ranger.authorization.kafka.authorizer.RangerKafkaAuthorizer
super.users=User:kafka;User:@@@ //acl相关，broker服务本身是采用kafka这个服务身份进行交互，只有配置成superuser才能获取集群内的metadata信息。
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer //acl相关，配置后才能启用acl
#设置为true，ACL机制为黑名单机制，只有黑名单中的用户无法访问
allow.everyone.if.no.acl.found=true
#默认为false，ACL机制为白名单机制，只有白名单中的用户可以访问。启动kafka时，需添加zookeeper.sasl.client.username=zookeeper，需要有该pricipals。
#allow.everyone.if.no.acl.found=false
3、配置kafka/config/kafka_jaas.conf
KafkaServer {
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
keyTab=“/data/kafka/kafka_2.12-2.8.0-krb/config/kafka.keytab”
storeKey=true
useTicketCache=false
serviceName=“kafka”
principal=“kafka/demo2@HADOOP.COM”;
};
KafkaClient {
com.sun.security.auth.module.Krb5LoginModule required
useTicketCache=false
useKeyTab=true
seKeyTab=true
keyTab=“/data/kafka/kafka_2.12-2.8.0-krb/config/kafka.keytab”
principal=“kafka/demo2@HADOOP.COM”
renewTicket=true
serviceName=“kafka”;
};
Client {
com.sun.security.auth.module.Krb5LoginModule required
useTicketCache=false
useKeyTab=true
seKeyTab=true
keyTab=“/data/kafka/kafka_2.12-2.8.0-krb/config/zk.keytab”
principal=“zookeeper/demo2@HADOOP.COM”
renewTicket=true
serviceName=“zookeeper”;
};
4、修改bin目录下kafka-run-class.sh，增加kerberos启动参数：
export KAFKA_OPTS=“-Djava.security.krb5.conf=/etc/krb5.conf -Djava.security.auth.login.config=/data/kafka/kafka_2.12-2.8.0-krb/config/kafka_jaas.conf -Dsun.security.krb5.debug=true”
-Djava.security.krb5.conf=/etc/krb5.conf -Djava.security.auth.login.config=/opt/kafka/config/kafka_server_jaas.conf
-Dzookeeper.sasl.client=Client #默认是Client，对应jaas中的Client
-Dzookeeper.sasl.client.username=zk ##默认是zookeeper，跟pricipals一致。
5、启动kafka
./bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
./bin/kafka-server-start.sh -daemon ./config/server.properties
常见问题：/etc/krb5.conf中配置的default realm必须和使用的realm一致，否则必须先做kdc互信。
##注意kafka broker与kdc的时间同步。
查看错误日志：kafkaServer.out
6、测试创建topic：
./bin/kafka-topics.sh --create --zookeeper demo2:12181 --replication-factor 1 --partitions 1 --topic test
./bin/kafka-topics.sh --list --zookeeper demo2:12181
6、客户端使用kafka
启用kerberos后，部分kafka管理脚本需要增加额外的参数才能使用
首先建立配置文件client.properties:
security.protocol=SASL_PLAINTEXT
sasl.kerberos.service.name=kafka
sasl.mechanism=GSSAPI
所以新命令的使用方式为:
./bin/kafka-consumer-groups.sh --bootstrap-server demo2:9092 --list --command-config ./client.properties
./bin/kafka-console-producer.sh --broker-list demo2:9092 --topic test --producer.config ./client.properties
./bin/kafka-console-consumer.sh --bootstrap-server demo2:9092 --topic test --consumer.config ./client.properties
7、登录zk： ./bin/zookeeper-shell.sh demo2:12181
三、Kakfa配置SSL安全认证
http://kafka.apache.org/documentation/#security_ssl
https://docs.confluent.io/platform/current/kafka/authentication_ssl.html
单向SSL认证：仅仅是在客户端，校验kafka服务器端的合法性。ssl.client.auth=requested。只需拷贝服务器端的kafka.client.truststore.jks即可。
双向SSL认证：除了客户端去校验kafka服务器端的合法性，服务器端还要校验客户端的合法性。ssl.client.auth=required。需要拷贝服务器端的kafka.client.truststore.jks和kafka.server.keystore.jks。
(一)、SSL认证之生成证书
https://blog.csdn.net/justry_deng/article/details/88383081 中文配置
SSL的基础知识、kafka证书生成过程，参考《 SSL安全认证是什么？》
证书生成脚本参考：
集群内选择其中一台broker节点，执行以下命令：
#! /bin/bash
CLUSTER_NAME=kafak2.8-ssl
BASE_DIR=/data/kafka/kafka_2.12-2.8.0/ssl
export PASSWORD=“@@@@123”
KEY_STORE=“ B A S E D I R / s e r v e r / k a f k a . s e r v e r . k e y s t o r e . j k s " C E R T O U T P U T P A T H = " BASE_DIR/server/kafka.server.keystore.jks" CERT_OUTPUT_PATH=" BASEDIR/server/kafka.server.keystore.jks"CERTOUTPUTPATH="BASE_DIR/ca”
CERT_AUTH_FILE=“ C E R T O U T P U T P A T H / c a − c e r t . p e m " T R U S T S T O R E P A T H = " CERT_OUTPUT_PATH/ca-cert.pem" TRUST_STORE_PATH=" CERTOUTPUTPATH/ca−cert.pem"TRUSTSTOREPATH="BASE_DIR/trust”
CLUSTER_CERT_FILE=“ B A S E D I R / s e r v e r / c l u s t e r − BASE_DIR/server/cluster- BASEDIR/server/cluster−{CLUSTER_NAME}-cert”
echo “Step1: Config env”
STORE_PASSWORD= P A S S W O R D K E Y P A S S W O R D = PASSWORD KEY_PASSWORD= PASSWORDKEYPASSWORD=PASSWORD
TRUST_KEY_PASSWORD= P A S S W O R D T R U S T S T O R E P A S S W O R D = PASSWORD TRUST_STORE_PASSWORD= PASSWORDTRUSTSTOREPASSWORD=PASSWORD
DAYS_VALID=3650
export Local_HOSTNAME=hostname -f
D_NAME=“CN=$Local_HOSTNAME, OU=asiainfo”
mkdir -p ${BASE_DIR}/{ca,server,client,trust}
echo “Step2: Create certificate to keystore.每个broker节点执行”
keytool -genkey -keystore $KEY_STORE -alias $CLUSTER_NAME -validity $DAYS_VALID -keyalg RSA -storetype pkcs12 -storepass $STORE_PASSWORD -keypass K E Y P A S S W O R D − d n a m e " KEY_PASSWORD -dname " KEYPASSWORD−dname"D_NAME" -ext “SAN=DNS:$Local_HOSTNAME”
echo “Step3: Create CA（证书签发机构）的证书.只需在其中一个节点上执行即可，执行完成后，会生成了两个文件cat-key、ca-cert，然后将这两个文件分别拷贝到所有broker节点上，后面需要用到。”
openssl req -new -x509 -keyout C E R T O U T P U T P A T H / c a − k e y − o u t " CERT_OUTPUT_PATH/ca-key -out " CERTOUTPUTPATH/ca−key−out"CERT_AUTH_FILE" -days “ D A Y S V A L I D " − p a s s i n p a s s : " DAYS_VALID" -passin pass:" DAYSVALID"−passinpass:"PASSWORD” -passout pass:“ P A S S W O R D " − s u b j " / C = C N / S T = b e i j i n g / L = b e i j i n g / O = a s i a i n f o / O U = a s i a i n f o / C N = PASSWORD" -subj "/C=CN/ST=beijing/L=beijing/O=asiainfo/OU=asiainfo/CN= PASSWORD"−subj"/C=CN/ST=beijing/L=beijing/O=asiainfo/OU=asiainfo/CN=Local_HOSTNAME”
#openssl req -x509 -config openssl-ca.cnf -newkey rsa:4096 -sha256 -nodes -out cacert.pem -outform PEM
echo “Step4: Import CA into kafka client truststore.每个broker节点执行”
keytool -keystore “ T R U S T S T O R E P A T H / k a f k a . c l i e n t . t r u s t s t o r e . j k s " − a l i a s C A R o o t − i m p o r t − f i l e " TRUST_STORE_PATH/kafka.client.truststore.jks" -alias CARoot -import -file " TRUSTSTOREPATH/kafka.client.truststore.jks"−aliasCARoot−import−file"CERT_AUTH_FILE” -storepass “ T R U S T S T O R E P A S S W O R D " − k e y p a s s " TRUST_STORE_PASSWORD" -keypass " TRUSTSTOREPASSWORD"−keypass"TRUST_KEY_PASS” -noprompt
echo “Step4: Import CA into kafka server truststore.每个broker节点执行”
keytool -keystore “ T R U S T S T O R E P A T H / k a f k a . s e r v e r . t r u s t s t o r e . j k s " − a l i a s C A R o o t − i m p o r t − f i l e " TRUST_STORE_PATH/kafka.server.truststore.jks" -alias CARoot -import -file " TRUSTSTOREPATH/kafka.server.truststore.jks"−aliasCARoot−import−file"CERT_AUTH_FILE” -storepass “ T R U S T S T O R E P A S S W O R D " − k e y p a s s " TRUST_STORE_PASSWORD" -keypass " TRUSTSTOREPASSWORD"−keypass"TRUST_KEY_PASS” -noprompt
echo “Step5: Export certificate from (Step2)keystore, prepare for signing.导出证书，准备签名.每个broker节点执行”
keytool -keystore “ K E Y S T O R E " − a l i a s " KEY_STORE" -alias " KEYSTORE"−alias"CLUSTER_NAME” -certreq -file “ C L U S T E R C E R T F I L E " − s t o r e p a s s " CLUSTER_CERT_FILE" -storepass " CLUSTERCERTFILE"−storepass"STORE_PASSWORD” -keypass “$KEY_PASSWORD” -noprompt
echo “Step6: Signing the certificate.证书签名.每个broker节点执行”
openssl x509 -req -CA “$CERT_AUTH_FILE” -CAkey C E R T O U T P U T P A T H / c a − k e y − i n " CERT_OUTPUT_PATH/ca-key -in " CERTOUTPUTPATH/ca−key−in"CLUSTER_CERT_FILE" -out “ C L U S T E R C E R T F I L E − s i g n e d " − d a y s " {CLUSTER_CERT_FILE}-signed" -days " CLUSTERCERTFILE−signed"−days"DAYS_VALID” -CAcreateserial -passin pass:“$PASSWORD”
echo “Setp7: Import CA into keystore”
keytool -keystore “ K E Y S T O R E " − a l i a s C A R o o t − i m p o r t − f i l e " KEY_STORE" -alias CARoot -import -file " KEYSTORE"−aliasCARoot−import−file"CERT_AUTH_FILE.每个broker节点执行” -storepass “ S T O R E P A S S W O R D " − k e y p a s s " STORE_PASSWORD" -keypass " STOREPASSWORD"−keypass"KEY_PASSWORD” -noprompt
echo “Setp8: Import signed certificate into keystore.每个broker节点执行”
keytool -keystore “ K E Y S T O R E " − a l i a s " KEY_STORE" -alias " KEYSTORE"−alias"{CLUSTER_NAME}” -import -file “ C L U S T E R C E R T F I L E − s i g n e d " − s t o r e p a s s " {CLUSTER_CERT_FILE}-signed" -storepass " CLUSTERCERTFILE−signed"−storepass"STORE_PASSWORD” -keypass “$KEY_PASSWORD” -noprompt
echo “done,list keystore”
keytool -list -v -keystore $KEY_STORE -storepass ${KEY_PASSWORD}
注意：以上步骤，只有步骤3，生成CA认证证书时，在其中一台主机上执行即可。集群内的所有主机使用同一个CA证书签发机构。
集群内其他broker节点，执行以下命令：执行除了Step3: Create CA（证书签发机构)以外的其他的命令。
（二）、修改config/server.properties
注意：SSL配置最好写在配置文件的最上面，否者可能导致Kafka配置SSL失败。
1、如果配置了SSL认证，那么原来的port和advertised.listeners可以注释掉。也可以保留PLAINTEXT配置，另外增加SSL协议的端口吗？
listeners=PLAINTEXT://host.name:port,SSL://host.name:9093
advertised.listeners=PLAINTEXT://localhost:9092,SSL://localhost:9093
#advertised.listeners=SSL://kafka-single:9095
ssl.keystore.location=/…/ssl/server/kafka.server.keystore.jks
ssl.keystore.password=@@@@123
ssl.key.password=@@@@123
ssl.truststore.location=/…/ssl/trust/kafka.server.truststore.jks
ssl.truststore.password=@@@@123##注意：ssl.truststore.password 在技术上是可选的，但强烈推荐设置。 如果未设置密码，则仍可访问信任库，但会禁用完整性检查。
ssl.client.auth=required ##是否需要客户端身份验证。如果设置为required，是双向认证，服务器端会验证客户端身份。如果设置为requested，是单向认证，则没有keystore证书的客户端也可连接broker。不建议设置为reqeusted。
ssl.keystore.type=PKCS12
ssl.truststore.type=JKS
2、其他可选参数
ssl.enabled.protocols=TLSv1.2,TLSv1.1,TLSv1 ##至少要列出broker端的协议之一。
ssl.cipher.suites= ##TLS or SSL
ssl.secure.random.implementation=SHA1PRNG
配置主机名验证。
kafka2.0.x开始， 默认情况下客户端连接以及broker之间的连接启用了服务器的主机名验证，为防止中间人攻击。
将ssl.endpoint.identification.algorithm设置为了HTTPS，即:需要验证主机名。
如果不需要验证主机名，那么可以这么设置ssl.endpoint.identification.algorithm=即可。
ssl.endpoint.identification.algorithm=HTTPS
设置内部访问也用SSL，默认值为security.inter.broker.protocol=PLAINTEXT
security.inter.broker.protocol=SSL
注：如果设置的内部broker的通讯协议PLAINTEXT，那么监听PLAINTEXT的时候就需要作相应的配置listeners=PLAINTEXT://host.name:port,SSL://host.name:port。
(三)、重启kafka
./bin/kafka-server-start.sh -daemon ./config/server.ssl.properties
如果配置SSL之前，存在Kafka数据，那么建议重新换一个位置来存放数据；
如果确保之前的数据已经没什么用了，也可以直接删除之前的数据
(四)、本机测试
1、测试9093端口可以访问：
openssl s_client -debug -connect demo3:9093 -tls1
以下说明连接成功。
2、测试创建Topic
./bin/kafka-topics.sh --list --zookeeper demo3:2181/kafka-2.8
./bin/kafka-topics.sh --create --zookeeper demo3:2181/kafka-2.8 --replication-factor 2 --partitions 10 --topic test2
3、测试生产、消费数据
首先，创建consumer.properties
group.id=demo3-group
client.id=demo3
security.protocol=SSL
ssl.truststore.location=/data/kafka/kafka_2.12-2.8.0/ssl/trust/kafka.client.truststore.jks
ssl.truststore.password=@@@@123
##如果服务器端设置了：ssl.client.auth=required，则必须提供kafka.server.keystore.jks文件。如果ssl.client.auth=requested，则可以只提供kafka.client.truststore.jks文件。
ssl.keystore.password=@@@@123
ssl.keystore.location=/data/kafka/kafka_2.12-2.8.0/ssl/server/kafka.server.keystore.jks
然后生产、消费数据：
./bin/kafka-console-producer.sh --broker-list demo3:9093 --topic test --producer.config ./ssl/producer.properties
./bin/kafka-console-consumer.sh --bootstrap-server demo3:9093 --topic test --from-beginning --consumer.config ./ssl/consumer.properties
三、远程Kakfa客户端配置https://docs.confluent.io/platform/current/kafka/authentication_ssl.html#clients
任意要与broker进行交互的主机，都要提供安全配置信息。包括：副本、kafka客户端，java客户端。
如果直接在其他主机上执行，消费、生产数据的命令时，会报错Broker连接失败：Bootstrap broker demo3:9093 (id: -1 rack: null) disconnected (org.apache.kafka.clients.NetworkClient)
1、将Kafka集群中的ssl/server/kafka.server.keystore.jks文件和ssl/server/kafka.client.truststore.jks文件，拷贝至客户端运行环境。
2、配置consumer.properties（参考以上配置）
#group.id=demo2-consumer-group
client.id=demo2
security.protocol=SSL
ssl.truststore.location=/data/kafka/kafka_2.12-2.8.0/demo3-ssl/trust/kafka.client.truststore.jks
ssl.truststore.password=@@@@123
##如果服务器端设置了：ssl.client.auth=required，则必须提供kafka.server.keystore.jks文件。如果ssl.client.auth=requested，则可以只提供kafka.client.truststore.jks文件。
ssl.keystore.password=@@@@123
ssl.keystore.location=/data/kafka/kafka_2.12-2.8.0/ssl/server/kafka.server.keystore.jks
3、执行消费命令
./bin/kafka-console-consumer.sh --bootstrap-server demo3:9093 --topic test --from-beginning --consumer.config ./ssl/consumer.properties
四、java访问使用SSL加密的kafka集群。
1、将Kafka集群中的ssl/server/kafka.server.keystore.jks文件和ssl/server/kafka.client.truststore.jks文件，拷贝至客户端
2、配置Properties类的参数
Properties props = new Properties();
props.put(“security.protocol”, “SSL”);
props.put(“ssl.truststore.location”, “E:\kafka\safe\client.truststore.jks”);
props.put(“ssl.truststore.password”, “test1234”);
props.put("ssl.keystore.location", "E:\\kafka\\safe\\client.keystore.jks");
props.put("ssl.keystore.password", "p2****02");
props.put("ssl.key.password", "p((((2");
KafkaConsumer consumer=new KafkaConsumer(props);