TracingProducer implements Producer<K, V> { @Override public Future<RecordMetadata> send(ProducerRecord<K, V> record) { return this.send(record, null); } TracingProducer(Producer<K, V> delegate, KafkaTracing kafkaTracing); @Override void initTransactions(); @Override void beginTransaction(); @Override void commitTransaction(); @Override void abortTransaction(); @Override Future<RecordMetadata> send(ProducerRecord<K, V> record); @Override Future<RecordMetadata> send(ProducerRecord<K, V> record, @Nullable Callback callback); @Override void flush(); @Override List<PartitionInfo> partitionsFor(String topic); @Override Map<MetricName, ? extends Metric> metrics(); @Override void close(); void close(long timeout, TimeUnit unit); void close(Duration duration); @Override void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
    String consumerGroupId); void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
    ConsumerGroupMetadata groupMetadata);  }
@Test public void should_add_b3_headers_to_records() { tracingProducer.send(producerRecord); List<String> headerKeys = mockProducer.history().stream() .flatMap(records -> Arrays.stream(records.headers().toArray())) .map(Header::key) .collect(Collectors.toList()); assertThat(headerKeys).containsOnly("b3"); }
@Test public void should_add_b3_headers_when_other_headers_exist() { ProducerRecord<Object, String> record = new ProducerRecord<>(TEST_TOPIC, TEST_KEY, TEST_VALUE); record.headers().add("tx-id", "1".getBytes()); tracingProducer.send(record); mockProducer.completeNext(); MutableSpan producerSpan = spans.get(0); assertThat(producerSpan.kind()).isEqualTo(PRODUCER); assertThat(lastHeaders(mockProducer)) .containsEntry("tx-id", "1") .containsEntry("b3", producerSpan.traceId() + "-" + producerSpan.id() + "-1"); }
@Test public void should_inject_child_context() { try (Scope scope = currentTraceContext.newScope(parent)) { tracingProducer.send(new ProducerRecord<>(TEST_TOPIC, TEST_KEY, TEST_VALUE)); mockProducer.completeNext(); } MutableSpan producerSpan = spans.get(0); assertThat(producerSpan.kind()).isEqualTo(PRODUCER); assertChildOf(producerSpan, parent); assertThat(lastHeaders(mockProducer)) .containsEntry("b3", producerSpan.traceId() + "-" + producerSpan.id() + "-1"); }
@Test public void should_add_parent_trace_when_context_injected_on_headers() { ProducerRecord<Object, String> record = new ProducerRecord<>(TEST_TOPIC, TEST_KEY, TEST_VALUE); tracingProducer.injector.inject(parent, new KafkaProducerRequest(record)); tracingProducer.send(record); mockProducer.completeNext(); MutableSpan producerSpan = spans.get(0); assertThat(producerSpan.kind()).isEqualTo(PRODUCER); assertChildOf(producerSpan, parent); assertThat(lastHeaders(mockProducer)) .containsEntry("b3", producerSpan.traceId() + "-" + producerSpan.id() + "-1"); }
@Test public void should_call_wrapped_producer() { tracingProducer.send(new ProducerRecord<>(TEST_TOPIC, TEST_KEY, TEST_VALUE)); assertThat(mockProducer.history()).hasSize(1); }
@Test public void send_should_set_name() { tracingProducer.send(new ProducerRecord<>(TEST_TOPIC, TEST_KEY, TEST_VALUE)); mockProducer.completeNext(); MutableSpan producerSpan = spans.get(0); assertThat(producerSpan.kind()).isEqualTo(PRODUCER); assertThat(producerSpan.name()).isEqualTo("send"); }
@Test public void send_should_tag_topic_and_key() { tracingProducer.send(new ProducerRecord<>(TEST_TOPIC, TEST_KEY, TEST_VALUE)); mockProducer.completeNext(); MutableSpan producerSpan = spans.get(0); assertThat(producerSpan.kind()).isEqualTo(PRODUCER); assertThat(producerSpan.tags()) .containsOnly(entry("kafka.topic", TEST_TOPIC), entry("kafka.key", TEST_KEY)); }
@Test public void send_shouldnt_tag_null_key() { tracingProducer.send(new ProducerRecord<>(TEST_TOPIC, null, TEST_VALUE)); mockProducer.completeNext(); MutableSpan producerSpan = spans.get(0); assertThat(producerSpan.kind()).isEqualTo(PRODUCER); assertThat(producerSpan.tags()) .containsOnly(entry("kafka.topic", TEST_TOPIC)); }
@Test public void send_shouldnt_tag_binary_key() { tracingProducer.send(new ProducerRecord<>(TEST_TOPIC, new byte[1], TEST_VALUE)); mockProducer.completeNext(); MutableSpan producerSpan = spans.get(0); assertThat(producerSpan.kind()).isEqualTo(PRODUCER); assertThat(producerSpan.tags()) .containsOnly(entry("kafka.topic", TEST_TOPIC)); }