Alpakka Kafka 3.0：开启auto.commit=true时能否指定偏移量/日期消费？

结论：可以，即便开启了自动提交偏移量，依然能通过Alpakka Kafka的API从指定偏移量或日期开始消费，但需注意自动提交机制对后续消费位置的影响。

1. 指定偏移量消费 #

通过Subscriptions.assignmentWithOffset方法直接指定目标分区和偏移量，覆盖默认的auto.offset.reset配置。示例代码：

import akka.kafka._
import akka.kafka.scaladsl.Consumer
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.clients.consumer.ConsumerConfig

val consumerSettings = ConsumerSettings(system, new StringDeserializer, new StringDeserializer)
  .withBootstrapServers("localhost:9092")
  .withProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true")
  .withProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")

// 指定要消费的分区和起始偏移量
val targetPartition = TopicPartition("your-topic", 0)
val startOffset = OffsetPosition(targetPartition, 150L)

// 构建从指定偏移量开始的消费流
val consumerSource = Consumer.plainSource(consumerSettings, Subscriptions.assignmentWithOffset(startOffset))

注意：首次启动会从指定偏移量开始，但自动提交会持续将当前消费位置写入Kafka的__consumer_offsets主题。若应用重启时，该消费者组已有提交的偏移量记录，会优先使用已提交的位置，而非再次使用指定的起始偏移量。

2. 指定日期消费 #

先将目标日期转换为时间戳，通过Consumer.Control的offsetsForTimes API查询对应时间戳的偏移量，再按指定偏移量的方式启动消费。示例代码：

import akka.kafka.scaladsl.Consumer
import akka.kafka.OffsetPosition
import org.apache.kafka.common.TopicPartition
import java.time.Instant
import scala.concurrent.Future
import scala.util.Success

val targetTimestamp = Instant.parse("2024-05-01T00:00:00Z").toEpochMilli
val targetPartition = TopicPartition("your-topic", 0)

// 先启动一个临时消费流获取Control实例
val (consumerControl, _) = Consumer.plainSource(consumerSettings, Subscriptions.topics("your-topic"))
  .toMat(Sink.ignore)(Keep.both)
  .run()

// 查询目标时间戳对应的偏移量
val offsetFuture: Future[Option[OffsetPosition]] = consumerControl
  .offsetsForTimes(Map(targetPartition -> targetTimestamp))
  .map(result => result.get(targetPartition).map(info => OffsetPosition(targetPartition, info.offset())))

// 根据查询结果启动定向消费
offsetFuture.onComplete {
  case Success(Some(startOffset)) =>
    val targetedSource = Consumer.plainSource(consumerSettings, Subscriptions.assignmentWithOffset(startOffset))
    // 处理消息逻辑
  case _ =>
    // 处理未找到对应偏移量的场景（比如时间戳早于分区最早消息）
}

同样需注意：自动提交会在消费过程中更新偏移量，应用重启时优先使用已提交的位置。

额外注意 #

如果要求每次应用启动都强制从指定位置开始消费，可在启动时调用consumerControl.seek方法重置偏移量，或者临时关闭自动提交，完成初始定位后再开启（但需自行处理偏移量提交逻辑）。

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