赞
踩
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失败。
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=HTTPS
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);
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。