Flink SQL连接Confluent Kafka Cloud遇TLSv1.3 SSLContext缺失错误排查

尝试用Flink SQL连接Confluent Cloud时触发如下错误：

Caused by: org.apache.kafka.common.KafkaException: java.security.NoSuchAlgorithmException: TLSv1.3 SSLContext not available
    at org.apache.kafka.common.security.ssl.DefaultSslEngineFactory.createSSLContext(DefaultSslEngineFactory.java:268)
    at org.apache.kafka.common.security.ssl.DefaultSslEngineFactory.configure(DefaultSslEngineFactory.java:173)
    at org.apache.kafka.common.security.ssl.SslFactory.instantiateSslEngineFactory(SslFactory.java:140)
    at org.apache.kafka.common.security.ssl.SslFactory.configure(SslFactory.java:97)
    at org.apache.kafka.common.network.SaslChannelBuilder.configure(SaslChannelBuilder.java:180)
    ... 18 more
Caused by: java.security.NoSuchAlgorithmException: TLSv1.3 SSLContext not available
    at java.base/sun.security.jca.GetInstance.getInstance(GetInstance.java:159)
    at java.base/javax.net.ssl.SSLContext.getInstance(SSLContext.java:168)
    at org.apache.kafka.common.security.ssl.DefaultSslEngineFactory.createSSLContext(DefaultSslEngineFactory.java:243)

使用的Flink SQL语句：

CREATE TABLE IF NOT EXISTS some_source_table
(
    headers     VARCHAR NOT NULL,
    id          VARCHAR NOT NULL,
    `timestamp` TIMESTAMP_LTZ(3) NULL,
    type        VARCHAR NOT NULL,
    contentJson VARCHAR NOT NULL
) WITH (
    'connector' = 'kafka',
    'topic-pattern' = 'kafka_topic__.+?',
    'properties.bootstrap.servers' = 'some.aws.confluent.cloud:9092',
    'properties.group.id' = 'some-id-1',
    'scan.startup.mode' = 'latest-offset',
    'format' = 'json',
    'json.timestamp-format.standard' = 'ISO-8601',
    'scan.topic-partition-discovery.interval'= '60000',
    'json.fail-on-missing-field' = 'false',
    'json.ignore-parse-errors' = 'true',
    'properties.security.protocol' = 'SASL_SSL',
    'properties.sasl.mechanism' = 'PLAIN',
    'properties.sasl.jaas.config' = 'org.apache.kafka.common.security.plain.PlainLoginModule required username=*** password=***;',
    'properties.ssl.endpoint.identification.algorithm' = 'https'
);

运行TaskManager的Kubernetes Pod中Java版本为11.0.18（通过System.getProperty("java.version")获取）。

补充的Gradle依赖配置：

implementation "org.apache.flink:flink-streaming-java:${flinkVersion}"
implementation "org.apache.flink:flink-table-api-java-bridge:${flinkVersion}"
implementation "org.apache.flink:flink-table-planner_${scalaVersion}:${flinkVersion}"
implementation "org.apache.flink:flink-json:${flinkVersion}"
implementation "org.apache.flink:flink-clients:${flinkVersion}"

//this seems to be needed : removing this is causing error in guice JsonProperty.Naming
implementation "com.fasterxml.jackson.core:jackson-databind:${jacksonVersion}"
implementation "com.fasterxml.jackson.datatype:jackson-datatype-jsr310:${jacksonVersion}"

implementation "org.apache.flink:flink-statebackend-rocksdb:${flinkVersion}"

//needed for a local flink ui to show when running environment = local
implementation "org.apache.flink:flink-runtime-web:${flinkVersion}"

implementation "org.apache.logging.log4j:log4j-core:${log4jVersion}"
implementation "org.apache.logging.log4j:log4j-api:${log4jVersion}"
implementation "org.apache.logging.log4j:log4j-slf4j-impl:${log4jVersion}"

implementation 'io.jsonwebtoken:jjwt:0.2'
implementation 'com.mashape.unirest:unirest-java:1.4.9'
implementation 'org.rocksdb:rocksdbjni:7.9.2'

// --------------------------------------------------------------
// Dependencies that should be part of the shadow jar, e.g.
// connectors. These must be in the flinkShadowJar configuration!
// --------------------------------------------------------------
flinkShadowJar files('libs/flink-connector-kafka-1.16.0.jar')
flinkShadowJar "org.apache.flink:flink-connector-base:${flinkVersion}"

flinkShadowJar "org.apache.commons:commons-text:1.10.0"
flinkShadowJar "org.projectlombok:lombok:1.18.26"
flinkShadowJar "com.fasterxml.jackson.core:jackson-databind:${jacksonVersion}"
flinkShadowJar "com.fasterxml.jackson.datatype:jackson-datatype-jsr310:${jacksonVersion}"
flinkShadowJar "com.sailpoint:atlas:${ATLAS_VERSION}"
flinkShadowJar "com.sailpoint:atlas-event:${ATLAS_VERSION}"
flinkShadowJar "com.google.inject:guice:5.1.0"
flinkShadowJar "org.apache.flink:flink-s3-fs-hadoop:${flinkVersion}"
flinkShadowJar 'io.jsonwebtoken:jjwt:0.2'
flinkShadowJar 'com.mashape.unirest:unirest-java:1.4.9'

请问可能遗漏了什么配置或存在什么问题？

核心原因 #

错误提示TLSv1.3 SSLContext not available，说明当前Java环境无法支持TLSv1.3协议。尽管Java 11默认支持TLSv1.3，但以下几种情况会触发该问题：

1. Java运行环境限制了TLS协议版本 #

部分自定义JRE镜像可能移除了TLSv1.3支持，或通过JVM参数禁用了该协议。

• 解决方法：
  • 检查TaskManager的JVM启动参数，确认没有-Djdk.tls.disabledAlgorithms=TLSv1.3这类禁用配置。
  • 替换为官方标准OpenJDK 11镜像，避免使用裁剪过的JRE。

2. Kafka客户端默认强制使用TLSv1.3，环境不支持 #

你手动引入了本地的flink-connector-kafka-1.16.0.jar，若其依赖的Kafka客户端版本过高，可能默认优先使用TLSv1.3，但当前环境无法适配。

• 解决方法：
  • 在Flink SQL的Kafka连接配置中，显式指定TLS版本为TLSv1.2，添加配置项：

    'properties.ssl.enabled.protocols' = 'TLSv1.2'
    

  • 确保Flink连接器版本与Flink核心版本一致（flinkVersion需为1.16.x），避免版本兼容性问题。

3. 依赖冲突导致SSL类加载异常 #

Gradle依赖中可能存在不同版本的Kafka客户端或SSL相关库，导致类加载混乱，无法初始化TLSv1.3上下文。

• 解决方法：
  • 移除本地的flink-connector-kafka-1.16.0.jar，改用Maven中央仓库的依赖，保证版本匹配：

    flinkShadowJar "org.apache.flink:flink-connector-kafka:${flinkVersion}"
    

  • 执行gradle dependencies查看依赖树，排查Kafka客户端版本冲突，排除多余的低/高版本依赖。

4. 适配Confluent Cloud的TLS要求 #

Confluent Cloud同时支持TLSv1.2和TLSv1.3，显式指定TLSv1.2可绕过当前环境的TLSv1.3支持问题，同时满足连接要求。

• 验证方法：修改Flink SQL的WITH配置块，添加'properties.ssl.enabled.protocols' = 'TLSv1.2'后重新运行任务。

内容的提问来源于stack exchange，提问作者hitesh