在Alpakka Kafka中，可以使用Consumer.committableSource方法来消费消息并处理偏移量。以下是一个包含代码示例的解决方法：

首先，确保在build.sbt文件中添加以下依赖项：

libraryDependencies += "com.typesafe.akka" %% "akka-stream-kafka" % "2.0.5"

然后，创建一个消费者并处理偏移量的代码示例：

import akka.actor.ActorSystem
import akka.kafka.ConsumerSettings
import akka.kafka.scaladsl.Consumer
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.Sink
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.common.serialization.StringDeserializer

object KafkaConsumerExample extends App {

  implicit val system = ActorSystem("kafka-consumer-example")
  implicit val materializer = ActorMaterializer()

  val kafkaConfig = system.settings.config.getConfig("akka.kafka.consumer")
  val consumerSettings = ConsumerSettings(kafkaConfig, new StringDeserializer, new StringDeserializer)
    .withBootstrapServers("localhost:9092")
    .withGroupId("my-group")
    .withProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")

  val topic = "my-topic"

  val control = Consumer.committableSource(consumerSettings, Subscriptions.topics(topic))
    .mapAsync(1) { msg =>
      // 处理消息
      println(s"Received message: ${msg.record.value()}")

      // 手动提交偏移量
      msg.committableOffset.commitScaladsl()
    }
    .to(Sink.ignore)
    .run()

  // 等待消费者停止
  control.shutdown()
}

以上示例中，首先创建了一个ActorSystem和ActorMaterializer。然后，从配置文件加载消费者的配置，并创建ConsumerSettings。在ConsumerSettings中，我们指定了Kafka集群的地址，消费者组ID以及自动偏移重置的配置。

然后，我们使用Consumer.committableSource方法创建一个消费流。在这个流中，我们可以处理每个接收到的消息，并在处理完成后手动提交偏移量。在上面的示例中，我们只是简单地打印接收到的消息，并使用commitScaladsl方法提交偏移量。

最后，我们将消费流连接到一个Sink.ignore，表示我们只关心消费消息，而不关心结果。运行流并使用control.shutdown()停止消费者。

注意：这只是一个简单的示例，用于演示如何在Alpakka Kafka中处理消费者偏移量。在实际应用中，您可能需要添加更多的逻辑来处理错误、重试和异常情况。