Spring Kafka Integration

@Service
public class OrderPublisher {
  private final KafkaTemplate<String, OrderEvent> kafkaTemplate;

  public OrderPublisher(KafkaTemplate<String, OrderEvent> kafkaTemplate) {
    this.kafkaTemplate = kafkaTemplate;
  }

  public CompletableFuture<SendResult<String, OrderEvent>> publish(OrderEvent event) {
    // Spring Kafka 3.x returns CompletableFuture (was ListenableFuture in 2.x)
    return kafkaTemplate.send("orders", event.getOrderId(), event)
        .whenComplete((result, ex) -> {
          if (ex != null) log.error("send failed", ex);
          else log.debug("offset={}", result.getRecordMetadata().offset());
        });
  }

  public void publishWithHeaders(OrderEvent event) {
    ProducerRecord<String, OrderEvent> record = new ProducerRecord<>(
        "orders", null, event.getOrderId(), event,
        List.of(new RecordHeader("eventType", event.getType().getBytes()))
    );
    kafkaTemplate.send(record);
  }
}

// 1. Value only — simplest
@KafkaListener(topics = "orders", groupId = "order-processor")
public void onOrder(OrderEvent order) {
  process(order);
}

// 2. Full ConsumerRecord — metadata access
@KafkaListener(topics = "orders")
public void onRecord(ConsumerRecord<String, OrderEvent> record) {
  log.info("partition={} offset={}", record.partition(), record.offset());
  process(record.value());
}

// 3. Headers via @Header
@KafkaListener(topics = "orders")
public void onWithHeaders(
    @Payload OrderEvent order,
    @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
    @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
    @Header(name = "eventType", required = false) String eventType) {
  process(order);
}

// 4. Manual commit — requires MANUAL ackMode on factory
@KafkaListener(topics = "orders", containerFactory = "manualAckFactory")
public void onManual(OrderEvent order, Acknowledgment ack) {
  process(order);
  ack.acknowledge();  // commit offset after successful processing
}

// 5. Batch listener
@KafkaListener(topics = "orders", containerFactory = "batchFactory")
public void onBatch(List<ConsumerRecord<String, OrderEvent>> records) {
  records.forEach(r -> process(r.value()));
}

@KafkaListener(topics = "inventory-requests")
@SendTo  // reply topic from KafkaHeaders.REPLY_TOPIC on incoming record
public InventoryResponse checkStock(@Payload InventoryRequest request) {
  return inventoryService.check(request);
}

@Bean
public ConcurrentKafkaListenerContainerFactory<String, OrderEvent> kafkaListenerContainerFactory(
    ConsumerFactory<String, OrderEvent> consumerFactory) {
  ConcurrentKafkaListenerContainerFactory<String, OrderEvent> factory =
      new ConcurrentKafkaListenerContainerFactory<>();
  factory.setConsumerFactory(consumerFactory);
  factory.setConcurrency(3);
  factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.RECORD);
  factory.getContainerProperties().setPollTimeout(Duration.ofSeconds(3).toMillis());
  return factory;
}

spring:
  kafka:
    listener:
      ack-mode: record
      concurrency: 3
      type: single   # or batch

@Bean
public DefaultErrorHandler errorHandler(KafkaTemplate<String, OrderEvent> template) {
  DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(
      template,
      (record, ex) -> new TopicPartition(record.topic() + ".DLT", record.partition())
  );

  // 3 retries, 2s apart
  DefaultErrorHandler handler = new DefaultErrorHandler(recoverer, new FixedBackOff(2000L, 3));

  // Skip retries for poison schema / validation errors
  handler.addNotRetryableExceptions(
      DeserializationException.class,
      ValidationException.class
  );
  return handler;
}

@Bean
public ConcurrentKafkaListenerContainerFactory<String, OrderEvent> factory(
    ConsumerFactory<String, OrderEvent> cf,
    DefaultErrorHandler errorHandler) {
  ConcurrentKafkaListenerContainerFactory<String, OrderEvent> f =
      new ConcurrentKafkaListenerContainerFactory<>();
  f.setConsumerFactory(cf);
  f.setCommonErrorHandler(errorHandler);
  return f;
}

ExponentialBackOffWithMaxRetries backOff = new ExponentialBackOffWithMaxRetries(5);
backOff.setInitialInterval(500L);
backOff.setMultiplier(2.0);
backOff.setMaxInterval(30_000L);

DefaultErrorHandler handler = new DefaultErrorHandler(recoverer, backOff);

@RetryableTopic(
    attempts = "4",
    backoff = @Backoff(delay = 1000, multiplier = 2.0, maxDelay = 30_000),
    dltStrategy = DltStrategy.FAIL_ON_ERROR,
    include = { TransientDataAccessException.class },
    exclude = { ValidationException.class }
)
@KafkaListener(topics = "orders", groupId = "order-processor")
public void process(@Payload OrderEvent order) {
  orderService.process(order);
}

@DltHandler
public void handleDlt(OrderEvent order, @Header(KafkaHeaders.EXCEPTION_MESSAGE) String error) {
  log.error("DLT orderId={} error={}", order.getOrderId(), error);
}

@Bean
public ProducerFactory<String, OrderEvent> transactionalProducerFactory() {
  Map<String, Object> props = new HashMap<>(producerProps());
  props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "order-service-txn");
  DefaultKafkaProducerFactory<String, OrderEvent> factory =
      new DefaultKafkaProducerFactory<>(props);
  factory.setTransactionIdPrefix("order-service-txn-");
  return factory;
}

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

@Transactional("kafkaTransactionManager")
@KafkaListener(topics = "orders", groupId = "enrichment-service")
public void enrichAndPublish(ConsumerRecord<String, OrderEvent> record) {
  OrderEvent enriched = enrichmentService.enrich(record.value());
  kafkaTemplate.send("orders-enriched", record.key(), enriched);
  // offset commit for this record included in same transaction
}

@Bean
public ChainedKafkaTransactionManager<String, OrderEvent> chainedTxManager(
    KafkaTransactionManager<String, OrderEvent> kafkaTm,
    DataSourceTransactionManager dbTm) {
  return new ChainedKafkaTransactionManager<>(kafkaTm, dbTm);
}

@Transactional("chainedKafkaTransactionManager")
public void updateDbAndPublish(OrderEvent event) {
  orderRepository.save(event.toEntity());
  kafkaTemplate.send("orders-processed", event.getOrderId(), event);
}

spring:
  kafka:
    consumer:
      properties:
        isolation.level: read_committed
    producer:
      transaction-id-prefix: order-service-txn-

@SpringBootTest
@EmbeddedKafka(
    partitions = 3,
    topics = { "orders", "orders.DLT" },
    brokerProperties = { "listeners=PLAINTEXT://localhost:0", "port=0" }
)
class OrderListenerIT {

  @Autowired
  private KafkaTemplate<String, OrderEvent> kafkaTemplate;

  @Autowired
  private EmbeddedKafkaBroker embeddedKafka;

  @Test
  void shouldProcessOrder() {
    OrderEvent event = new OrderEvent("ord-1", "PLACED");
    kafkaTemplate.send("orders", "ord-1", event);

    Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(
        "test-group", "true", embeddedKafka);
    Consumer<String, OrderEvent> consumer = new DefaultKafkaConsumer<>(
        consumerProps, new StringDeserializer(), new JsonDeserializer<>(OrderEvent.class));
    embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "orders-processed");

    ConsumerRecords<String, OrderEvent> records =
        KafkaTestUtils.getRecords(consumer, Duration.ofSeconds(10));
    assertThat(records.count()).isEqualTo(1);
    assertThat(records.iterator().next().value().getOrderId()).isEqualTo("ord-1");
  }
}

@Testcontainers
@SpringBootTest
class OrderServiceContainerIT {

  @Container
  static KafkaContainer kafka = new KafkaContainer(
      DockerImageName.parse("confluentinc/cp-kafka:7.6.0"));

  @DynamicPropertySource
  static void kafkaProps(DynamicPropertyRegistry registry) {
    registry.add("spring.kafka.bootstrap-servers", kafka::getBootstrapServers);
  }

  @Autowired
  private KafkaTemplate<String, OrderEvent> kafkaTemplate;

  @Test
  void publishesToRealBroker() {
    kafkaTemplate.send("orders", "k1", new OrderEvent("ord-99", "PLACED"));
    // assert via @KafkaListener test spy or consuming test consumer
  }
}

@Test
void kafkaTemplateUsesMockProducer() {
  MockProducer<String, String> mockProducer = new MockProducer<>(true, null, null);
  DefaultKafkaProducerFactory<String, String> pf = new DefaultKafkaProducerFactory<>(
      Map.of(), () -> mockProducer, null);
  KafkaTemplate<String, String> template = new KafkaTemplate<>(pf);

  template.send("orders", "key", "value");
  assertThat(mockProducer.history()).hasSize(1);
  assertThat(mockProducer.history().get(0).topic()).isEqualTo("orders");
}

spring:
  kafka:
    bootstrap-servers: kafka:9092
    properties:
      schema.registry.url: http://schema-registry:8081
      specific.avro.reader: true
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
      properties:
        auto.register.schemas: false
        use.latest.version: true
    consumer:
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer

@Bean
public SchemaRegistryClient schemaRegistryClient(
    @Value("${spring.kafka.properties.schema.registry.url}") String url) {
  Map<String, Object> configs = Map.of(
      AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, url
  );
  return new CachedSchemaRegistryClient(
      List.of(url.split(",")),
      1000,
      List.of(new AvroSchemaProvider(), new ProtobufSchemaProvider()),
      configs
  );
}

// Generated from Order.avsc via avro-maven-plugin
@KafkaListener(topics = "orders", groupId = "order-processor")
public void onOrder(Order order) {
  log.info("orderId={} status={}", order.getOrderId(), order.getStatus());
}

@Bean
public ConsumerFactory<String, Order> avroConsumerFactory(
    @Value("${spring.kafka.bootstrap-servers}") String bootstrap) {
  Map<String, Object> props = new HashMap<>();
  props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap);
  props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-processor");
  props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
  props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class);
  props.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, true);
  props.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://schema-registry:8081");
  return new DefaultKafkaConsumerFactory<>(props);
}