Kafka client examples

The broker address is SERVER:9092. If authentication is enabled ([[auth.users]] in serve.toml) — add SASL/PLAIN as in the examples below; without it, drop the sasl_*/security.protocol parameters. For access from other machines the broker must have kafka_advertised_addr configured (see configuration).

Connection check

kcat (no auth)

kcat -b SERVER:9092 -L                          # metadata: broker + topics
echo "hello" | kcat -b SERVER:9092 -t demo -P   # produce
kcat -b SERVER:9092 -t demo -C -o beginning -e  # consume from the beginning

kcat (SASL/PLAIN)

kcat -b SERVER:9092 \
     -X security.protocol=SASL_PLAINTEXT -X sasl.mechanism=PLAIN \
     -X sasl.username=app -X sasl.password=<pass> -L

Producer

Python

# pip install confluent-kafka  (librdkafka — verified e2e with Roxa)
from confluent_kafka import Producer

p = Producer({
    "bootstrap.servers": "SERVER:9092",
    # with authentication enabled:
    "security.protocol": "SASL_PLAINTEXT",
    "sasl.mechanism": "PLAIN",
    "sasl.username": "app",
    "sasl.password": "<pass>",
})

p.produce("demo", key="order-1", value=b"hello roxa")
p.flush()  # the ack arrives after the batch is durably written to storage

Go

// go get github.com/twmb/franz-go/pkg/kgo
package main

import (
    "context"
    "github.com/twmb/franz-go/pkg/kgo"
    "github.com/twmb/franz-go/pkg/sasl/plain"
)

func main() {
    cl, err := kgo.NewClient(
        kgo.SeedBrokers("SERVER:9092"),
        // with authentication enabled:
        kgo.SASL(plain.Auth{User: "app", Pass: "<pass>"}.AsMechanism()),
    )
    if err != nil { panic(err) }
    defer cl.Close()

    rec := &kgo.Record{Topic: "demo", Key: []byte("order-1"), Value: []byte("hello roxa")}
    if err := cl.ProduceSync(context.Background(), rec).FirstErr(); err != nil {
        panic(err)
    }
}

Java

// The Java client speaks the same protocol specification, but is not yet exercised
// in Roxa's e2e tests — verify your scenario on the demo environment first.
Properties props = new Properties();
props.put("bootstrap.servers", "SERVER:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
// with authentication enabled:
props.put("security.protocol", "SASL_PLAINTEXT");
props.put("sasl.mechanism", "PLAIN");
props.put("sasl.jaas.config",
    "org.apache.kafka.common.security.plain.PlainLoginModule required " +
    "username=\"app\" password=\"<pass>\";");

try (var producer = new KafkaProducer<String, String>(props)) {
    producer.send(new ProducerRecord<>("demo", "order-1", "hello roxa")).get();
}

Consumer (a group with rebalancing)

Python

from confluent_kafka import Consumer

c = Consumer({
    "bootstrap.servers": "SERVER:9092",
    "group.id": "billing",
    "auto.offset.reset": "earliest",
    # + the same sasl parameters as the producer, if authentication is enabled
})
c.subscribe(["demo"])

while True:
    msg = c.poll(1.0)
    if msg is None:
        continue
    if msg.error():
        raise Exception(msg.error())
    print(msg.key(), msg.value())
    c.commit(msg)

Go

cl, _ := kgo.NewClient(
    kgo.SeedBrokers("SERVER:9092"),
    kgo.ConsumerGroup("billing"),
    kgo.ConsumeTopics("demo"),
)
defer cl.Close()

for {
    fetches := cl.PollFetches(context.Background())
    fetches.EachRecord(func(r *kgo.Record) {
        fmt.Printf("%s = %s\n", r.Key, r.Value)
    })
    cl.CommitUncommittedOffsets(context.Background())
}

Idempotent producer

Enabled the standard way (enable.idempotence=true in librdkafka clients) — Roxa supports InitProducerId and retry deduplication within a session.