Alpakka Kafka流在连接错误后无法重启，尽管使用了RestartSource

import akka.actor.ActorSystem
import akka.kafka.scaladsl.Consumer
import akka.kafka.{ConsumerSettings, Subscriptions}
import akka.stream.ActorMaterializer
import akka.stream.Supervision
import akka.stream.scaladsl.RestartSource
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.common.serialization.StringDeserializer

import scala.concurrent.duration._

object AlpakkaKafkaExample extends App {
  implicit val system: ActorSystem = ActorSystem("AlpakkaKafkaExample")
  implicit val materializer: ActorMaterializer = ActorMaterializer()

  val consumerSettings: ConsumerSettings[String, String] =
    ConsumerSettings(system, new StringDeserializer, new StringDeserializer)
      .withBootstrapServers("localhost:9092")
      .withGroupId("my-group")
      .withProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")

  val supervisionDecider: Supervision.Decider = {
    case _: org.apache.kafka.common.errors.TimeoutException =>
      Supervision.Restart
    case _ =>
      Supervision.Stop
  }

  val kafkaSource = Consumer.plainSource(consumerSettings, Subscriptions.topics("my-topic"))
    .map { record =>
      // 处理Kafka消息
      record.value()
    }

  val restartSource = RestartSource.onFailuresWithBackoff(
    minBackoff = 3.seconds,
    maxBackoff = 30.seconds,
    randomFactor = 0.2
  ) { () =>
    kafkaSource
      .withAttributes(akka.stream.Attributes.supervisionStrategy(supervisionDecider))
  }

  restartSource.runForeach { message =>
    // 处理流中的消息
    println(message)
  }
}