Spark-Kafka-Streaming：偏移管理-无法手动提交工作（Java）

Question

我们正在使用JavaInputDStream

我们按照Spark Streaming-Kafka集成指南中的描述实施了偏移管理，但现在我们刚刚意识到偏移管理不适合我们，并且如果当前小批量中出现故障，Stream不会再次读取消息。即使我们跳过这一行，它也不会再次读取消息：

((CanCommitOffsets) stream.inputDStream()).commitAsync(offsetRanges);

我们将代码分解为以下内容，并期望流最终在循环中一次又一次地读取相同的消息，但事实并非如此：

stream.foreachRDD(recordRDD -> {
   final OffsetRange[] offsetRanges = ((HasOffsetRanges) recordRDD.rdd()).offsetRanges();
   if (!recordRDD.isEmpty()) {
      LOGGER.info("Processing some Data: " + recordRDD.rdd().count());
   }
});

消费者配置参数enable.auto.commit设置为false，初始化JavaInputDStream后也会显示在日志中。我们在测试中的嵌入式Kafka Broker和开发阶段的Kafka-Server都面临同样的问题。目前两者都以独立模式运行。

我们尝试的是：

• 代理配置：增加 offsets.commit.timeout.ms
• 使用者/流配置：将隔离级别设置为“read_committed”
• 使用者/流配置：将自动偏移量重置设置为最早
• 火花：将火花、流媒体、非垃圾邮件设置为虚假
• 火花：增加火花的值
• 流：将流相位持续时间调整为比迷你批处理更长的时间
• 流：启用检查点
• 流：更改的位置策略

所有这些都不起作用，看起来我们搜索了整个网络，却没有找到问题的原因。看起来流忽略了enable.auto.commit配置，只是在阅读了当前RDD的消息后提交。无论我们尝试什么，我们的流都只读取每条消息一次。

我遗漏了什么不同的方法或事实吗？

Answer 1

经过更多的测试，我们发现只有在实际批处理期间流停止/崩溃时，手动提交才有效。如果流停止并重新启动，它将再次消耗失败的数据。

因此，我们目前所做的是，每当我们检测到故障时，直接停止流 JavaStreamContext.stop（假）。在此之后，计划程序将再次启动流，调度程序会验证流是否在一段规律的时间内处于活动状态，如果不是，则启动它。

这不是一个优雅的解决方案，但它首先适用于我们。