Transactions & EOS

package main

import (
	"context"
	"log"

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

func main() {
	ctx := context.Background()

	cl, err := kgo.NewClient(
		kgo.SeedBrokers("localhost:9092"),
		kgo.TransactionalID("orders-producer"),
	)
	if err != nil {
		log.Fatalf("client: %v", err)
	}
	defer cl.Close()

	// BeginTransaction lazily drives InitProducerId on the first call.
	if err := cl.BeginTransaction(); err != nil {
		log.Fatalf("begin transaction: %v", err)
	}

	record := &kgo.Record{Topic: "orders", Value: []byte("txn-value")}
	if err := cl.ProduceSync(ctx, record).FirstErr(); err != nil {
		_ = cl.EndTransaction(ctx, kgo.TryAbort)
		log.Fatalf("produce: %v", err)
	}

	if err := cl.EndTransaction(ctx, kgo.TryCommit); err != nil {
		log.Fatalf("commit: %v", err)
	}
	log.Println("committed: txn-value")
}

from confluent_kafka import KafkaException, Producer

producer = Producer({
    "bootstrap.servers": "localhost:9092",
    "transactional.id": "orders-producer",
})
producer.init_transactions()

producer.begin_transaction()
try:
    producer.produce("orders", value=b"txn-value")
    producer.commit_transaction()
    print("committed: txn-value")
except KafkaException as err:
    if err.args[0].txn_requires_abort():
        producer.abort_transaction()
    else:
        raise

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.errors.ProducerFencedException;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public final class TransactionalProduce {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("transactional.id", "orders-producer");
        props.put("key.serializer", StringSerializer.class.getName());
        props.put("value.serializer", StringSerializer.class.getName());

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            producer.initTransactions();
            try {
                producer.beginTransaction();
                producer.send(new ProducerRecord<>("orders", "txn-value"));
                producer.commitTransaction();
                System.out.println("committed: txn-value");
            } catch (ProducerFencedException fenced) {
                // A zombie holding a stale (PID, epoch) cannot recover from this — give up.
                throw fenced;
            } catch (Exception e) {
                producer.abortTransaction();
            }
        }
    }
}

const { Kafka } = require("kafkajs");

const kafka = new Kafka({ brokers: ["localhost:9092"] });
const producer = kafka.producer({
  transactionalId: "orders-producer",
  maxInFlightRequests: 1,
  idempotent: true,
});

async function main() {
  await producer.connect();

  const transaction = await producer.transaction();
  try {
    await transaction.send({ topic: "orders", messages: [{ value: "txn-value" }] });
    await transaction.commit();
    console.log("committed: txn-value");
  } catch (err) {
    await transaction.abort();
    throw err;
  } finally {
    await producer.disconnect();
  }
}

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

using Confluent.Kafka;

var config = new ProducerConfig
{
    BootstrapServers = "localhost:9092",
    TransactionalId = "orders-producer",
};

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

producer.BeginTransaction();
try
{
    producer.Produce("orders", new Message<Null, string> { Value = "txn-value" });
    producer.CommitTransaction();
    Console.WriteLine("committed: txn-value");
}
catch (KafkaException)
{
    producer.AbortTransaction();
    throw;
}

use rdkafka::config::ClientConfig;
use rdkafka::producer::{BaseProducer, BaseRecord, Producer};
use rdkafka::util::Timeout;
use std::time::Duration;

fn main() {
    let producer: BaseProducer = ClientConfig::new()
        .set("bootstrap.servers", "localhost:9092")
        .set("transactional.id", "orders-producer")
        .set("enable.idempotence", "true")
        .create()
        .expect("producer creation failed");

    producer.init_transactions(Timeout::Never).expect("init_transactions failed");
    producer.begin_transaction().expect("begin_transaction failed");

    producer
        .send(BaseRecord::to("orders").payload("txn-value").key("order-1"))
        .expect("send failed");

    producer.flush(Duration::from_secs(10)).expect("flush failed");
    match producer.commit_transaction(Timeout::Never) {
        Ok(()) => println!("committed: txn-value"),
        Err(err) => {
            eprintln!("commit failed, aborting: {err}");
            producer
                .abort_transaction(Duration::from_secs(10))
                .expect("abort_transaction failed");
        }
    }
}