Implementing Idempotent Kafka Consumers for True Exactly-Once Semantics

CREATE TABLE orders (
    order_id UUID PRIMARY KEY,
    customer_id VARCHAR(255) NOT NULL,
    order_details JSONB,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

-- The idempotency key table
CREATE TABLE processed_kafka_messages (
    consumer_group VARCHAR(255) NOT NULL,
    topic VARCHAR(255) NOT NULL,
    partition INT NOT NULL,
    message_offset BIGINT NOT NULL,
    processed_at TIMESTAMPTZ DEFAULT NOW(),
    PRIMARY KEY (consumer_group, topic, partition, message_offset)
);

// Order.java
@Entity
@Table(name = "orders")
public class Order {
    @Id
    private UUID orderId;
    private String customerId;
    // ... other fields, getters, setters
}

// ProcessedKafkaMessage.java
@Entity
@Table(name = "processed_kafka_messages")
@IdClass(ProcessedKafkaMessageId.class)
public class ProcessedKafkaMessage {
    @Id
    private String consumerGroup;
    @Id
    private String topic;
    @Id
    private int partition;
    @Id
    private long messageOffset;
    // ... getters, setters
}

// ProcessedKafkaMessageId.java (Composite Key Class)
public class ProcessedKafkaMessageId implements Serializable {
    private String consumerGroup;
    private String topic;
    private int partition;
    private long messageOffset;
    // ... equals, hashCode, getters, setters
}

// application.yml - CRITICAL CONFIGURATION
spring:
  kafka:
    consumer:
      group-id: order-processing-group
      auto-offset-reset: earliest
      enable-auto-commit: false # We will manage commits manually
    listener:
      ack-mode: MANUAL_IMMEDIATE # Gives us control over acknowledgment

// OrderConsumerService.java
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

@Service
public class OrderConsumerService {

    private final OrderRepository orderRepository;
    private final ProcessedMessageRepository processedMessageRepository;
    private final String consumerGroupId = "order-processing-group";

    // Constructor injection

    @KafkaListener(topics = "order.events", groupId = consumerGroupId)
    @Transactional("transactionManager") // Ensure this uses the DB transaction manager
    public void consumeOrderEvent(String payload, 
                                  @Header(KafkaHeaders.RECEIVED_TOPIC) String topic, 
                                  @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition, 
                                  @Header(KafkaHeaders.OFFSET) long offset, 
                                  Acknowledgment acknowledgment) {

        // 1. Idempotency Check
        if (processedMessageRepository.existsByConsumerGroupAndTopicAndPartitionAndMessageOffset(
            consumerGroupId, topic, partition, offset)) {
            
            // Message already processed, skip business logic
            System.out.println("Skipping duplicate message: " + topic + "-" + partition + "@" + offset);
            acknowledgment.acknowledge(); // Acknowledge to commit the offset
            return;
        }

        // 2. Process Business Logic
        try {
            // Assuming payload is a JSON representation of an Order
            Order order = deserializeOrder(payload);
            orderRepository.save(order);

            // 3. Record the message as processed
            ProcessedKafkaMessage processedMessage = new ProcessedKafkaMessage(
                consumerGroupId, topic, partition, offset);
            processedMessageRepository.save(processedMessage);

            // 4. Acknowledge the message (which commits the offset)
            // This happens after the transaction commits successfully thanks to Spring's magic
            acknowledgment.acknowledge();

        } catch (Exception e) {
            // On any exception, the transaction will be rolled back.
            // The message will not be acknowledged, so Kafka will redeliver it.
            // Implement proper error handling (e.g., dead-letter queue)
            System.err.println("Error processing message: " + e.getMessage());
            // Do NOT acknowledge here.
        }
    }

    private Order deserializeOrder(String payload) {
        // Implementation for deserializing JSON to Order object
        return new Order(); // Placeholder
    }
}

CREATE TABLE outbox_messages (
    id UUID PRIMARY KEY,
    aggregate_type VARCHAR(255) NOT NULL,
    aggregate_id VARCHAR(255) NOT NULL,
    destination_topic VARCHAR(255) NOT NULL,
    payload JSONB NOT NULL,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

// Simplified example inside a @Transactional method
@Transactional
public void processPayment(PaymentEvent payment) {
    // 1. Update business state
    Invoice invoice = invoiceRepository.findById(payment.getInvoiceId());
    invoice.setStatus("PAID");
    invoiceRepository.save(invoice);

    // 2. Create the outbox message
    String payload = createInvoicePaidEventPayload(invoice);
    OutboxMessage outboxMessage = new OutboxMessage(
        UUID.randomUUID(),
        "Invoice",
        invoice.getId().toString(),
        "invoice.events",
        payload
    );
    outboxRepository.save(outboxMessage);

    // 3. The transaction commits here, atomically saving both.
}

{
  "name": "outbox-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "postgres",
    "database.password": "password",
    "database.dbname": "mydatabase",
    "database.server.name": "myserver",
    "table.include.list": "public.outbox_messages",
    "transforms": "outbox",
    "transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter",
    "transforms.outbox.route.by.field": "destination_topic",
    "transforms.outbox.table.field.event.key": "aggregate_id",
    "transforms.outbox.table.field.event.payload": "payload",
    "tombstones.on.delete": "false" 
  }
}

// Producer Properties
Properties producerProps = new Properties();
producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
producerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
producerProps.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "my-transform-process-1"); // Must be unique and stable

// Consumer Properties
Properties consumerProps = new Properties();
consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "my-transform-group");
consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
consumerProps.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed"); // Only read committed messages

KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps);
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);

consumer.subscribe(Collections.singletonList("input.topic"));
producer.initTransactions(); // Initializes the producer for transactions

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(Long.MAX_VALUE));
    
    try {
        producer.beginTransaction();

        for (ConsumerRecord<String, String> record : records) {
            // 1. Process the record
            String processedValue = record.value().toUpperCase();

            // 2. Produce the transformed record to the output topic
            producer.send(new ProducerRecord<>("output.topic", record.key(), processedValue));
        }

        // 3. Send consumed offsets to be part of the transaction
        Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = new HashMap<>();
        for (TopicPartition partition : records.partitions()) {
            List<ConsumerRecord<String, String>> partitionedRecords = records.records(partition);
            long lastOffset = partitionedRecords.get(partitionedRecords.size() - 1).offset();
            offsetsToCommit.put(partition, new OffsetAndMetadata(lastOffset + 1));
        }
        producer.sendOffsetsToTransaction(offsetsToCommit, consumer.groupMetadata());

        // 4. Commit the entire transaction
        producer.commitTransaction();

    } catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
        // Unrecoverable errors, close the producer
        producer.close();
        break;
    } catch (KafkaException e) {
        // Abortable error, roll back the transaction
        producer.abortTransaction();
        // The consumer will re-seek to the last committed offset on the next poll
    }
}
producer.close();