TracingConsumer implements Consumer<K, V> { public ConsumerRecords<K, V> poll(Duration timeout) { return poll(timeout.toMillis()); } TracingConsumer(Consumer<K, V> delegate, KafkaTracing kafkaTracing); ConsumerRecords<K, V> poll(Duration timeout); ConsumerRecords<K, V> poll(long timeout); @Override Set<TopicPartition> assignment(); @Override Set<String> subscription(); @Override void subscribe(Collection<String> topics); @Override void subscribe(Collection<String> topics, ConsumerRebalanceListener callback); @Override void assign(Collection<TopicPartition> partitions); @Override void subscribe(Pattern pattern, ConsumerRebalanceListener callback); void subscribe(Pattern pattern); @Override void unsubscribe(); @Override void commitSync(); void commitSync(Duration timeout); @Override void commitSync(Map<TopicPartition, OffsetAndMetadata> offsets); void commitSync(Map<TopicPartition, OffsetAndMetadata> offsets, Duration timeout); @Override void commitAsync(); @Override void commitAsync(OffsetCommitCallback callback); @Override void commitAsync(Map<TopicPartition, OffsetAndMetadata> offsets,
    OffsetCommitCallback callback); @Override void seek(TopicPartition partition, long offset); void seek(TopicPartition topicPartition, OffsetAndMetadata offsetAndMetadata); @Override void seekToBeginning(Collection<TopicPartition> partitions); @Override void seekToEnd(Collection<TopicPartition> partitions); @Override long position(TopicPartition partition); long position(TopicPartition partition, Duration timeout); @Override OffsetAndMetadata committed(TopicPartition partition); OffsetAndMetadata committed(TopicPartition partition, Duration timeout); Map<TopicPartition, OffsetAndMetadata> committed(Set<TopicPartition> partitions); Map<TopicPartition, OffsetAndMetadata> committed(
    Set<TopicPartition> partitions, Duration timeout); @Override Map<MetricName, ? extends Metric> metrics(); @Override List<PartitionInfo> partitionsFor(String topic); List<PartitionInfo> partitionsFor(String topic, Duration timeout); @Override Map<String, List<PartitionInfo>> listTopics(); @Override Map<String, List<PartitionInfo>> listTopics(Duration timeout); @Override Set<TopicPartition> paused(); @Override void pause(Collection<TopicPartition> partitions); @Override void resume(Collection<TopicPartition> partitions); @Override Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes(
    Map<TopicPartition, Long> timestampsToSearch); Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes(
    Map<TopicPartition, Long> timestampsToSearch, Duration timeout); @Override Map<TopicPartition, Long> beginningOffsets(Collection<TopicPartition> partitions); Map<TopicPartition, Long> beginningOffsets(Collection<TopicPartition> partitions,
    Duration timeout); @Override Map<TopicPartition, Long> endOffsets(Collection<TopicPartition> partitions); Map<TopicPartition, Long> endOffsets(Collection<TopicPartition> partitions,
    Duration timeout); ConsumerGroupMetadata groupMetadata(); void enforceRebalance(); @Override void close(); void close(long timeout, TimeUnit unit); void close(Duration timeout); @Override void wakeup();  }
@Test public void should_add_new_trace_headers_if_b3_missing() { consumer.addRecord(consumerRecord); Consumer<String, String> tracingConsumer = kafkaTracing.consumer(consumer); ConsumerRecords<String, String> poll = tracingConsumer.poll(10); assertThat(poll) .extracting(ConsumerRecord::headers) .flatExtracting(KafkaTest::lastHeaders) .extracting(Map.Entry::getKey) .containsOnly("b3"); MutableSpan consumerSpan = spans.get(0); assertThat(consumerSpan.kind()).isEqualTo(CONSUMER); assertThat(consumerSpan.parentId()).isNull(); }
@Test public void should_createChildOfTraceHeaders() { addB3MultiHeaders(parent, consumerRecord); consumer.addRecord(consumerRecord); Consumer<String, String> tracingConsumer = kafkaTracing.consumer(consumer); ConsumerRecords<String, String> poll = tracingConsumer.poll(10); assertThat(poll) .extracting(ConsumerRecord::headers) .flatExtracting(TracingConsumerTest::lastHeaders) .hasSize(1) .allSatisfy(e -> { assertThat(e.getKey()).isEqualTo("b3"); assertThat(e.getValue()).startsWith(parent.traceIdString()); }); assertChildOf(spans.get(0), parent); }