从检查点启动时,Flink 无法为 kafka 消费者恢复操作员状态
Flink unable to restore operator state for a kafka consumer when starting from checkpoint
我们有一个 Streaming Job,它有 20 个独立的管道,每个管道都有 one/many 个 Kafka 主题源。
当我们使用新的 jar(我又添加了一个管道)和 AllowNonRestoredState=true 重新启动作业时,我们注意到从检查点恢复 Operatorstate 的奇怪行为。
- 我们当前用于添加管道的配置是静态的,我们基本上是在更改代码以添加任何新管道。
- 我们还没有为任何 Operator 设置任何 UID。
- 当我们从检查点使用相同的 jar 重新启动管道时,我们看到它恢复了所有操作符状态,没有任何丢失状态。
- 我们为所有运算符保持 1 的并行度。
- Kafka 的一些源 Operator 无法使用日志
org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase [] - Consumer subtask 0 has no restore state.
恢复其 Operator 状态
由于这个 Kafka 消费者偏移量被设置为所有分区的 EARLIEST。
Flink Version:1.13.0
这是 Flink 中的已知问题吗?
注意 - 在某些情况下,我们在不同组的两个管道中有相同的主题。
根据我的理解,这不应影响任何状态恢复,因为 Kafka 联合状态位于管道的每个 Kafka 消费者源上。
由于您没有为操作员明确提供 UID,因此您依赖于自动生成的 UID。只要作业的拓扑结构保持不变,它们就是稳定的。添加新管道时,这可能会更改部分或全部以前的 auto-generated UID,并使该状态无法恢复。
如果您想确保能够恢复状态,请在所有有状态运算符上设置 UID。详情参见 Flink 的 Production Readiness Checklist。
如果您想显式设置与当前 auto-generated 匹配的 UID,以便您可以安全地改进作业,您可以通过检查 运行 作业找到每个运算符的散列 UID REST API(每个运算符的 vertexId 是其哈希 UID)。然后,您可以在代码中的相同运算符上将这些散列 UID 与 setUidHash()
结合使用。请参阅 Flink 文档中的 Matching Operator State。
我们有一个 Streaming Job,它有 20 个独立的管道,每个管道都有 one/many 个 Kafka 主题源。
当我们使用新的 jar(我又添加了一个管道)和 AllowNonRestoredState=true 重新启动作业时,我们注意到从检查点恢复 Operatorstate 的奇怪行为。
- 我们当前用于添加管道的配置是静态的,我们基本上是在更改代码以添加任何新管道。
- 我们还没有为任何 Operator 设置任何 UID。
- 当我们从检查点使用相同的 jar 重新启动管道时,我们看到它恢复了所有操作符状态,没有任何丢失状态。
- 我们为所有运算符保持 1 的并行度。
- Kafka 的一些源 Operator 无法使用日志
org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase [] - Consumer subtask 0 has no restore state.
恢复其 Operator 状态 由于这个 Kafka 消费者偏移量被设置为所有分区的 EARLIEST。
Flink Version:1.13.0
这是 Flink 中的已知问题吗?
注意 - 在某些情况下,我们在不同组的两个管道中有相同的主题。 根据我的理解,这不应影响任何状态恢复,因为 Kafka 联合状态位于管道的每个 Kafka 消费者源上。
由于您没有为操作员明确提供 UID,因此您依赖于自动生成的 UID。只要作业的拓扑结构保持不变,它们就是稳定的。添加新管道时,这可能会更改部分或全部以前的 auto-generated UID,并使该状态无法恢复。
如果您想确保能够恢复状态,请在所有有状态运算符上设置 UID。详情参见 Flink 的 Production Readiness Checklist。
如果您想显式设置与当前 auto-generated 匹配的 UID,以便您可以安全地改进作业,您可以通过检查 运行 作业找到每个运算符的散列 UID REST API(每个运算符的 vertexId 是其哈希 UID)。然后,您可以在代码中的相同运算符上将这些散列 UID 与 setUidHash()
结合使用。请参阅 Flink 文档中的 Matching Operator State。