Production Hardening Checklist

spring:
  kafka:
    bootstrap-servers: ${KAFKA_BROKERS}  # from the environment

    producer:
      acks: all                          # default
      retries: 2147483647                # default, bounded by delivery.timeout.ms
      properties:
        enable.idempotence: true         # default
        max.in.flight.requests.per.connection: 5   # default; idempotence needs <= 5
        max.block.ms: 10000              # how long send() may block on metadata or a full buffer (default 60000)
        delivery.timeout.ms: 120000      # default (2 minutes)
        compression.type: lz4            # none, gzip, snappy, lz4 or zstd (default none)
        linger.ms: 5                     # default in Kafka 4.0 and later
        batch.size: 65536                # default is 16384

    consumer:
      max-poll-records: 500              # default
      properties:
        max.poll.interval.ms: 300000     # default (5 minutes): a consumer that polls less often is removed from the group
        session.timeout.ms: 45000        # default
        heartbeat.interval.ms: 15000     # default is 3000; the docs say no higher than 1/3 of session.timeout.ms
        fetch.min.bytes: 1               # default
        fetch.max.wait.ms: 500           # default
        isolation.level: read_committed  # only if you consume transactional topics (default read_uncommitted)

    listener:
      ack-mode: BATCH                    # container default
      missing-topics-fatal: false        # whether the container fails to start when a topic is missing (default false)
      concurrency: ${KAFKA_CONCURRENCY:3}
      idle-event-interval: 60000         # publish an idle event when no records arrive

    admin:
      properties:
        request.timeout.ms: 30000        # default

@Component
public class KafkaHealthIndicator implements HealthIndicator {

    private final KafkaAdmin kafkaAdmin;

    @Override
    public Health health() {
        try {
            // Any successful admin call proves the cluster is reachable
            kafkaAdmin.describeTopics("__consumer_offsets");
            return Health.up()
                .withDetail("cluster", "reachable")
                .build();
        } catch (Exception e) {
            return Health.down(e)
                .withDetail("cluster", "unreachable")
                .build();
        }
    }
}