我正在尝试使用alpakka消费来自kafka的消息。我没有从akka参与者那里得到任何消费者已经停止的错误,但它没有消费任何消息。下面是我的代码
val consumerSettings = ConsumerSettings(GatewaySettings.DataPlatformKafkaConsumer.kafkaConsumer添加了调试语句,下面是我的调
例如,我的kafka消费者(在consumer Group 1中)正在阅读来自Kafka主题A的消息。
现在,如果该消费者在失败之前消费了12条消息。当消费者重新启动,并且现在它有不同的消费者组(即消费者组2)时,问题1 -?在重启时,它会从偏移量(或位置)中停止的地方继续,因为该偏移量是由Kafka和