我已经在Google群组上问过这个问题，但还没有收到任何回复。所以在这里为不同的受众发布这个。
我们在我们的应用程序中使用了Reactive-Kafka。我们有一个如下的场景，如果在处理消息时发生任何异常，我们希望停止向消费者发送消息。消息应该在规定的时间后或在消费者端的明确请求下重试。使用我们目前的方法，假设消费者的数据库关闭了一段时间，它仍然会尝试从kafka读取并处理消息，但由于DB问题，处理失败。这将使应用程序不必要地忙碌。相反，我们希望暂停消费者以在规定的时间内接收消息（例如，等待30分钟重试）。我们不知道如何处理那个案子。
有可能做同样的事情吗？我错过了什么吗？
下面是从响应式kafka中获取的示例代码：
Consumer.committableSource(consumerSettings, Subscriptions.topics("topic1"))
.mapAsync(1) { msg =>
Future {
/**
* Unreliable consumer, for e.g. saving to DB might not be successful due to DB is down
*/
}.map(_ => msg.committableOffset).recover {
case ex => {
/**
* HOW TO DO ????????
* On exception, I would like to tell stream to stop sending messages and pause the consumer and try again within stipulated time
* or on demand from the last committed offset
*/
throw ex
}
}
}
.batch(max = 20, first => CommittableOffsetBatch.empty.updated(first)) { (batch, elem) =>
batch.updated(elem)
}
.mapAsync(3)(_.commitScaladsl())
.runWith(Sink.ignore)
请注意，您可能需要将src的物化值从akka. kafka.caladsl.消费者.Control映射到akka.Not用，以便在恢复期中引用它：
val src = Consumer.committableSource(consumerSettings, Subscriptions.topics("topic1"))
.mapAsync(1) { msg =>
Future {
/**
* Unreliable consumer, for e.g. saving to DB might not be successful due to DB is down
*/
}.map(_ => msg.committableOffset)
.mapMaterializedValue(_ => akka.NotUsed)
有一个用于此目的的恢复与重试组合器。有关参考，请参阅此答案和文档。
你可以提取你的消息来源
val src = Consumer.committableSource(consumerSettings, Subscriptions.topics("topic1"))
.mapAsync(1) { msg =>
Future {
/**
* Unreliable consumer, for e.g. saving to DB might not be successful due to DB is down
*/
}.map(_ => msg.committableOffset)
然后做
src
.recoverWithRetries(attempts = -1, {case e: MyDatabaseException =>
logger.error(e)
src.delay(30.minutes, DelayOverflowStrategy.backpressure)})
...
（尝试重试=-1意味着无限期重试）
