Get started with Krabka

// Maven dependency: org.apache.kafka:kafka-clients:4.3.1
// Produce a record with the standard Apache Kafka Java client.
import org.apache.kafka.clients.producer.*;
import java.util.Properties;

public class App {
    public static void main(String[] args) throws InterruptedException, java.util.concurrent.ExecutionException {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        Producer<String, String> producer = new KafkaProducer<>(props);
        producer.send(new ProducerRecord<>("quickstart-events", "user-1", "Hello Krabka from Java!")).get();
        System.out.println("Record delivered to quickstart-events.");
        producer.close();
    }
}

# python -m pip install confluent-kafka==2.15.1
# Save as app.py and run: python app.py
from confluent_kafka import Producer

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

def delivered(error, message):
    if error is not None:
        raise RuntimeError(str(error))
    print(f"Delivered to partition {message.partition()} at offset {message.offset()}.")

producer.produce(
    "quickstart-events",
    key="user-1",
    value="Hello Krabka from Python!",
    on_delivery=delivered,
)
if producer.flush(30) != 0:
    raise TimeoutError("Message delivery timed out")

// go get github.com/twmb/franz-go/pkg/kgo
// Produce a record with the franz-go Kafka client.
package main

import (
    "context"
    "fmt"

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

func main() {
    cl, err := kgo.NewClient(kgo.SeedBrokers("localhost:9092"))
    if err != nil { panic(err) }
    defer cl.Close()

    rec := &kgo.Record{Topic: "quickstart-events", Key: []byte("user-1"), Value: []byte("Hello Krabka from Go!")}
    if err := cl.ProduceSync(context.Background(), rec).FirstErr(); err != nil { panic(err) }
    fmt.Println("Produced message to Krabka!")
}

// npm install [email protected]
// Save as app.cjs and run: node app.cjs
const { Kafka } = require('kafkajs');

async function main() {
    const kafka = new Kafka({ clientId: 'quickstart', brokers: ['localhost:9092'] });
    const producer = kafka.producer();
    await producer.connect();
    try {
        await producer.send({
            topic: 'quickstart-events',
            messages: [{ key: 'user-1', value: 'Hello Krabka from KafkaJS!' }],
        });
        console.log('Record delivered to quickstart-events.');
    } finally {
        await producer.disconnect();
    }
}

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

// Cargo.toml:
//   krabka-client-producer = { git = "https://github.com/krabka-io/krabka-client-rs", rev = "c4ce581587772b3750d88c61f3389d9626a20235" }
//   plus the [patch.crates-io] block from that revision's Cargo.toml (see /docs/streams-rs)
//   bytes = "1"
//   tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
use bytes::Bytes;
use krabka_client_producer::{Acks, Producer, ProducerRecord};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let producer = Producer::builder()
        .bootstrap("localhost:9092")
        .acks(Acks::All)
        .build()
        .await?;

    let metadata = producer
        .send(ProducerRecord {
            topic: "quickstart-events".into(),
            key: Some(Bytes::from("user-1")),
            value: Some(Bytes::from("Hello Krabka from Rust!")),
            ..Default::default()
        })
        .await?;

    producer.flush().await?;
    producer.close().await?;
    println!("Delivered to partition {} at offset {}.", metadata.partition, metadata.offset);
    Ok(())
}

KAFKA_TOOLS=docker.io/apache/kafka@sha256:01b9a4030e54c6068e66eb3ba4cb82c0d89238629ef1c30d79b86036bf89b1b7
KAFKA_NETWORK="container:$(docker compose ps -q broker)"
docker run --rm --network "$KAFKA_NETWORK" "$KAFKA_TOOLS" /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 --create --if-not-exists --topic quickstart-events --partitions 1 --replication-factor 1
printf 'hello krabka\n' | docker run --rm -i --network "$KAFKA_NETWORK" "$KAFKA_TOOLS" \
  /opt/kafka/bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic quickstart-events
docker run --rm --network "$KAFKA_NETWORK" "$KAFKA_TOOLS" /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 --topic quickstart-events --from-beginning --max-messages 1 --timeout-ms 30000