# Consuming (/connectors/kafka/how-to/consuming) Consuming from KubeMQ over the Kafka connector is the same client code you already run against Apache Kafka: subscribe with a consumer group, poll, and commit — with no client-library swap and no code change. Every offset a consumer sees maps one-to-one onto the Events Store `Sequence` of the record it read — durable, restart-stable, and identical across every node of a cluster, so resuming a group after a restart or a rebalance lands exactly where it left off. This guide walks group subscription, committing offsets (automatically or by hand), and seeking to a specific offset or timestamp, in `kcat` and seven client libraries. ## Subscribing with a consumer group [#subscribing-with-a-consumer-group] A `group.id` joins the classic consumer-group protocol — `JoinGroup`/`SyncGroup`/`Heartbeat` coordination assigns each member a slice of the topic's partitions, and the group's committed offsets are durable and leader-linearized. Two consumers with different `group.id` values subscribed to the same topic each get their own independent copy of every record and their own offset position — groups don't compete with each other, only members **within** a group do. See [Consumer Groups](/connectors/kafka/concepts/consumer-groups) for the full protocol, generations, and static membership (`group.instance.id`), which lets a restarting consumer rejoin without triggering a rebalance at all. ## Consume, commit, and seek [#consume-commit-and-seek] Every example below joins `my-group`, polls one record from `orders`, commits its offset manually, then demonstrates seeking to a specific offset and to the first offset after a timestamp (backed by the connector's `ListOffsets` API). ```bash # 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 ``` ```go 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) } ``` ```python 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() ``` ```java 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 consumer = new KafkaConsumer<>(props)) { consumer.subscribe(List.of("orders")); ConsumerRecords records = consumer.poll(Duration.ofSeconds(10)); for (ConsumerRecord 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 found = consumer.offsetsForTimes(Map.of(tp, oneHourAgo)); OffsetAndTimestamp match = found.get(tp); if (match != null) { consumer.seek(tp, match.offset()); } } } } ``` ```typescript import { Kafka } from "kafkajs"; async function main(): Promise { 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); }); ``` ```csharp 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(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]); } ``` ```ruby 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 ``` ```rust 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> { 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(()) } ``` ## Committing offsets: automatic or manual [#committing-offsets-automatic-or-manual] Every pinned client defaults to **automatic** commits on a periodic interval (`enable.auto.commit=true`, or kafkajs's `autoCommit` on `consumer.run`). Auto-commit is the simpler default and works fine for workloads that can tolerate re-processing a handful of records after a crash. Disabling it and committing explicitly — as every example above does — trades a small amount of code for control over exactly when a message is considered "done": commit **after** the work that consumes it completes, not right after it's delivered, so a crash between delivery and processing redelivers the message instead of silently losing it. `OffsetCommit` and `OffsetFetch` are durable and leader-linearized on KubeMQ, so a committed offset survives a restart and is visible identically from any node in a cluster. ## Reading from the beginning or latest [#reading-from-the-beginning-or-latest] Where a consumer group with no prior committed offset starts reading is a client-side reset policy, not a KubeMQ setting: | Client | From the beginning | From latest | | ------------------------ | --------------------------------------------------- | -------------------------------- | | `kcat` | `-o beginning` | `-o end` | | franz-go (Go) | `kgo.ConsumeResetOffset(kgo.NewOffset().AtStart())` | `...AtEnd()` | | confluent-kafka (Python) | `"auto.offset.reset": "earliest"` | `"latest"` | | kafka-clients (Java) | `auto.offset.reset=earliest` | `latest` | | kafkajs | `subscribe({ ..., fromBeginning: true })` | `fromBeginning: false` (default) | | Confluent.Kafka (C#) | `AutoOffsetReset.Earliest` | `.Latest` | | rdkafka (Ruby) | `"auto.offset.reset" => "earliest"` | `"latest"` | | rust-rdkafka | `"auto.offset.reset" => "earliest"` | `"latest"` | This policy only applies the **first** time a group has no committed offset for a partition — once a group has committed, it always resumes from there. ## Error quick reference [#error-quick-reference] Most consume-side failures trace back to authorization or a coordinator-wide cap rather than anything about the record being read: | Trigger | Result | | -------------------------------------------------------------------- | --------------------------------------------------------------------------------------- | | More than `MaxGroups` consumer groups active coordinator-wide | New group rejected — see [Limits & rules](/connectors/kafka/reference/limits-and-rules) | | A duplicate or displaced static member (`group.instance.id`) rejoins | `FENCED_INSTANCE_ID(82)` | | Fetching or committing without the required ACL grant | `TOPIC_AUTHORIZATION_FAILED` / `GROUP_AUTHORIZATION_FAILED` | The full error-code table lives in [Error codes](/connectors/kafka/reference/error-codes). ## Related [#related]