Flink、Kafka和JDBC水槽

Question

我有一个Flink 1.11作业，它使用来自Kafka主题的消息，键入它们，过滤它们（keyBy后跟自定义ProcessFunction），并通过JDBC接收器将它们保存到db中（如下所述：https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/connectors/jdbc.html）

Kafka消费者使用以下选项初始化：

properties.setProperty("auto.offset.reset", "earliest")
kafkaConsumer = new FlinkKafkaConsumer(topic, deserializer, properties)
kafkaConsumer.setStartFromGroupOffsets()
kafkaConsumer.setCommitOffsetsOnCheckpoints(true)

在集群上启用了检查点。

我想要实现的是保证将所有过滤后的数据保存到数据库中，即使数据库关闭了6个小时，或者在保存到数据库时出现编程错误并且需要更新、重新部署和重新启动作业。

要做到这一点，Kafka偏移的任何检查点都应该意味着

1. 从Kafka读取的数据处于Flink操作符状态，等待过滤/传递到接收器，并将作为Flink操作符检查点的一部分进行检查，或

在查看JdbcSink的实现时，我发现它并没有真正保留任何将被检查点/恢复的内部状态——相反，它的检查点是对数据库的写入。现在，如果此写入在检查点期间失败，并且Kafka偏移量确实被保存，我将处于“丢失”数据的情况——从Kafka的后续读取将从已提交的偏移量恢复，并且当db写入失败时正在飞行的任何数据现在不再从Kafka读取，也不在db中。

那么，有没有办法停止推进Kafka偏移量，只要管道已满（Kafka-

Answer 1

我可以看到3个选项：

1. 使用您的Flink版本试用JDBC 1.13连接器。它很有可能只是工作。
2. 如果这不能立即起作用，请检查是否可以将其反向移植到1.11。不应该有太多更改。
3. 编写您自己的2阶段提交接收器，可以通过扩展TwoPhaseCONSinkFunction或使用Checkpoint edFunction和Checkpoint Listener实现您自己的SinkFunction。基本上，您在成功的检查点后创建一个新事务，并使用通知检查点完成提交它。