Generate Kafka Producer Code

Expert in Apache Kafka producer development, specializing in robust, performant, production-ready producer implementations across Java, Python, and Node.js. Covers Kafka's delivery semantics - at-most-once with acks=0, at-least-once with acks=1, exactly-once with acks=all plus enable.idempotence=true - key configuration parameters (bootstrap.servers, serializers, batch.size, linger.ms, buffer.memory, retry behavior), custom partitioning, error handling, Schema Registry integration, and monitoring.

Use this when implementing or hardening a Kafka producer in Java, Python, or Node.js: choosing delivery semantics, custom partitioning logic, batching or compression tuning, error handling, Avro/Schema Registry serialization, or health-check and monitoring instrumentation.

Not needed for consumer-side Kafka code, or for a language or client library not covered here.

Produces runnable producer implementations: a Java KafkaProducer wrapper with performance (batch.size, linger.ms, buffer.memory) and reliability (acks=all, unlimited retries, enable.idempotence) settings and an async send callback logging partition and offset; a custom Java Partitioner routing key-prefixed "priority-" messages to partition 0 and murmur2-hashing the rest; a Python KafkaMessageProducer class wrapping kafka-python with JSON serialization, gzip compression, batch sending, and a flush()-backed send_batch method; and a Node.js KafkaProducerClient built on kafkajs with idempotent production, retry-with-backoff broker connection settings, transactional sends via producer.transaction() with commit/abort, and connect/disconnect lifecycle methods.

public void sendMessage(String topic, String key, String value) {
    ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);
    
    producer.send(record, (metadata, exception) -> {
        if (exception != null) {
            System.err.println("Error sending message: " + exception.getMessage());
        } else {
            System.out.println("Message sent to partition " + metadata.partition() + 
                             " with offset " + metadata.offset());
        }
    });
}

Performance guidance: set batch.size to 16KB-64KB, linger.ms to 1-10ms, use a compression.type of gzip, snappy, or lz4 for large messages, and tune buffer.memory to expected volume with connection pooling. Error handling distinguishes retriable exceptions, which are auto-retried, from non-retriable ones, which are routed to a dead-letter queue or alternative handling. Schema Registry integration configures KafkaAvroSerializer with a schema.registry.url, disabling auto-schema-registration and pinning to the latest schema version. Monitoring tracks record-send-rate, batch-size-avg, record-error-rate, and buffer-available-bytes, plus a simple health check that calls producer.partitionsFor(...) to confirm broker connectivity.

Production deployments should also implement proper connection pooling, graceful shutdown procedures, and comprehensive error handling throughout.

Backend engineers building or hardening Kafka producers in Java, Python, or Node.js who need concrete, production-ready configuration and error-handling patterns across delivery semantics and client libraries, rather than generic Kafka conceptual documentation.