flink sql1.18.0连接SASL_PLAINTEXT认证的kafka3.3.1

阅读此文默认读者对docker、docker-compose有一定了解。

环境

docker-compose运行了一个jobmanager、一个taskmanager和一个sql-client。

如下：

version: "2.2"
services:jobmanager:image: flink:1.18.0-scala_2.12container_name: jobmanagerports:- "7081:8081"command: jobmanagervolumes:- ./jobmanager:/opt/flinkenvironment:- |FLINK_PROPERTIES=jobmanager.rpc.address: jobmanagertaskmanager:image: flink:1.18.0-scala_2.12container_name: taskmanager1depends_on:- jobmanagercommand: taskmanagervolumes:- ./taskmanager1:/opt/flinkscale: 1environment:- |FLINK_PROPERTIES=jobmanager.rpc.address: jobmanagertaskmanager.numberOfTaskSlots: 32sql-client:image: flink:1.18.0-scala_2.12container_name: sql-client-1command: bin/sql-client.shvolumes:- ./sql-client:/opt/flinkdepends_on:- jobmanagerenvironment:- |FLINK_PROPERTIES=jobmanager.rpc.address: jobmanagerrest.address: jobmanager

注意三个容器都映射了/opt/flink目录。需要先将/opt/flink目录拷贝到跟docker-compose.yml同一目录下，并分别重命名，如下图：

三个文件夹内容是一样的，只是名字不一样。

以上环境介绍完毕。

添加fllink-connector-kafka驱动

在maven官网下载相应jar包，分别放入上述三个文件夹的lib目录下。例如jobmanager:

启动sql-client，我用docker-compose启动的，因此命令为：docker-compose run sql-client.

创建kafka表

CREATE TABLE TestTable (`event_time` TIMESTAMP(3) METADATA FROM 'timestamp',`partition` BIGINT METADATA VIRTUAL,`offset` BIGINT METADATA VIRTUAL,`presetBit` STRING,`imageTime` STRING,`imageName` STRING,`thumbnailWidth` BIGINT,`size` BIGINT,`thumbnailSize` BIGINT,`behavior` STRING,`imageUri` STRING,`presetId` STRING
) WITH ('connector'='kafka','topic'='mytopic','properties.bootstrap.servers'='localhost:9092','properties.group.id'='testGroup','scan.startup.mode'='earliest-offset','format'='json','properties.security.protocol'='SASL_PLAINTEXT','properties.sasl.mechanism'='PLAIN','properties.sasl.jaas.config'='org.apache.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="your-username" password="your-pwd";'
);

说明：

①标识字段为kafka自带字段，topic中没有也会自带。

②标识的字段为topic中存储的字段，根据自己topic来。

③为自己的topic名称

④为kafka集群地址

⑤后面的username和password需要根据实际情况修改。

创建表以后执行select * from TestTable,可以看到类似下图的内容：

flink官网留下的坑

坑主要出现在最后一行。

flink官网是这样写的

首先指定的类不对，应该是org.apache.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule，如果按照官网写会报错：Caused by: javax.security.auth.login.LoginException: No LoginModule found for org.apache.kafka.common.security.plain.PlainLoginModule。

其次，username和password的双引号不需要写反斜杠，写反斜杠反而会报错。