Transactional Producer

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092");
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "order-processor-0"); // stable, unique per instance

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions();    // fence any previous zombie with same transactional.id

try {
    producer.beginTransaction();
    // Atomically write to multiple partitions
    producer.send(new ProducerRecord<>("orders-processed", orderId, processedJson));
    producer.send(new ProducerRecord<>("inventory-reserved", itemId, reserveJson));
    // Atomically commit consumer offset as part of the same transaction
    producer.sendOffsetsToTransaction(
        Map.of(new TopicPartition("orders-raw", partition),
               new OffsetAndMetadata(offset + 1)),
        consumerGroupMetadata
    );
    producer.commitTransaction();
} catch (ProducerFencedException e) {
    producer.close();           // this instance has been fenced by a newer one
} catch (KafkaException e) {
    producer.abortTransaction(); // roll back — consumer offset not advanced
    // Retry from the unconsumed offset
}

# Consumer config for EOS
spring.kafka.consumer.properties.isolation.level=read_committed
# (default is read_uncommitted — consumers see all records including aborted ones)

# What read_committed means:
# - Records written in a committed transaction: VISIBLE
# - Records written in an aborted transaction: HIDDEN (skipped)
# - Records written without a transaction: VISIBLE immediately
# - Records in an in-flight (uncommitted) transaction: HELD — consumer waits

# The consumer's last stable offset (LSO) limits what read_committed can see:
# LSO = offset of the earliest open (uncommitted) transaction
# If a transaction is never committed/aborted, LSO stops advancing → consumer lag!
# Monitor: kafka_log_LogStartOffset vs consumer group offset

@Configuration
class KafkaTransactionConfig {
    @Bean
    ProducerFactory<String, String> producerFactory() {
        var props = Map.of(
            ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092",
            ProducerConfig.TRANSACTIONAL_ID_CONFIG,  "spring-tx-",  // Spring appends unique suffix
            ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true"
        );
        return new DefaultKafkaProducerFactory<>(props);
    }

    @Bean
    KafkaTransactionManager<String, String> kafkaTransactionManager(
            ProducerFactory<String, String> pf) {
        return new KafkaTransactionManager<>(pf);
    }
}

@Service
class OrderProcessor {

    @Transactional(transactionManager = "kafkaTransactionManager")
    public void process(ConsumerRecord<String, Order> record) {
        Order processed = enrich(record.value());
        // Both sends are in the same Kafka transaction
        kafkaTemplate.send("orders-enriched", record.key(), processed);
        kafkaTemplate.send("audit-log", record.key(), toAuditEvent(processed));
        // Spring commits offsets + transaction atomically on method return
    }
}