java - 如何在不出现 "No suitable driver found"异常的情况下重新运行 Apache Flink Postgres JDBC 作业

Question

我有一个来自 starter Maven project 的 Flink 工作.该作业具有打开 Postgres JDBC 连接的源。我正在使用 example docker-compose.yml 在我自己的 Flink session 集群上执行作业.
当我第一次提交作业时，它会成功执行。当我再次尝试提交时，出现以下错误:

Caused by: java.sql.SQLException: No suitable driver found for jdbc:postgresql://host.docker.internal:5432/postgres?user=postgres&password=mypassword
    at java.sql.DriverManager.getConnection(DriverManager.java:689)
    at java.sql.DriverManager.getConnection(DriverManager.java:270)
    at com.myorg.project.JdbcPollingSource.run(JdbcPollingSource.java:25)
    at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:110)
    at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:66)
    at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:269)

我必须重新启动我的集群才能重新运行我的工作。为什么会这样？如何在无需重启集群的情况下再次提交我的作业？
Maven 入门项目的唯一补充是:

<dependency>
    <groupId>org.postgresql</groupId>
    <artifactId>postgresql</artifactId>
    <version>42.2.24</version>
</dependency>

Flink源码只是打开一个JDBC连接，如下:

package com.mycompany;

import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.Connection;
import java.sql.DriverManager;

public class JdbcSource extends RichSourceFunction<Integer> {

    private final String connString;

    public JdbcSource(String connString) {
        this.connString = connString;
    }

    @Override
    public void run(SourceContext<Integer> ctx) throws Exception {
        try (Connection conn = DriverManager.getConnection(this.connString)) {

        }
    }

    @Override
    public void cancel() {
    }
}

我已经在 Flink 1.14.0 和 1.13.2 版本上测试过，结果相同。
请注意 this question提供了使用Class.forName("org.postgresql.Driver");的解决方案在我的 RichSourceFunction 内.但是，我想知道发生了什么。

Answer 1

第一个问题可以引用JDBC driver cannot be found when reading a DataSet from an SQL database in Apache Flink .
其次，如果您使用 session 模式。无需重启集群即可轻松重新运行 Flink 作业。您可以登录作业管理器 shell ，然后使用命令重新运行作业。Class.forName("org.postgresql.Driver");将触发静态方法块，因此您的 DriverManager 可以获取驱动程序类。看:

// from org.postgresql.Driver
    static {
        try {
            register();
        } catch (SQLException var1) {
            throw new ExceptionInInitializerError(var1);
        }
    }