TracingProducerInterceptor extends BaseTracingInterceptor implements ProducerInterceptor<K, V> { @Override public ProducerRecord<K, V> onSend(ProducerRecord<K, V> record) { TraceContextOrSamplingFlags traceContextOrSamplingFlags = extractor.extract(record.headers()); Span span = tracing.tracer().nextSpan(traceContextOrSamplingFlags); tracing.propagation().keys().forEach(key -> record.headers().remove(key)); injector.inject(span.context(), record.headers()); if (!span.isNoop()) { if (record.key() instanceof String && !"".equals(record.key())) { span.tag(KafkaTagKey.KAFKA_KEY, record.key().toString()); } if (remoteServiceName != null) { span.remoteServiceName(remoteServiceName); } span.tag(KafkaTagKey.KAFKA_TOPIC, record.topic()) .tag(KafkaTagKey.KAFKA_CLIENT_ID, clientId) .name(SEND_OPERATION) .kind(Span.Kind.PRODUCER) .start(); } span.finish(); LOGGER.debug("Producer Record intercepted: {}", span.context()); return record; }  @Override ProducerRecord<K, V> onSend(ProducerRecord<K, V> record); @Override void onAcknowledgement(RecordMetadata recordMetadata, Exception exception); @Override void close();  }
@Test public void shouldNotTouchRecords() { final TracingProducerInterceptor<String, String> interceptor = new TracingProducerInterceptor<>(); interceptor.configure(map); final ProducerRecord<String, String> tracedRecord = interceptor.onSend(record); Assert.assertEquals(record, tracedRecord); }
@Test public void shouldCreateSpanOnSend() { final TracingProducerInterceptor<String, String> interceptor = new TracingProducerInterceptor<>(); interceptor.configure(map); interceptor.tracing = tracing; interceptor.onSend(record); final Span span = spans.getLast(); assertNotNull(span); }