Kafka

# 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

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))
	})
}

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()

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<String, String> 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<String, String> consumer = new KafkaConsumer<>(consumerProps)) {
            consumer.subscribe(Collections.singletonList("orders"));
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(10));
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("received: %s%n", record.value());
            }
        }
    }
}

import { Kafka } from "kafkajs";

const kafka = new Kafka({
  clientId: "kubemq-kafka-example",
  brokers: ["localhost:9092"],
});

async function main(): Promise<void> {
  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);
});

using Confluent.Kafka;

const string bootstrapServers = "localhost:9092";
const string topic = "orders";

var producerConfig = new ProducerConfig { BootstrapServers = bootstrapServers };
using (var producer = new ProducerBuilder<Null, string>(producerConfig).Build())
{
    await producer.ProduceAsync(topic, new Message<Null, string> { 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<Ignore, string>(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();

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

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));
        }
    }
}