Implementing Custom Interceptors

// ProducerInterceptor<K,V> and ConsumerInterceptor<K,V>

public class AuditInterceptor implements ProducerInterceptor<String,String> {
  public ProducerRecord<String,String> onSend(ProducerRecord<String,String> r){
    return r; }
  public void onAcknowledgement(RecordMetadata m, Exception e){}
  public void close(){}
  public void configure(Map<String,?> c){}
}

public ConsumerRecords<String,String> onConsume(
    ConsumerRecords<String,String> records){
  return records; // inspect/transform
}

public ProducerRecord<String,String> onSend(ProducerRecord<String,String> r){
  r.headers().add("trace-id", traceId().getBytes());
  return r;
}

public ConsumerRecords<String,String> onConsume(
    ConsumerRecords<String,String> recs){
  meter.mark(recs.count());
  return recs;
}

props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,
  "com.acme.AuditInterceptor,com.acme.MetricsInterceptor");

public void onAcknowledgement(RecordMetadata m, Exception e){
  if (e != null) errors.inc(); else sent.inc();
}

r.headers().add("traceparent",
  span.context().toTraceparent().getBytes());

String redacted = r.value().replaceAll("\\d{16}", "****");
return new ProducerRecord<>(r.topic(), r.key(), redacted);

String enriched = addEnvelope(r.value(), System.currentTimeMillis());
return new ProducerRecord<>(r.topic(), r.partition(), r.key(), enriched);