# Getting Started (/connectors/kafka/tutorials/getting-started)



Get a message flowing through the KubeMQ Kafka connector in minutes. You point a stock Kafka
client — `kcat`, or any of seven client libraries — at KubeMQ by repointing
`bootstrap.servers`, produce one record to a topic, and consume it back with a consumer
group. There's no KubeMQ SDK and no client-library swap: the connector speaks the real
Produce/Fetch/group-coordinator wire protocol, so any unmodified Kafka client just works. By
the end of this page you'll have run a complete produce-then-consume round-trip against a
local KubeMQ server, using whichever client you already have installed.

## Prerequisites [#prerequisites]

* A running **kubemq-server**, with the Kafka connector **enabled** and reachable on **port
  9092** (plain TCP) — the step below shows how.
* One of the eight clients in the tabs below: `kcat`, or a client library for your language.
  There is no KubeMQ SDK — every example on this page is a stock, unmodified Kafka client.

<Steps>
  <Step>
    ### Enable the connector [#enable-the-connector]

    The Kafka connector is **disabled by default** — a stock kubemq-server does **not** bind
    ports `9092`/`9093` until you turn it on. Enable it with its enable variable:

    <RunKubeMQ ports="[9092, 9093, 50000]" env="{ CONNECTORS_KAFKA_ENABLE: 'true' }" />

    <Callout type="warn">
      **The enable variable is `CONNECTORS_KAFKA_ENABLE`.** A stock server does not serve the Kafka
      wire protocol until you set this to `true`. For Kubernetes, set `spec.kafka.enabled: true` in
      the `KubemqCluster` CR (Helm: `kafka.enabled: true` in your values file).
    </Callout>

    Once enabled, the connector binds two listeners — `9092` for plain TCP and `9093` for TLS —
    and every produced record lands on KubeMQ's auto-selected `next` storage engine. You don't
    need to configure this yourself: on a fresh store, enabling Kafka &#x2A;*auto-selects `next`** with
    no manual `store.engine` step. See [Storage Engines → Zero-config engine
    selection](/configure/reference/storage-engines#zero-config-engine-selection) for the
    full selection rules, including what happens on an existing store.

    You also don't need to pre-create the `orders` topic used below — the connector auto-creates
    a topic on its first `Produce` or `Fetch` call, the same as real Kafka's
    `auto.create.topics.enable` default.
  </Step>

  <Step>
    ### Produce a message [#produce-a-message]

    Every example below produces one record with the value `hello kubemq` to the topic `orders`
    on `bootstrap.servers=localhost:9092` (or, for `kcat`, `-b localhost:9092`). No KubeMQ SDK —
    just a stock Kafka client, repointed.

    <Tabs groupId="language" items="['kcat', 'Go', 'Python', 'Java', 'JavaScript', 'C#', 'Ruby', 'Rust']">
      <Tab value="kcat">
        `kcat` (the librdkafka CLI) needs no client code at all — pipe the payload straight to the
        broker:

        ```bash
        echo "hello kubemq" | kcat -b localhost:9092 -t orders -P
        ```
      </Tab>

      <Tab value="Go">
        The Go example uses `franz-go`, the client this connector's own conformance harness is
        validated against:

        ```go
        package main

        import (
        	"context"
        	"log"

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

        func main() {
        	client, err := kgo.NewClient(
        		kgo.SeedBrokers("localhost:9092"),
        		kgo.DefaultProduceTopic("orders"),
        	)
        	if err != nil {
        		log.Fatalf("client: %v", err)
        	}
        	defer client.Close()

        	record := &kgo.Record{Value: []byte("hello kubemq")}
        	if err := client.ProduceSync(context.Background(), record).FirstErr(); err != nil {
        		log.Fatalf("produce: %v", err)
        	}
        	log.Println("produced: hello kubemq")
        }
        ```
      </Tab>

      <Tab value="Python">
        ```python
        from confluent_kafka import Producer

        producer = Producer({"bootstrap.servers": "localhost:9092"})


        def delivery_report(err, msg):
            if err is not None:
                raise RuntimeError(f"delivery failed: {err}")
            print(f"produced: {msg.value().decode()} (partition {msg.partition()}, offset {msg.offset()})")


        producer.produce("orders", value=b"hello kubemq", callback=delivery_report)
        producer.flush(10)
        ```
      </Tab>

      <Tab value="Java">
        ```java
        import org.apache.kafka.clients.producer.KafkaProducer;
        import org.apache.kafka.clients.producer.ProducerRecord;
        import org.apache.kafka.clients.producer.RecordMetadata;
        import org.apache.kafka.common.serialization.StringSerializer;

        import java.util.Properties;
        import java.util.concurrent.ExecutionException;

        public final class Produce {
            public static void main(String[] args) throws ExecutionException, InterruptedException {
                Properties props = new Properties();
                props.put("bootstrap.servers", "localhost:9092");
                props.put("key.serializer", StringSerializer.class.getName());
                props.put("value.serializer", StringSerializer.class.getName());

                try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
                    RecordMetadata meta = producer.send(new ProducerRecord<>("orders", "hello kubemq")).get();
                    System.out.printf("produced: hello kubemq (partition %d, offset %d)%n",
                            meta.partition(), meta.offset());
                }
            }
        }
        ```
      </Tab>

      <Tab value="JavaScript">
        ```javascript
        const { Kafka } = require("kafkajs");

        const kafka = new Kafka({ brokers: ["localhost:9092"] });
        const producer = kafka.producer();

        async function main() {
          await producer.connect();
          await producer.send({
            topic: "orders",
            messages: [{ value: "hello kubemq" }],
          });
          console.log("produced: hello kubemq");
          await producer.disconnect();
        }

        main().catch((err) => {
          console.error("produce failed:", err);
          process.exit(1);
        });
        ```
      </Tab>

      <Tab value="C#">
        ```csharp
        using Confluent.Kafka;

        var config = new ProducerConfig { BootstrapServers = "localhost:9092" };

        using var producer = new ProducerBuilder<Null, string>(config).Build();

        var result = await producer.ProduceAsync("orders", new Message<Null, string> { Value = "hello kubemq" });
        Console.WriteLine($"produced: hello kubemq (partition {result.Partition}, offset {result.Offset})");
        ```
      </Tab>

      <Tab value="Ruby">
        ```ruby
        require "rdkafka"

        config = Rdkafka::Config.new("bootstrap.servers" => "localhost:9092")
        producer = config.producer

        handle = producer.produce(topic: "orders", payload: "hello kubemq")
        handle.wait(max_wait_timeout_ms: 10_000)
        puts "produced: hello kubemq"

        producer.close
        ```
      </Tab>

      <Tab value="Rust">
        ```rust
        use rdkafka::config::ClientConfig;
        use rdkafka::producer::{FutureProducer, FutureRecord};
        use std::time::Duration;

        #[tokio::main]
        async fn main() {
            let producer: FutureProducer = ClientConfig::new()
                .set("bootstrap.servers", "localhost:9092")
                .create()
                .expect("producer creation failed");

            let record = FutureRecord::<str, _>::to("orders").payload("hello kubemq");
            match producer.send(record, Duration::from_secs(10)).await {
                Ok((partition, offset)) => {
                    println!("produced: hello kubemq (partition {partition}, offset {offset})")
                }
                Err((err, _)) => eprintln!("produce failed: {err}"),
            }
        }
        ```
      </Tab>
    </Tabs>

    Whichever client you ran, the connector auto-created the `orders` topic on that first write and
    appended your record to it. The next step reads it back.
  </Step>

  <Step>
    ### Consume and verify [#consume-and-verify]

    Consume the record back with a consumer group named `orders-group`. Each example joins the
    group, reads one record, and prints its value alongside the partition and offset the
    connector assigned — the offset maps one-to-one onto the underlying Events Store `Sequence`, durable and
    stable across a restart.

    <Tabs groupId="language" items="['kcat', 'Go', 'Python', 'Java', 'JavaScript', 'C#', 'Ruby', 'Rust']">
      <Tab value="kcat">
        `-G` puts `kcat` into consumer-group mode; `-c 1` exits after one message:

        ```bash
        kcat -b localhost:9092 -G orders-group -c 1 orders
        ```
      </Tab>

      <Tab value="Go">
        ```go
        package main

        import (
        	"context"
        	"fmt"
        	"log"
        	"time"

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

        func main() {
        	client, err := kgo.NewClient(
        		kgo.SeedBrokers("localhost:9092"),
        		kgo.ConsumerGroup("orders-group"),
        		kgo.ConsumeTopics("orders"),
        	)
        	if err != nil {
        		log.Fatalf("client: %v", err)
        	}
        	defer client.Close()

        	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
        	defer cancel()

        	fetches := client.PollFetches(ctx)
        	if errs := fetches.Errors(); len(errs) > 0 {
        		log.Fatalf("fetch: %v", errs)
        	}
        	fetches.EachRecord(func(record *kgo.Record) {
        		fmt.Printf("consumed: %s (partition %d, offset %d)\n", record.Value, record.Partition, record.Offset)
        	})
        }
        ```
      </Tab>

      <Tab value="Python">
        ```python
        from confluent_kafka import Consumer

        consumer = Consumer({
            "bootstrap.servers": "localhost:9092",
            "group.id": "orders-group",
            "auto.offset.reset": "earliest",
        })
        consumer.subscribe(["orders"])

        msg = consumer.poll(10.0)
        if msg is None:
            raise SystemExit("no message received within timeout")
        if msg.error():
            raise RuntimeError(msg.error())

        print(f"consumed: {msg.value().decode()} (partition {msg.partition()}, offset {msg.offset()})")
        consumer.close()
        ```
      </Tab>

      <Tab value="Java">
        ```java
        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.common.serialization.StringDeserializer;

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

        public final class Consume {
            public static void main(String[] args) {
                Properties props = new Properties();
                props.put("bootstrap.servers", "localhost:9092");
                props.put("group.id", "orders-group");
                props.put("key.deserializer", StringDeserializer.class.getName());
                props.put("value.deserializer", StringDeserializer.class.getName());
                props.put("auto.offset.reset", "earliest");

                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> record : records) {
                        System.out.printf("consumed: %s (partition %d, offset %d)%n",
                                record.value(), record.partition(), record.offset());
                    }
                }
            }
        }
        ```
      </Tab>

      <Tab value="JavaScript">
        ```javascript
        const { Kafka } = require("kafkajs");

        const kafka = new Kafka({ brokers: ["localhost:9092"] });
        const consumer = kafka.consumer({ groupId: "orders-group" });

        async function main() {
          await consumer.connect();
          await consumer.subscribe({ topic: "orders", fromBeginning: true });

          await consumer.run({
            eachMessage: async ({ partition, message }) => {
              console.log(`consumed: ${message.value.toString()} (partition ${partition}, offset ${message.offset})`);
              await consumer.disconnect();
            },
          });
        }

        main().catch((err) => {
          console.error("consume failed:", err);
          process.exit(1);
        });
        ```
      </Tab>

      <Tab value="C#">
        ```csharp
        using Confluent.Kafka;

        var config = new ConsumerConfig
        {
            BootstrapServers = "localhost:9092",
            GroupId = "orders-group",
            AutoOffsetReset = AutoOffsetReset.Earliest,
        };

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

        var result = consumer.Consume(TimeSpan.FromSeconds(10));
        if (result is null)
        {
            throw new TimeoutException("no message received within timeout");
        }
        Console.WriteLine($"consumed: {result.Message.Value} (partition {result.Partition}, offset {result.Offset})");

        consumer.Close();
        ```
      </Tab>

      <Tab value="Ruby">
        ```ruby
        require "rdkafka"

        config = Rdkafka::Config.new(
          "bootstrap.servers" => "localhost:9092",
          "group.id" => "orders-group",
          "auto.offset.reset" => "earliest"
        )
        consumer = config.consumer
        consumer.subscribe("orders")

        message = consumer.poll(10_000)
        raise "no message received within timeout" if message.nil?

        puts "consumed: #{message.payload} (partition #{message.partition}, offset #{message.offset})"

        consumer.close
        ```
      </Tab>

      <Tab value="Rust">
        ```rust
        use rdkafka::config::ClientConfig;
        use rdkafka::consumer::{BaseConsumer, Consumer};
        use rdkafka::message::Message;
        use std::time::Duration;

        fn main() {
            let consumer: BaseConsumer = ClientConfig::new()
                .set("bootstrap.servers", "localhost:9092")
                .set("group.id", "orders-group")
                .set("auto.offset.reset", "earliest")
                .create()
                .expect("consumer creation failed");

            consumer.subscribe(&["orders"]).expect("subscribe failed");

            match consumer.poll(Duration::from_secs(10)) {
                Some(Ok(message)) => {
                    let payload = message
                        .payload()
                        .map(|p| String::from_utf8_lossy(p).to_string())
                        .unwrap_or_default();
                    println!("consumed: {payload} (partition {}, offset {})", message.partition(), message.offset());
                }
                Some(Err(err)) => eprintln!("consume failed: {err}"),
                None => eprintln!("no message received within timeout"),
            }
        }
        ```
      </Tab>
    </Tabs>

    A successful round-trip prints the record you produced, plus the partition and offset the
    connector assigned it:

    ```text
    produced: hello kubemq
    consumed: hello kubemq (partition 0, offset 0)
    ```
  </Step>
</Steps>

You just repointed a stock Kafka client at KubeMQ, produced a record, and consumed it back
through a real consumer group — the same round-trip you'd run against any Kafka cluster,
with no client-library swap and no code change beyond `bootstrap.servers`. From here, dig
into a single feature end-to-end — producing with keys and durability guarantees, consuming
with manual offset control, or how topics and partitions map onto KubeMQ's storage layer.

## Next steps [#next-steps]

<Cards>
  <Card title="Producing" href="/connectors/kafka/how-to/producing" description="Keys and headers, the acks durability setting, batching, and the idempotent producer, across kcat and seven client libraries." />

  <Card title="Consuming" href="/connectors/kafka/how-to/consuming" description="Consumer groups, committing offsets manually or automatically, and seeking by offset or timestamp." />

  <Card title="Architecture" href="/connectors/kafka/concepts/architecture" description="The 9092/9093 wire-protocol listeners and how topics and partitions map to Events Store logs on the next engine." />

  <Card title="Capabilities" href="/connectors/kafka/reference/capabilities" description="Every implemented Kafka API and its version range, from Produce/Fetch to transactions and share groups." />
</Cards>