Kafka未提交的消息不再被消费

Question

我正在处理 kafka 消息并使用带有手动偏移提交的 spark 流插入 kudu 表，这是我的代码。

val topicsSet = topics.split(",").toSet
val kafkaParams = Map[String, Object](
  ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> brokers,
  ConsumerConfig.GROUP_ID_CONFIG -> groupId,
  ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
  ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
  ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG -> (false: java.lang.Boolean),
  ConsumerConfig.AUTO_OFFSET_RESET_CONFIG -> "earliest" //"latest" //"earliest"
 )
val stream = KafkaUtils.createDirectStream[String, String](
                        ssc,
                        PreferConsistent,
                        Subscribe[String, String](topicsSet, kafkaParams)
                       )
stream.foreachRDD { rdd =>
var offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
//offsetRanges.foreach(println)
var msgOffsetsRdd = rdd.map(msg =>{
val msgOffset = OffsetRange(msg.topic(), msg.partition(),  msg.offset(), msg.offset()+1)
        println(msg)
        msgOffset 
      }
    )
   val msgOffsets = msgOffsetsRdd.collect() //here idea was to get only processed messages offsets for commit
   stream.asInstanceOf[CanCommitOffsets].commitAsync(msgOffsets)
}

让我们列出这个示例在将数据插入 kudu 时出现错误，我需要再次处理这些消息，如果我停止工作并重新启动它，我能够得到未提交的消息，我们不能得到所有未提交的消息吗流媒体？

Answer 1

你有消息，为什么不放一个重试逻辑以防失败。 当您重新连接时，Kafka 会给您相同的消息，以防您的消费者崩溃，不确定 Kafka 是否会在连接仍然打开时给出相同的消息。

如果失败是由于目标数据存储不可用，您可以在代码中包含一些重试逻辑，或者如果插入失败的消息格式不正确，您可以将这些消息保存到临时缓存、数据存储或其他 kafka 主题中以重试稍后或检查这些消息有什么问题。