# Kafka (/connectors/kafka) Point your Kafka clients at KubeMQ by repointing `bootstrap.servers` — no client-library swap, no code change. The **Kafka connector** is a built-in, wire-protocol bridge inside kubemq-server that speaks the real Kafka request/response protocol — length-prefixed frames, flexible versions, the `kmsg` codec — over plain TCP and TLS. Any off-the-shelf Kafka client — Java `kafka-clients`/AdminClient/Spring, librdkafka/`kcat`/`confluent-kafka`, or `franz-go`/`sarama`/`segmentio` — connects to KubeMQ exactly as it would to a real Kafka broker. ## What is the Kafka connector [#what-is-the-kafka-connector] The Kafka connector implements Kafka's Produce/Fetch/group-coordinator/admin request surface directly inside kubemq-server — there's no separate broker process to run and no protocol-translation layer for your application to configure around. Every produced record lands in a persistent, ordered, replayable **Events Store** channel whose `Sequence` maps one-to-one onto a Kafka offset: durable, restart-stable, and identical across every node of a cluster. Key capabilities: * **Native produce/consume with classic consumer groups** — the large majority of everyday Kafka usage, a straight repoint with no client changes. * **OAUTHBEARER/OIDC federated auth**, alongside SASL/PLAIN, SASL/SCRAM, mTLS client certificates, and Kafka ACL enforcement. * **Compacted topics** (`cleanup.policy=compact`), which unlock the compaction-dependent ecosystem — Kafka Connect and Kafka Streams both run against it. * **Transactions and exactly-once semantics (EOS)** — a transactional producer with `read_committed` isolation and producer fencing. * **Static membership** (`group.instance.id`, KIP-345), so a consumer restart rejoins its group without triggering a rebalance. **Opt-in — disabled by default.** A stock kubemq-server does **not** bind ports `9092` / `9093` until you turn the connector on with `CONNECTORS_KAFKA_ENABLE=true` (Docker) or `spec.kafka.enabled: true` (Kubernetes). See [Getting started](/connectors/kafka/tutorials/getting-started). ## How it works [#how-it-works] A Kafka client dials the connector's TCP listener and issues the same request types it would send to a real broker. A `Produce` request is appended to the topic's Events Store log; a `Fetch` request — including one driven by a consumer group's assigned partitions — reads back from that same log. Because each Kafka offset maps one-to-one onto the log's durable `Sequence`, both directions stay durable and ordered through the message broker underneath. *A `Produce` request from a Kafka client is appended to the topic's Events Store log `kafka.{topic}` through the message broker; a `Fetch` request — including one driven by a consumer group's assigned partitions — reads back from that same log, so each Kafka offset maps one-to-one onto the log's durable `Sequence`.* **Runs on the auto-selected `next` storage engine.** Kafka's headline capabilities — compacted topics and the quorum-fsynced ack contract — exist only on KubeMQ's `next` storage engine. On a **fresh** store, enabling Kafka **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. ## Ports & protocol surface [#ports--protocol-surface] | Port | Transport | Protocol | Notes | | ------ | ------------ | ---------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------ | | `9092` | Plain TCP | Kafka wire protocol — `PLAINTEXT` / `SASL_PLAINTEXT` | Default listener. **Opt-in** — bound only when `CONNECTORS_KAFKA_ENABLE=true`. Disable by setting `CONNECTORS_KAFKA_PORT=""`. Must differ from the TLS port. | | `9093` | TLS over TCP | Kafka wire protocol — `SSL` / `SASL_SSL` | Reuses the server-wide `Security` block (cert + key) — there is no Kafka-specific TLS option. **Required for OAUTHBEARER** (refused on plaintext). | Clients read one setting, `bootstrap.servers` (or the client's equivalent), pointed at `localhost:9092` for the plaintext listener. See [Configuration](/connectors/kafka/concepts/configuration) for the opt-in flag, the advertised-host/port pair, and where the full field-by-field settings reference lives. ## Send & receive a message [#send--receive-a-message] The example below produces one message to the `orders` topic and reads it back on `localhost:9092` — the only change versus a real Kafka client is the **`bootstrap.servers` value**. No KubeMQ SDK, no code change: every tab uses a stock Kafka client library. ```bash # Produce one message to topic "orders" echo "hello from kcat" | kcat -P -b localhost:9092 -t orders # Consume from the beginning and exit after one message kcat -C -b localhost:9092 -t orders -o beginning -c 1 ``` ```go package main import ( "context" "fmt" "log" "github.com/twmb/franz-go/pkg/kgo" ) func main() { ctx := context.Background() // bootstrap.servers repoint only — no KubeMQ SDK, no code change. cl, err := kgo.NewClient( kgo.SeedBrokers("localhost:9092"), kgo.ConsumeTopics("orders"), kgo.ConsumeResetOffset(kgo.NewOffset().AtStart()), ) if err != nil { log.Fatal(err) } defer cl.Close() record := &kgo.Record{Topic: "orders", Value: []byte("hello from franz-go")} if err := cl.ProduceSync(ctx, record).FirstErr(); err != nil { log.Fatalf("produce: %v", err) } fmt.Println("produced to orders") fetches := cl.PollFetches(ctx) fetches.EachRecord(func(r *kgo.Record) { fmt.Printf("received: %s\n", string(r.Value)) }) } ``` ```python from confluent_kafka import Consumer, Producer BOOTSTRAP = "localhost:9092" def main() -> None: producer = Producer({"bootstrap.servers": BOOTSTRAP}) producer.produce("orders", value=b"hello from confluent-kafka") producer.flush() print("produced to orders") consumer = Consumer({ "bootstrap.servers": BOOTSTRAP, "group.id": "orders-consumer", "auto.offset.reset": "earliest", }) consumer.subscribe(["orders"]) msg = consumer.poll(timeout=10.0) if msg is not None and msg.error() is None: print(f"received: {msg.value().decode()!r}") consumer.close() if __name__ == "__main__": main() ``` ```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.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.serialization.StringDeserializer; import org.apache.kafka.common.serialization.StringSerializer; import java.time.Duration; import java.util.Collections; import java.util.Properties; public final class Main { public static void main(String[] args) { String bootstrap = "localhost:9092"; Properties producerProps = new Properties(); producerProps.put("bootstrap.servers", bootstrap); producerProps.put("key.serializer", StringSerializer.class.getName()); producerProps.put("value.serializer", StringSerializer.class.getName()); try (KafkaProducer producer = new KafkaProducer<>(producerProps)) { producer.send(new ProducerRecord<>("orders", "hello from kafka-clients")); producer.flush(); System.out.println("produced to orders"); } Properties consumerProps = new Properties(); consumerProps.put("bootstrap.servers", bootstrap); consumerProps.put("group.id", "orders-consumer"); consumerProps.put("key.deserializer", StringDeserializer.class.getName()); consumerProps.put("value.deserializer", StringDeserializer.class.getName()); consumerProps.put("auto.offset.reset", "earliest"); try (KafkaConsumer consumer = new KafkaConsumer<>(consumerProps)) { consumer.subscribe(Collections.singletonList("orders")); ConsumerRecords records = consumer.poll(Duration.ofSeconds(10)); for (ConsumerRecord record : records) { System.out.printf("received: %s%n", record.value()); } } } } ``` ```typescript import { Kafka } from "kafkajs"; const kafka = new Kafka({ clientId: "kubemq-kafka-example", brokers: ["localhost:9092"], }); async function main(): Promise { const producer = kafka.producer(); await producer.connect(); await producer.send({ topic: "orders", messages: [{ value: "hello from kafkajs" }], }); console.log("produced to orders"); await producer.disconnect(); const consumer = kafka.consumer({ groupId: "orders-consumer" }); await consumer.connect(); await consumer.subscribe({ topic: "orders", fromBeginning: true }); await consumer.run({ eachMessage: async ({ message }) => { console.log(`received: ${message.value?.toString()}`); await consumer.disconnect(); process.exit(0); }, }); } main().catch((err) => { console.error(err); process.exit(1); }); ``` ```csharp using Confluent.Kafka; const string bootstrapServers = "localhost:9092"; const string topic = "orders"; var producerConfig = new ProducerConfig { BootstrapServers = bootstrapServers }; using (var producer = new ProducerBuilder(producerConfig).Build()) { await producer.ProduceAsync(topic, new Message { Value = "hello from Confluent.Kafka" }); Console.WriteLine("produced to orders"); } var consumerConfig = new ConsumerConfig { BootstrapServers = bootstrapServers, GroupId = "orders-consumer", AutoOffsetReset = AutoOffsetReset.Earliest, }; using var consumer = new ConsumerBuilder(consumerConfig).Build(); consumer.Subscribe(topic); var result = consumer.Consume(TimeSpan.FromSeconds(10)); if (result is not null) Console.WriteLine($"received: {result.Message.Value}"); consumer.Close(); ``` ```ruby require "rdkafka" BOOTSTRAP = "localhost:9092" producer_config = Rdkafka::Config.new("bootstrap.servers" => BOOTSTRAP) producer = producer_config.producer producer.produce(topic: "orders", payload: "hello from rdkafka-ruby").wait puts "produced to orders" producer.close consumer_config = Rdkafka::Config.new( "bootstrap.servers" => BOOTSTRAP, "group.id" => "orders-consumer", "auto.offset.reset" => "earliest" ) consumer = consumer_config.consumer consumer.subscribe("orders") consumer.each do |message| puts "received: #{message.payload}" break end consumer.close ``` ```rust use rdkafka::config::ClientConfig; use rdkafka::consumer::{BaseConsumer, Consumer}; use rdkafka::message::Message; use rdkafka::producer::{BaseProducer, BaseRecord, Producer}; use std::time::Duration; fn main() { let bootstrap = "localhost:9092"; let producer: BaseProducer = ClientConfig::new() .set("bootstrap.servers", bootstrap) .create() .expect("producer creation failed"); producer .send( BaseRecord::to("orders") .payload("hello from rust-rdkafka") .key("order-1"), ) .expect("produce failed"); producer.flush(Duration::from_secs(10)).expect("flush failed"); println!("produced to orders"); let consumer: BaseConsumer = ClientConfig::new() .set("bootstrap.servers", bootstrap) .set("group.id", "orders-consumer") .set("auto.offset.reset", "earliest") .create() .expect("consumer creation failed"); consumer.subscribe(&["orders"]).expect("subscribe failed"); if let Some(result) = consumer.poll(Duration::from_secs(10)) { let msg = result.expect("poll failed"); if let Some(payload) = msg.payload() { println!("received: {}", String::from_utf8_lossy(payload)); } } } ``` ## Supported clients [#supported-clients] The connector speaks the genuine Kafka wire protocol, so any conformant client library works — there is no KubeMQ SDK and no proto bindings. The examples above pin one client per language; alternates are noted where the ecosystem commonly uses more than one. | Language | Client library | Notes | | ----------------------- | ------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------- | | kcat | [`kcat`](https://github.com/edenhill/kcat) (librdkafka CLI) | The canonical quick-check Kafka client — no code, just a bootstrap address. | | Go | [`github.com/twmb/franz-go`](https://github.com/twmb/franz-go) | Matches the connector's own conformance-harness client; `sarama` and `segmentio/kafka-go` also work. | | Python | [`confluent-kafka`](https://github.com/confluentinc/confluent-kafka-python) (librdkafka bindings) | | | Java | [`org.apache.kafka:kafka-clients`](https://kafka.apache.org/documentation/#api) | AdminClient and Spring Kafka run on top of the same client unchanged. | | JavaScript / TypeScript | [`kafkajs`](https://kafka.js.org/) | | | C# / .NET | [`Confluent.Kafka`](https://github.com/confluentinc/confluent-kafka-dotnet) (librdkafka bindings) | | | Ruby | [`rdkafka`](https://github.com/karafka/rdkafka-ruby) (librdkafka bindings) | | | Rust | [`rdkafka`](https://github.com/fede1024/rust-rdkafka) (librdkafka bindings) | | **Share groups (KIP-932) are supported in preview, not GA.** Queue-style acquire/acknowledge consumption is implemented and advertised, but only **franz-go** and Java's preview `KafkaShareConsumer` ship a share-consumer API today — GA is pending the multi-client conformance matrix. See the [fitness matrix](/connectors/kafka/reference/fitness-matrix) for the full verdict. Not every capability above is proven to the same tier — see the [fitness matrix](/connectors/kafka/reference/fitness-matrix) for the full supported / caveat / roadmap / unsupported breakdown before you commit to a migration. Every Kafka setting — ports, advertised host/port, SASL mechanisms, OAUTHBEARER, and the advanced tuning knobs — is documented field-by-field in [the Kafka settings reference](/configure/reference/connectors#kafka). Moving an *existing* Kafka, MSK, or Confluent workload onto KubeMQ starts with the read-only `kmq assess kafka` command, then the `kmq migrate` tool — [Migrate from Kafka](/connectors/kafka/how-to/migrate-from-kafka) walks through the full assess → replicate → translate → cutover playbook, and the full [`kmq` CLI reference](/operate/kmq-cli) documents both commands. ## Next steps [#next-steps]