TracingConsumerInterceptor extends BaseTracingInterceptor implements ConsumerInterceptor<K, V> { public ConsumerRecords<K, V> onConsume(ConsumerRecords<K, V> records) { if (records.isEmpty() || tracing.isNoop()) { return records; } Map<String, Span> consumerSpansForTopic = new LinkedHashMap<>(); for (TopicPartition partition : records.partitions()) { String topic = partition.topic(); List<ConsumerRecord<K, V>> recordsInPartition = records.records(partition); for (ConsumerRecord<K, V> record : recordsInPartition) { TraceContextOrSamplingFlags extracted = extractor.extract(record.headers()); if (extracted.samplingFlags() != null && extracted.extra().isEmpty()) { Span consumerSpanForTopic = consumerSpansForTopic.get(topic); if (consumerSpanForTopic == null) { consumerSpansForTopic.put(topic, consumerSpanForTopic = tracing.tracer().nextSpan(extracted).name("poll") .kind(Span.Kind.CONSUMER) .tag(KafkaTagKey.KAFKA_TOPIC, topic) .tag(KafkaTagKey.KAFKA_GROUP_ID, groupId) .tag(KafkaTagKey.KAFKA_CLIENT_ID, clientId) .start()); } injector.inject(consumerSpanForTopic.context(), record.headers()); } else { Span span = tracing.tracer().nextSpan(extracted); if (!span.isNoop()) { span.name(POLL_OPERATION).kind(Span.Kind.CONSUMER) .tag(KafkaTagKey.KAFKA_TOPIC, topic) .tag(KafkaTagKey.KAFKA_CLIENT_ID, clientId) .tag(KafkaTagKey.KAFKA_GROUP_ID, groupId); if (remoteServiceName != null) { span.remoteServiceName(remoteServiceName); } span.start().finish(); } tracing.propagation().keys().forEach(key -> record.headers().remove(key)); injector.inject(span.context(), record.headers()); } } } consumerSpansForTopic.values().forEach(span -> { if (remoteServiceName != null) { span.remoteServiceName(remoteServiceName); } span.finish(); LOGGER.debug("Consumer Record intercepted: {}", span.context()); }); return records; }  ConsumerRecords<K, V> onConsume(ConsumerRecords<K, V> records); @Override void onCommit(Map<TopicPartition, OffsetAndMetadata> map); @Override void close();  }
@Test public void shouldNotTouchRecords() { final Map<TopicPartition, List<ConsumerRecord<String, String>>> topicPartitionAndRecords = new HashMap<>(); final ConsumerRecord<String, String> record = new ConsumerRecord<>("topic", 0, 0L, "k", "v"); topicPartitionAndRecords.put( new TopicPartition("topic", 0), Collections.singletonList(record)); final ConsumerRecords<String, String> records = new ConsumerRecords<>(topicPartitionAndRecords); final TracingConsumerInterceptor<String, String> interceptor = new TracingConsumerInterceptor<>(); interceptor.configure(map); final ConsumerRecords tracedRecords = interceptor.onConsume(records); assertEquals(records, tracedRecords); }
@Test public void shouldCreateSpansOnConsume() { final Map<TopicPartition, List<ConsumerRecord<String, String>>> topicPartitionAndRecords = new HashMap<>(); final ConsumerRecord<String, String> record = new ConsumerRecord<>("topic", 0, 0L, "k", "v"); topicPartitionAndRecords.put( new TopicPartition("topic", 0), Arrays.asList(record, record, record)); final ConsumerRecords<String, String> records = new ConsumerRecords<>(topicPartitionAndRecords); final TracingConsumerInterceptor<String, String> interceptor = new TracingConsumerInterceptor<>(); interceptor.configure(map); interceptor.tracing = tracing; interceptor.onConsume(records); final Spliterator<Span> span = spans.spliterator(); long count = StreamSupport.stream(span, false).count(); assertEquals(3, count); }