故障时优雅重启Reactive-Kafka消费者流及旧消费者强制终止方案问询

我之前也碰到过类似的Akka Kafka消费者无法优雅关闭、残留旧实例的问题，结合官方文档和社区实践，给你几个能有效解决的方案：

1. 给Shutdown添加超时+唤醒阻塞的消费者 #

默认的consumerControl.shutdown()可能因为消费者阻塞在poll操作上而无法及时响应，我们可以给shutdown加上超时，超时后主动唤醒消费者打破阻塞，再强制关闭：

streamComplete
  .flatMap { _ =>
    // 给shutdown设置超时，避免无限等待
    consumerControl.shutdown()
      .withTimeout(10.seconds, akka.pattern.Status.Failure(new TimeoutException("Consumer shutdown timed out")))
      .map(_ => logger.info(s"3.consumer id = ${consumer.id} SHUTDOWN at ${DateTime.now} GRACEFULLY:CLOSED FROM UPSTREAM"))
  }
  .recoverWith {
    case _: TimeoutException =>
      // 调用Kafka原生消费者的wakeup()，打断阻塞的poll操作
      consumerControl.asInstanceOf[akka.kafka.Consumer.Control]
        .underlying()
        .ifPresent(_.wakeup())
      // 再次执行shutdown，此时消费者已被唤醒，能正常进入关闭流程
      consumerControl.shutdown()
        .map(_ => logger.info(s"3.consumer id = ${consumer.id} SHUTDOWN at ${DateTime.now} FORCED:CLOSED AFTER TIMEOUT"))
    case ex =>
      logger.error(s"Consumer stream failed: ${ex.getMessage}", ex)
      consumerControl.shutdown()
        .map(_ => logger.info(s"3.consumer id = ${consumer.id} SHUTDOWN at ${DateTime.now} ERROR:CLOSED FROM UPSTREAM"))
  }

2. 维护活跃消费者集合，重启时主动清理 #

利用RestartSource的重启回调，维护一个线程安全的活跃消费者控制对象集合，每次重启前强制清理未关闭的旧消费者：

import java.util.concurrent.ConcurrentHashMap
import scala.jdk.CollectionConverters._

// 线程安全的集合，保存活跃的消费者控制对象
val activeConsumers = new ConcurrentHashMap[String, akka.kafka.Consumer.Control]()

RestartSource.withBackoff(
  minBackoff = 20.seconds,
  maxBackoff = 5.minutes,
  randomFactor = 0.2
) { () => 
  Consumer.committableSource(customizedSettings, subscriptions)
    .watchTermination() { (control, done) =>
      // 保存消费者控制对象到集合
      val consumerId = consumer.id // 这里需要确保能获取到当前消费者的ID
      activeConsumers.put(consumerId, control)
      // 流完成后从集合移除
      done.onComplete { _ =>
        activeConsumers.remove(consumerId)
      }(system.dispatcher)
      done
    }
    .flatMapConcat(sourceFunction)
}
// 重启前清理旧消费者
.onRestart { cause =>
  logger.info(s"Restarting stream due to: ${cause.getMessage}, cleaning up old consumers")
  activeConsumers.values().asScala.foreach { control =>
    if (!control.isShutdown.isCompleted) {
      // 先尝试优雅关闭，超时则强制唤醒
      control.shutdown()
        .withTimeout(5.seconds)
        .recover { case _ =>
          control.underlying().ifPresent(_.wakeup())
          control.shutdown()
        }(system.dispatcher)
    }
  }
}
.viaMat(KillSwitches.single)(Keep.right)
.toMat(Sink.ignore)(Keep.left)
.run

3. 调整消费者配置，降低响应延迟 #

通过缩短poll间隔和设置停止超时，让消费者更易响应关闭信号：

val customizedSettings = ConsumerSettings(system, keyDeserializer, valueDeserializer)
  .withPollInterval(100.millis) // 缩短poll间隔，让消费者更频繁检查关闭信号
  .withStopTimeout(5.seconds) // 设置Akka Kafka消费者的停止超时，超时后强制终止
  .withCloseTimeout(3.seconds) // 设置关闭底层Kafka消费者的超时

额外注意事项 #

• 确保sourceFunction中的HTTP请求是可取消的：比如使用Http().singleRequest时设置请求超时，并且在流终止时取消未完成的请求，避免阻塞流的终止流程。
• 避免在流处理逻辑中做长时间阻塞操作：尽量使用异步API，否则会导致消费者无法及时处理关闭命令。

内容的提问来源于stack exchange，提问作者Rabzu