Consuming

# Consume as part of a consumer group. kcat auto-commits on its own interval; there's no
# per-message manual-commit flag on the CLI — pass -X enable.auto.commit=false to disable
# commits entirely instead of committing per message.
kcat -b localhost:9092 -t orders -C -G my-group -o beginning

# Seek to a specific offset (non-group mode; the read starts there).
kcat -b localhost:9092 -t orders -C -p 0 -o 100 -c 5

# Seek to the first offset after a timestamp (ms since epoch).
kcat -b localhost:9092 -t orders -C -p 0 -o s@1700000000000 -c 5

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/twmb/franz-go/pkg/kadm"
	"github.com/twmb/franz-go/pkg/kgo"
)

func main() {
	ctx := context.Background()

	cl, err := kgo.NewClient(
		kgo.SeedBrokers("localhost:9092"),
		kgo.ConsumerGroup("my-group"),
		kgo.ConsumeTopics("orders"),
		kgo.DisableAutoCommit(), // manual commit below; drop this to auto-commit instead
	)
	if err != nil {
		panic(err)
	}
	defer cl.Close()

	fetches := cl.PollFetches(ctx)
	fetches.EachRecord(func(r *kgo.Record) {
		fmt.Printf("partition=%d offset=%d value=%s\n", r.Partition, r.Offset, string(r.Value))
	})
	if err := cl.CommitUncommittedOffsets(ctx); err != nil { // manual commit
		panic(err)
	}

	// Seek to a specific offset. Safe here because no PollFetches is in flight and the
	// group isn't mid-rebalance — see SetOffsets' docs for the caveats.
	cl.SetOffsets(map[string]map[int32]kgo.EpochOffset{
		"orders": {0: {Epoch: -1, Offset: 0}},
	})

	// Seek to the first offset after a timestamp: resolve it via ListOffsetsAfterMilli,
	// then feed the result back into SetOffsets.
	adm := kadm.NewClient(cl)
	oneHourAgo := time.Now().Add(-time.Hour).UnixMilli()
	listed, err := adm.ListOffsetsAfterMilli(ctx, oneHourAgo, "orders")
	if err != nil {
		panic(err)
	}
	offsets := make(map[string]map[int32]kgo.EpochOffset)
	listed.Each(func(o kadm.ListedOffset) {
		if offsets[o.Topic] == nil {
			offsets[o.Topic] = make(map[int32]kgo.EpochOffset)
		}
		offsets[o.Topic][o.Partition] = kgo.EpochOffset{Epoch: o.LeaderEpoch, Offset: o.Offset}
	})
	cl.SetOffsets(offsets)
}

import time

from confluent_kafka import Consumer, TopicPartition

conf = {
    "bootstrap.servers": "localhost:9092",
    "group.id": "my-group",
    "enable.auto.commit": False,  # manual commit below; True to auto-commit instead
    "auto.offset.reset": "earliest",
}
consumer = Consumer(conf)
consumer.subscribe(["orders"])

msg = consumer.poll(timeout=10.0)
if msg is not None and msg.error() is None:
    print(f"partition={msg.partition()} offset={msg.offset()} value={msg.value()!r}")
    consumer.commit(message=msg)  # manual commit

# Seek to a specific offset (valid once the partition is actively assigned).
consumer.seek(TopicPartition("orders", 0, 0))

# Seek to the first offset after a timestamp (ms since epoch).
one_hour_ago_ms = int((time.time() - 3600) * 1000)
resolved = consumer.offsets_for_times([TopicPartition("orders", 0, one_hour_ago_ms)])
for tp in resolved:
    if tp.offset >= 0:
        consumer.seek(tp)

consumer.close()

import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
import org.apache.kafka.common.TopicPartition;

public final class Main {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "my-group");
        props.put("enable.auto.commit", "false"); // manual commit below; "true" to auto-commit
        props.put("auto.offset.reset", "earliest");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            consumer.subscribe(List.of("orders"));

            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(10));
            for (ConsumerRecord<String, String> r : records) {
                System.out.printf("partition=%d offset=%d value=%s%n", r.partition(), r.offset(), r.value());
            }
            consumer.commitSync(); // manual commit of the positions just polled

            TopicPartition tp = new TopicPartition("orders", 0);
            consumer.seek(tp, 0); // seek to a specific offset

            // Seek to the first offset after a timestamp (ms since epoch).
            long oneHourAgo = System.currentTimeMillis() - 3_600_000;
            Map<TopicPartition, OffsetAndTimestamp> found = consumer.offsetsForTimes(Map.of(tp, oneHourAgo));
            OffsetAndTimestamp match = found.get(tp);
            if (match != null) {
                consumer.seek(tp, match.offset());
            }
        }
    }
}

import { Kafka } from "kafkajs";

async function main(): Promise<void> {
  const kafka = new Kafka({ brokers: ["localhost:9092"] });
  const admin = kafka.admin();
  const consumer = kafka.consumer({ groupId: "my-group" });

  await consumer.connect();
  await consumer.subscribe({ topics: ["orders"], fromBeginning: true });

  await consumer.run({
    autoCommit: false, // manual commit below; true auto-commits instead
    eachMessage: async ({ topic, partition, message }) => {
      console.log(`partition=${partition} offset=${message.offset} value=${message.value?.toString()}`);
      await consumer.commitOffsets([
        { topic, partition, offset: (Number(message.offset) + 1).toString() },
      ]);
    },
  });

  // Seek to a specific offset — any in-flight batch for that partition is discarded.
  consumer.seek({ topic: "orders", partition: 0, offset: "0" });

  // Seek to the first offset after a timestamp (ms since epoch).
  await admin.connect();
  const oneHourAgo = Date.now() - 3_600_000;
  const resolved = await admin.fetchTopicOffsetsByTimestamp("orders", oneHourAgo);
  for (const { partition, offset } of resolved) {
    consumer.seek({ topic: "orders", partition, offset });
  }
  await admin.disconnect();
}

main().catch((err) => {
  console.error(err);
  process.exit(1);
});

using Confluent.Kafka;

var config = new ConsumerConfig
{
    BootstrapServers = "localhost:9092",
    GroupId = "my-group",
    EnableAutoCommit = false, // manual commit below; true to auto-commit instead
    AutoOffsetReset = AutoOffsetReset.Earliest,
};

using var consumer = new ConsumerBuilder<string, string>(config).Build();
consumer.Subscribe("orders");

var result = consumer.Consume(TimeSpan.FromSeconds(10));
if (result != null)
{
    Console.WriteLine($"partition={result.Partition.Value} offset={result.Offset.Value} value={result.Message.Value}");
    consumer.Commit(result); // manual commit
}

// Seek to a specific offset.
consumer.Seek(new TopicPartitionOffset("orders", new Partition(0), new Offset(0)));

// Seek to the first offset after a timestamp (ms since epoch).
var oneHourAgo = DateTime.UtcNow.AddHours(-1);
var found = consumer.OffsetsForTimes(
    new[] { new TopicPartitionTimestamp("orders", new Partition(0), new Timestamp(oneHourAgo)) },
    TimeSpan.FromSeconds(10));
if (found.Count > 0 && found[0].Offset.Value >= 0)
{
    consumer.Seek(found[0]);
}

require "rdkafka"

config = Rdkafka::Config.new(
  :"bootstrap.servers"  => "localhost:9092",
  :"group.id"           => "my-group",
  :"enable.auto.commit" => false, # manual commit below; true to auto-commit instead
  :"auto.offset.reset"  => "earliest",
)
consumer = config.consumer
consumer.subscribe("orders")

consumer.each do |message|
  puts "partition=#{message.partition} offset=#{message.offset} value=#{message.payload}"
  consumer.commit # manual commit of the current position
  break
end

# Seek to a specific offset (the next poll on that partition resumes there).
consumer.seek_by("orders", 0, 0)

# Seek to the first offset after a timestamp (ms since epoch).
one_hour_ago_ms = (Time.now.to_i - 3600) * 1000
query = Rdkafka::Consumer::TopicPartitionList.new
query.add_topic_and_partitions_with_offsets("orders", 0 => one_hour_ago_ms)
consumer.offsets_for_times(query).to_h.each do |topic, partitions|
  partitions.each { |p| consumer.seek_by(topic, p.partition, p.offset) if p.offset }
end

use std::time::{Duration, SystemTime, UNIX_EPOCH};

use rdkafka::config::ClientConfig;
use rdkafka::consumer::{CommitMode, Consumer, StreamConsumer};
use rdkafka::message::Message;
use rdkafka::topic_partition_list::{Offset, TopicPartitionList};
use rdkafka::util::Timeout;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let consumer: StreamConsumer = ClientConfig::new()
        .set("bootstrap.servers", "localhost:9092")
        .set("group.id", "my-group")
        .set("enable.auto.commit", "false") // manual commit below; "true" to auto-commit
        .set("auto.offset.reset", "earliest")
        .create()?;
    consumer.subscribe(&["orders"])?;

    let msg = consumer.recv().await?;
    println!(
        "partition={} offset={} value={:?}",
        msg.partition(),
        msg.offset(),
        msg.payload().map(String::from_utf8_lossy)
    );
    consumer.commit_message(&msg, CommitMode::Sync)?; // manual commit

    // Seek to a specific offset.
    consumer.seek("orders", 0, Offset::Offset(0), Timeout::After(Duration::from_secs(5)))?;

    // Seek to the first offset after a timestamp (ms since epoch).
    let one_hour_ago_ms = SystemTime::now().duration_since(UNIX_EPOCH)?.as_millis() as i64 - 3_600_000;
    let mut query = TopicPartitionList::new();
    query.add_partition_offset("orders", 0, Offset::Offset(one_hour_ago_ms))?;
    let resolved = consumer.offsets_for_times(query, Timeout::After(Duration::from_secs(5)))?;
    for elem in resolved.elements() {
        if let Offset::Offset(o) = elem.offset() {
            consumer.seek("orders", elem.partition(), Offset::Offset(o), Timeout::After(Duration::from_secs(5)))?;
        }
    }
    Ok(())
}