Akka Streams 在阶段失败的情况下重新创建流

Question

我有一个非常简单的 Akka Streams 流，它使用 alpakka 从 Kafka 读取 msg，对 msg 执行一些操作并将其索引到 Elasticsearch。

我正在使用 CommitableSource，因此我采用了至少一次策略。我仅在对 ES 的索引成功时才提交我的偏移量，如果它失败，我将再次阅读该消息，因为形成最新的已知偏移量。

 val decider: Supervision.Decider = {
    case _:Throwable =>  Supervision.Restart
    case _           => Supervision.Restart
  }

  val config: Config = context.system.settings.config.getConfig("akka.kafka.consumer")

  val flow: Flow[CommittableMessage[String, String], Done, NotUsed] =
    Flow[CommittableMessage[String,String]].
      map(msg => Event(msg.committableOffset,Success(Json.parse(msg.record.value()))))
    .mapAsync(10) { event => indexEvent(event.json.get).map(f=> event.copy(json = f))}
      .mapAsync(10)(f => {
    f.json match {
      case Success(_)=> f.committableOffset.commitScaladsl()
      case Failure(ex) => throw new StreamFailedException(ex.getMessage,ex)
    }
      })

  val r: Flow[CommittableMessage[String, String], Done, NotUsed] = RestartFlow.onFailuresWithBackoff(
    minBackoff = 3.seconds,
    maxBackoff = 3.seconds,
    randomFactor = 0.2, // adds 20% "noise" to vary the intervals slightly
    maxRestarts = 20 // limits the amount of restarts to 20
  )(() => {
    println("Creating flow")
    flow
  })

  val consumerSettings: ConsumerSettings[String, String] =
    ConsumerSettings(config, new StringDeserializer, new StringDeserializer)
      .withBootstrapServers("localhost:9092")
      .withGroupId("group1")
      .withProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")

  val restartSource: Source[CommittableMessage[String, String], NotUsed] = RestartSource.withBackoff(
    minBackoff = 3.seconds,
    maxBackoff = 30.seconds,
    randomFactor = 0.2, // adds 20% "noise" to vary the intervals slightly
    maxRestarts = 20 // limits the amount of restarts to 20
  ) {() =>
    Consumer.committableSource(consumerSettings, Subscriptions.topics("test"))
  }


  implicit val mat: ActorMaterializer = ActorMaterializer(ActorMaterializerSettings(context.system).withSupervisionStrategy(decider))



  restartSource
    .via(flow)
    .toMat(Sink.ignore)(Keep.both).run()

我想要实现的是重新启动整个流程 Source -> Flow-> Sink。如果由于某种原因我无法在 Elastic 中索引消息。

我尝试了以下方法：

• Supervision.Decider - 看起来流程被重新创建但没有 消息是从 Kafka 中提取的，显然是因为它记得它 偏移量。
• RestartSource - 看起来不像以太，因为异常发生在流程阶段。
• RestartFlow - 也无济于事，因为它只重新启动 Flow，但我需要从上次成功的偏移量重新启动 Source。

有什么优雅的方法可以做到这一点吗？

Answer 1

您可以组合可重新启动的源、流和接收器。没有人阻止您为图表的每个部分执行可重新启动的源/流/接收器

更新：

代码示例

val sourceFactory = () => Source(1 to 10).via(Flow.fromFunction(x => { println("problematic flow"); x }))
RestartSource.withBackoff(4.seconds, 4.seconds, 0.2)(sourceFactory)