Google Dataflow JDBCIO Java传SSL证书报jks文件不存在解决方法

报错核心由3个代码逻辑错误导致：

• 证书下载逻辑执行位置错误：ConsumerFactoryFn.apply()在Pipeline提交阶段的本地机器运行，而JdbcIO的实际读取逻辑在分布式Worker节点执行，你将证书写入提交机本地/tmp路径，Worker节点上不存在该文件，自然触发文件不存在报错。
• JDBC连接属性键名大小写错误：Presto/Trino JDBC驱动对SSL配置键大小写敏感，正确键名为sslTrustStorePath、sslTrustStorePassword，首字母大写的配置会被驱动忽略，驱动会按默认路径查找证书文件，最终触发找不到.jks的报错。
• 临时文件逻辑冗余且存在隐患：你硬编码/tmp路径写文件，Beam Worker对/tmp目录有权限限制、自动清理规则，硬编码路径容易触发权限问题、文件被提前清理的问题；额外调用的createTempFile逻辑完全没有用到生成的文件路径，属于无效代码。

1. 移除Pipeline构建阶段（主方法内、apply传参时直接执行的逻辑）下载证书的代码，这类代码只会在任务提交机执行，Worker节点无法感知。
2. 实现可序列化的数据源供应类，在Worker节点初始化JDBC连接前，从GCS下载证书到Worker本地合法临时目录，不要硬编码固定路径。
3. 修正JDBC连接属性的键名大小写，和驱动官方要求保持一致。
4. 给下载到本地的证书文件设置全局可读权限，避免JDBC驱动无权限读取文件。

自定义SSL数据源供应类 #

该类会在Worker节点初始化JDBC连接时执行，保证证书存在于当前运行节点的本地路径：

import com.google.cloud.storage.Blob;
import com.google.cloud.storage.Storage;
import com.google.cloud.storage.StorageOptions;
import com.google.common.io.ByteStreams;
import org.apache.beam.sdk.transforms.SerializableSupplier;
import org.apache.commons.dbcp2.DriverManagerDataSource;
import javax.sql.DataSource;
import java.io.OutputStream;
import java.nio.channels.Channels;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Properties;

public class PrestoSslDataSourceProvider implements SerializableSupplier<DataSource> {
    private final String jdbcUrl;
    private final String username;
    private final String password;
    private final String certBucket;
    private final String certObjectPath;
    private final String trustStorePwd;

    public PrestoSslDataSourceProvider(String jdbcUrl, String username, String password,
                                       String certBucket, String certObjectPath, String trustStorePwd) {
        this.jdbcUrl = jdbcUrl;
        this.username = username;
        this.password = password;
        this.certBucket = certBucket;
        this.certObjectPath = certObjectPath;
        this.trustStorePwd = trustStorePwd;
    }

    @Override
    public DataSource get() {
        try {
            // 在Worker节点临时目录创建证书文件，避免硬编码/tmp的权限问题
            Path localTrustStorePath = Files.createTempFile("presto-ssl", ".jks");
            
            // 从GCS下载证书到本地临时路径
            Storage storage = StorageOptions.getDefaultInstance().getService();
            Blob certBlob = storage.get(certBucket, certObjectPath);
            try (OutputStream os = Files.newOutputStream(localTrustStorePath)) {
                ByteStreams.copy(Channels.newInputStream(certBlob.reader()), os);
            }

            // 设置文件可读权限，避免驱动无权限访问
            localTrustStorePath.toFile().setReadable(true, false);

            // 配置JDBC连接参数，注意SSL属性键名全小写开头
            Properties connProps = new Properties();
            connProps.setProperty("user", username);
            connProps.setProperty("password", password);
            connProps.setProperty("SSL", "true");
            connProps.setProperty("sslTrustStorePath", localTrustStorePath.toAbsolutePath().toString());
            connProps.setProperty("sslTrustStorePassword", trustStorePwd);

            return new DriverManagerDataSource(jdbcUrl, connProps);
        } catch (Exception e) {
            throw new RuntimeException("JDBC SSL数据源初始化失败", e);
        }
    }
}

Pipeline调用逻辑 #

替换原来的withDataSourceConfiguration配置，传入自定义数据源供应类：

pipeline.apply("Read from JDBC presto", JdbcIO.<TableRow>read()
                .withDataSourceProviderFn(new PrestoSslDataSourceProvider(
                        jdbcUrl,
                        "username",
                        "pswd",
                        "bucketname",
                        "certificate.jks",
                        "pswd"
                ))
                .withQuery("select * from tablename")
                .withCoder(TableRowJsonCoder.of())
                .withRowMapper(resultSet -> {
                    ResultSetMetaData meta = resultSet.getMetaData();
                    TableRow outputRow = new TableRow();
                    for (int i = 1; i <= meta.getColumnCount(); i++) {
                        outputRow.set(meta.getColumnName(i), resultSet.getObject(i));
                    }
                    return outputRow;
                }))
        .apply("Write to BigQuery",
                BigQueryIO.writeTableRows()
                        .withoutValidation()
                        .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_NEVER)
                        .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND)
                        .to("projectID:dataset.tablename"));

• 提前给Beam Worker使用的服务账号授予GCS证书存储桶的storage.objects.get权限，否则会出现证书下载失败的问题。
• 如果使用的是Trino JDBC驱动而非旧版Presto驱动，SSL配置键名规则一致，注意不要写错拼写和大小写。
• 临时证书文件无需手动删除，JVM退出时会自动清理临时目录下的文件。

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