华为云 MRS Flink 连接 Kafka 异常记录

exception flink kafka kerberos
flink
发布日期:   2023-11-24
文章字数:   2.8k
阅读时长:   11 分
阅读次数:  

1. 背景

最近在华为云 MRS 环境中提交 Flink 任务，任务中需要通过 Kafka Connector 读取 Kafka 数据。集群开启了安全认证，Kafka 侧使用 Kerberos 认证，Flink 版本与连接器依赖如下：

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka</artifactId>
    <version>1.15.0-h0.cbu.mrs.320.r11</version>
    <exclusions>
        <exclusion>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
        </exclusion>
    </exclusions>
</dependency>

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.4.0-h0.cbu.mrs.320.r11</version>
</dependency>

这次主要遇到了两个问题：

  1. Kerberos 认证阶段报 Server not found in Kerberos database。
  2. Kafka Source 获取 topic metadata 超时，报 Timeout expired while fetching topic metadata。

这两个错误都发生在 Flink Source 打开 Kafka Consumer 的阶段，也就是还没有真正开始消费数据之前。

2. 问题一：Kerberos 认证失败

2.1 报错现象

任务启动后，Flink Kafka Source 初始化失败，核心报错如下：

org.apache.kafka.common.errors.SaslAuthenticationException:
An error: (java.security.PrivilegedActionException: javax.security.sasl.SaslException: GSS initiate failed
[Caused by GSSException: No valid credentials provided
(Mechanism level: Server not found in Kerberos database (7) - LOOKING_UP_SERVER)])
occurred when evaluating SASL token received from the Kafka Broker.

继续向下看，可以看到更底层的 Kerberos 异常：

Caused by: sun.security.krb5.KrbException:
Server not found in Kerberos database (7) - LOOKING_UP_SERVER

这里最关键的不是 No valid credentials provided，而是后面的：

Server not found in Kerberos database

它表示客户端向 KDC 请求 Kafka Broker 的服务票据时，KDC 没有找到对应的服务端 principal。

2.2 相关配置要求

Flink 官方文档中说明，Flink Kafka Connector 支持 Kerberos 认证。使用 Kerberos 时，需要在 Flink 中配置 keytab、principal，并且把 KafkaClient 加到 security.kerberos.login.contexts 中。随后 Kafka 客户端还需要配置：

security.protocol=SASL_PLAINTEXT
sasl.mechanism=GSSAPI
sasl.kerberos.service.name=kafka

其中 sasl.kerberos.service.name 需要和 Kafka Broker 侧配置的服务名一致。Flink 文档也明确说明，如果客户端和服务端的 service name 不一致，会导致认证失败。

华为云 MRS 的 Kafka 安全模式文档也说明，安全集群中 Kafka 应用需要进行 Kafka、ZooKeeper、Kerberos 认证，需要生成 JAAS 文件并设置相关环境变量。

2.3 原因分析

Kafka 使用 SASL/GSSAPI 认证时，客户端并不是只拿用户 principal 去认证，还要根据 Broker 信息去请求服务端 principal。通常形式类似：

kafka/hostname@REALM

这里容易出问题的是 hostname。Kafka 客户端会根据 Broker 返回的地址、DNS 解析结果、主机名反查结果等信息构造服务端 principal。如果构造出来的 principal 在 Kerberos 数据库中不存在，就会出现：

Server not found in Kerberos database

结合这次报错，可以重点怀疑下面几个方向：

检查项 说明
sasl.kerberos.service.name 客户端配置的服务名是否和 Broker 侧一致，通常是 kafka
bootstrap.servers 是否使用了 Kafka 安全端口，MRS SASL_PLAINTEXT 场景通常使用 sasl.port
Broker 主机名 bootstrap.servers 中的主机名是否能在 Flink TaskManager 容器中正确解析
FQDN/短主机名 Kerberos principal 中使用的是 FQDN 还是短主机名，客户端解析结果要和它一致
keytab/principal Flink 提交任务时使用的用户凭据是否正确下发到 TaskManager
krb5.conf Kerberos realm、KDC 地址是否正确

从本质上看，这个错误不是 Kafka topic 不存在，也不是 Flink 算子逻辑异常，而是 Kafka 客户端在认证 Broker 服务端身份时找不到对应的 Kerberos 服务主体。

2.4 处理方式

处理这类问题时，可以按照下面的顺序检查。

2.4.1 确认 Kafka 安全协议

先确认 Kafka 使用的是 SASL_PLAINTEXT 还是 SASL_SSL。华为云 DLI/MRS 文档中也提到，MRS 使用 Kerberos 认证时，需要根据 Kafka 服务配置中的 security.protocol 设置客户端协议。

如果是 SASL_PLAINTEXT：

security.protocol=SASL_PLAINTEXT
sasl.mechanism=GSSAPI
sasl.kerberos.service.name=kafka

如果是 SASL_SSL，还需要额外配置 truststore：

security.protocol=SASL_SSL
sasl.mechanism=GSSAPI
sasl.kerberos.service.name=kafka
ssl.truststore.location=/path/to/truststore.jks
ssl.truststore.password=xxx

在 Flink 集群配置中，需要保证 KafkaClient 可以拿到 Kerberos 登录信息。

security.kerberos.login.keytab: /path/to/user.keytab
security.kerberos.login.principal: user@REALM
security.kerberos.login.contexts: Client,KafkaClient

如果任务运行在 YARN 上，还要确认 keytab、krb5.conf、jaas.conf 等文件是否下发到了 TaskManager 所在容器。很多时候本地客户端能访问 Kafka，但 Flink 容器中访问失败，就是因为认证文件只在提交节点存在，没有进入实际运行容器。

2.4.3 确认 Broker 地址和 principal 匹配

Kafka Broker 发布给客户端的地址由 Broker 配置控制。Kafka 文档中说明，advertised.listeners 是发布给客户端使用的监听地址，在云环境中它可能和实际绑定地址不同。

因此需要确认：

# 在 Flink TaskManager 所在节点或容器中检查
getent hosts kafka-broker-1
hostname -f

如果 Kerberos 中的 principal 是：

kafka/kafka-broker-1.example.com@EXAMPLE.COM

那么客户端最终解析出来的 Broker 主机名也应该和 kafka-broker-1.example.com 对得上。不要一边使用 IP，一边在 Kerberos 中注册 FQDN；也不要一边使用短主机名，一边在 Kerberos 中只有 FQDN principal。

2.4.4 使用 Kafka 自带命令先验证

在提交 Flink 作业之前，建议先在同一台客户端机器上用 Kafka 命令验证认证是否正常。

kafka-topics.sh \
  --bootstrap-server kafka-broker-1:21007 \
  --command-config client.properties \
  --list

client.properties 示例：

security.protocol=SASL_PLAINTEXT
sasl.mechanism=GSSAPI
sasl.kerberos.service.name=kafka

如果 Kafka 自带命令都无法通过认证，Flink 作业也基本不会成功。

3. 问题二：获取 Topic Metadata 超时

3.1 报错现象

第二个问题中，Flink Source 初始化时失败，核心报错如下：

org.apache.kafka.common.errors.TimeoutException:
Timeout expired while fetching topic metadata

后面 Flink 又报：

org.apache.flink.runtime.JobException:
Recovery is suppressed by NoRestartBackoffTimeStrategy

Caused by: org.apache.kafka.common.errors.TimeoutException:
Timeout expired while fetching topic metadata

这里 NoRestartBackoffTimeStrategy 只是说明当前作业没有配置失败重启策略，所以 Source 初始化失败后作业直接失败。真正的根因仍然是 Kafka 客户端获取 topic metadata 超时。

3.2 官方解释

华为云 DLI 文档对这个错误的说明比较直接：Flink JobManager 提示 Timeout expired while fetching topic metadata，表示 Flink 作业尝试获取 Kafka topic 元数据时超时。此时应先检查 Flink 作业和 Kafka 之间的网络连通性，确保运行 Flink 作业的队列能够访问 Kafka 所在的 VPC 网络。

虽然这里的环境是 MRS Flink，不是 DLI Flink SQL，但错误语义是一样的：Kafka 客户端没有在超时时间内拿到 topic metadata。

3.3 常见原因

这个错误不一定只表示网络完全不通，也可能是认证协议或端口配置错误。常见原因如下：

原因 说明
网络不通 Flink TaskManager 所在节点无法访问 Kafka Broker 地址
端口错误 使用了普通端口连接安全 Kafka，或使用了错误的 SASL/SSL 端口
协议不匹配 Broker 要求 SASL_PLAINTEXT 或 SASL_SSL，客户端仍按明文普通连接访问
advertised.listeners 不可达 客户端能连上 bootstrap 地址，但 Broker 返回的 advertised 地址不可达
topic 不存在或无权限 客户端无法获取目标 topic metadata
Kerberos 文件未下发 TaskManager 容器中缺少 keytab、krb5.conf、jaas.conf
Kafka 客户端依赖冲突 Flink Connector 和 kafka-clients 版本不匹配，导致认证或协议行为异常

在云环境中，尤其要注意 advertised.listeners。客户端先连接 bootstrap.servers，随后会根据 Broker 返回的 metadata 再连接具体 Broker。如果返回的是 Flink 所在网络无法访问的地址，就会表现为获取 metadata 超时。

3.4 排查步骤

在 Flink TaskManager 所在节点上测试 Kafka 端口是否可达。

nc -vz kafka-broker-1 21007
nc -vz kafka-broker-2 21007

如果使用 SASL_SSL，端口可能不是 21007，需要以 Kafka 服务配置中的端口为准。

3.4.2 检查 Kafka 连接配置

Flink Kafka Source 中至少需要确认下面几个配置：

bootstrap.servers=kafka-broker-1:21007,kafka-broker-2:21007
group.id=test_group
security.protocol=SASL_PLAINTEXT
sasl.mechanism=GSSAPI
sasl.kerberos.service.name=kafka

如果是 SSL 场景，还需要补充 truststore 配置。

3.4.3 用 Kafka 命令验证 topic metadata

同样建议先用 Kafka 自带命令验证 topic 是否可访问。

kafka-topics.sh \
  --bootstrap-server kafka-broker-1:21007 \
  --command-config client.properties \
  --describe \
  --topic dwd_picture_features

如果这里也超时，说明问题不在 Flink 业务代码，而是在网络、端口、认证或 topic 权限上。

3.4.4 检查依赖版本

草稿中记录的依赖是 MRS 定制版本：

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka</artifactId>
    <version>1.15.0-h0.cbu.mrs.320.r11</version>
    <exclusions>
        <exclusion>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
        </exclusion>
    </exclusions>
</dependency>

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.4.0-h0.cbu.mrs.320.r11</version>
</dependency>

这里需要注意两点：

  1. 不要混用开源原版 kafka-clients 和 MRS 定制 Connector。
  2. 如果使用 MRS 提供的 Flink/Kafka 版本，优先使用对应 MRS 版本的依赖包。

如果包冲突，可能出现本地编译正常、提交到集群后认证行为异常的问题。

下面是安全 Kafka 场景下比较常见的一组配置，实际端口和协议需要以 MRS Kafka 服务配置为准。

Properties props = new Properties();
props.setProperty("bootstrap.servers", "kafka-broker-1:21007,kafka-broker-2:21007");
props.setProperty("group.id", "test_group");
props.setProperty("security.protocol", "SASL_PLAINTEXT");
props.setProperty("sasl.mechanism", "GSSAPI");
props.setProperty("sasl.kerberos.service.name", "kafka");

如果使用 SASL_SSL，则需要改成：

props.setProperty("security.protocol", "SASL_SSL");
props.setProperty("ssl.truststore.location", "/path/to/truststore.jks");
props.setProperty("ssl.truststore.password", "xxx");

Flink 集群层面还要配置 Kerberos 登录信息：

security.kerberos.login.keytab: /path/to/user.keytab
security.kerberos.login.principal: user@REALM
security.kerberos.login.contexts: Client,KafkaClient

5. 总结

这两个问题虽然报错不同，但都发生在 Kafka Source 初始化阶段。

问题 关键报错 主要排查方向
Kerberos 服务端 principal 找不到 Server not found in Kerberos database 检查 sasl.kerberos.service.name、Broker 主机名、FQDN、keytab、krb5.conf
获取 topic metadata 超时 Timeout expired while fetching topic metadata 检查网络连通性、端口、安全协议、advertised.listeners、topic 权限

排查这类问题时，不建议一开始就怀疑 Flink 算子逻辑。更有效的顺序是：

  1. 在 Flink 运行节点测试 Kafka 端口连通性。
  2. 使用 Kafka 自带命令验证认证和 topic 访问。
  3. 检查 Flink 作业中 Kafka 参数是否和 MRS Kafka 服务配置一致。
  4. 检查 keytab、krb5.conf、jaas.conf 是否真正下发到 TaskManager 容器。
  5. 最后再看 Flink Connector 和 kafka-clients 依赖是否和 MRS 版本匹配。

6. 参考资料

文章作者: hnbian
文章链接: https://www.hnbian.cn/posts/bcc705f6.html
版权声明: 本博客所有文章除特別声明外，均采用 CC BY 4.0 许可协议。转载请注明来源 hnbian !
exception flink kafka kerberos
评论
  目录