# Configure offset when consuming kafka topic

**URL:** <https://discuss.akka.io/t/configure-offset-when-consuming-kafka-topic/6222>\
**Category:** Akka Streams & Alpakka\
**Tags:** scala, kafka, alpakka\
**Created:** [April 6, 2020, 6:40pm UTC](https://discuss.akka.io/t/configure-offset-when-consuming-kafka-topic/6222 "2020-04-06T18:40:36Z")\
**Posts on this page:** 2\
**Page:** 1

<div class="post-metadata">

**Author:** ![vasily802](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/vasily802/32/1270_2.png) [@vasily802](https://discuss.akka.io/u/vasily802)\
**Post date:** [April 6, 2020, 6:40pm UTC](https://discuss.akka.io/t/configure-offset-when-consuming-kafka-topic/6222/1 "2020-04-06T18:40:36Z")

</div>

I am using Alpakka-kafka in scala to consume a Kafka topic. Here’s my code:

```auto
    val kafkaConsumerSettings: ConsumerSettings[String, String] =
      ConsumerSettings(actorSystem, new StringDeserializer, new StringDeserializer)
        .withBootstrapServers(kafkaConfig.server)
        .withGroupId(kafkaConfig.group)
        .withProperties(
          ConsumerConfig.MAX_POLL_RECORDS_CONFIG -> "100",
          ConsumerConfig.AUTO_OFFSET_RESET_CONFIG -> "earliest",
          CommonClientConfigs.SECURITY_PROTOCOL_CONFIG -> "SSL"
        )

    Consumer
        .plainSource(kafkaConsumerSettings, Subscriptions.topics(kafkaConfig.topic))
        .runWith(Sink.foreach(println))

```

However, consumer only starts polling from the first uncommitted message in topic. I would like to always start from offset 0, regardless of messages being committed. With Alpakka consumer, how do I specify offset manually?

---

<div class="post-metadata">

**Author:** ![seglo](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/seglo/32/1767_2.png) [@seglo](https://discuss.akka.io/u/seglo)\
**Post date:** [April 6, 2020, 8:37pm UTC](https://discuss.akka.io/t/configure-offset-when-consuming-kafka-topic/6222/2 "2020-04-06T20:37:48Z")

</div>

Hi @vasily802,

You have several different options here in Alpakka Kafka, but basically you must seek to offset 0 (or seek to earliest) explicitely. However, the configuration you have in your example:

```auto
ConsumerConfig.AUTO_OFFSET_RESET_CONFIG -> "earliest",

```

… should work if you’re not committing offsets back to Kafka. This consumer property says that if a committed offset isn’t found within Kafka for the `group.id` and subscribed partition specified, then start from the earliest available offset. If you never commit offsets then I would expect any time you run this code that it would begin from the earliest offset. The earliest offset may be 0 or something else, depending on the broker retention settings for that topic.

Another way to start from the earliest offset is to implement your own [`PartitionAssignmentHandler`](https://doc.akka.io/api/alpakka-kafka/snapshot/akka/kafka/scaladsl/PartitionAssignmentHandler.html) and in the `onAssign` method use the `RestrictedConsumer` instance to [`seekToBeginning`](https://doc.akka.io/api/alpakka-kafka/snapshot/akka/kafka/RestrictedConsumer.html#seekToBeginning(tps:java.util.Collection%5Borg.apache.kafka.common.TopicPartition%5D):Unit).

A third way is to manage offsets yourself and use the `Consumer.plainPartitionedManualOffsetSource` to seek to a specific offset manually. See the Alpakka Kafka [“Offset storage external to Kafka”](https://doc.akka.io/docs/alpakka-kafka/current/consumer.html#offset-storage-external-to-kafka) docs for examples. You could pair this option with the [`MetadataClient`](https://doc.akka.io/docs/alpakka-kafka/current/consumer-metadata.html) to get the beginning offsets for your topic subscription and then seek to it in the `getOffsetsOnAssign` function passed to `plainPartitionedManualOffsetSource`.