What is a ProducerInterceptor, and what do onSend and onAcknowledgement do, including thread and ordering semantics?

## Purpose A `ProducerInterceptor<K,V>` is a pluggable hook in the producer's send pipeline, used for **cross-cutting concerns** — tracing, metrics, auditing, header injection — without touching business code. Configure a chain: ``` interceptor.classes=com.acme.TracingInterceptor,com.acme.AuditInterceptor ``` ## The interface ``` public interface ProducerInterceptor<K,V> extends Configurable, AutoCloseable { ProducerRecord<K,V> onSend(ProducerRecord<K,V> record); void onAcknowledgement(RecordMetadata metadata, Exception exception); void close(); } ``` ### onSend - Called **on the application/caller thread** from inside `KafkaProducer.send()`, **before serialization and before the partitioner runs**. - Receives the record and **returns a (possibly new) record**. Returning a modified record lets you add headers or rewrite fields. The returned record from one interceptor is the input to the next. - It sees the record **pre-serialization** (objects, not bytes), so it can read the typed key/value. Changing the key or target partition **alters routing** — generally you only add headers. - If `onSend` throws, the exception is **caught and logged**; the original (pre-interceptor) record continues — interceptors must not break sends. ### onAcknowledgement - Called when the record is **acknowledged by the broker** or when the send **fails** (e.g. serialization error, buffer exhaustion, broker error). - Exactly one of `(metadata, exception)` is non-null: success → metadata with topic/partition/offset; failure → exception, and metadata has partition info but offset -1. - Runs on the producer **I/O (Sender) thread** — the same thread that batches and transmits. **It must be fast and non-blocking**: any blocking call (network, lock, heavy compute) here stalls *all* in-flight sends for the producer. It also runs *before* the user's `Callback`/`Future` completes, so heavy work here delays delivery callbacks. ## Chain ordering - `interceptor.classes` defines order. `onSend` executes **first→last**, threading each output into the next. - `onAcknowledgement` is invoked on every interceptor as well; an exception in one is caught/logged and does not stop the others. ## Lifecycle - `configure(Map)` (from `Configurable`) runs at producer startup; `close()` at producer close. - Interceptors are **shared across all sends** of the producer instance, so they must be thread-safe if they hold state (onSend on caller threads, onAcknowledgement on the Sender thread can run concurrently). ## Common real uses - **Distributed tracing**: OpenTelemetry's `TracingProducerInterceptor` injects W3C traceparent headers in onSend. - **Confluent Monitoring Interceptor**: records produce metrics for Control Center. - **Auditing/metrics**: count records, measure end-to-end ack latency in onAcknowledgement. ## Edge cases / gotchas - Blocking in onAcknowledgement is the classic production incident — it serializes onto the single Sender thread and tanks throughput. - onSend running before serialization means an interceptor can't see or fix the serialized bytes — for that you'd write a serializer, not an interceptor. - Because onSend exceptions are swallowed, a buggy interceptor can silently no-op; test it explicitly. - Interceptors do not run for records dropped before send (they aren't a validation layer you can rely on to reject).