content: /*
   * Copyright The OpenTelemetry Authors
   * SPDX-License-Identifier: Apache-2.0
   */
  
  package io.opentelemetry.javaagent.instrumentation.kafkaclients.v0_11;
  
  import static org.assertj.core.api.Assertions.assertThat;
  
  import io.opentelemetry.api.trace.SpanKind;
  import io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal.KafkaClientPropagationBaseTest;
  import io.opentelemetry.instrumentation.testing.junit.AgentInstrumentationExtension;
  import io.opentelemetry.instrumentation.testing.junit.InstrumentationExtension;
  import java.time.Duration;
  import java.util.Collections;
  import org.apache.kafka.clients.consumer.ConsumerRecord;
  import org.apache.kafka.clients.consumer.ConsumerRecords;
  import org.apache.kafka.clients.producer.ProducerRecord;
  import org.junit.jupiter.api.DisplayName;
  import org.junit.jupiter.api.Test;
  import org.junit.jupiter.api.extension.RegisterExtension;
  
  class KafkaClientPropagationDisabledTest extends KafkaClientPropagationBaseTest {
    @RegisterExtension
    static final InstrumentationExtension testing = AgentInstrumentationExtension.create();
  
    @DisplayName("should not read remote context when consuming messages if propagation is disabled")
    @Test
    void testReadRemoteContextWhenPropagationIsDisabled() throws InterruptedException {
      String message = "Testing without headers";
      producer.send(new ProducerRecord<>(SHARED_TOPIC, message));
  
      testing.waitAndAssertTraces(
          trace ->
              trace.hasSpansSatisfyingExactly(
                  span ->
                      span.hasName(SHARED_TOPIC + " publish")
                          .hasKind(SpanKind.PRODUCER)
                          .hasNoParent()
                          .hasAttributesSatisfyingExactly(sendAttributes(null, message, false))));
  
      awaitUntilConsumerIsReady();
  
      ConsumerRecords<?, ?> records = poll(Duration.ofSeconds(5));
      assertThat(records.count()).isEqualTo(1);
  
      // iterate over records to generate spans
      for (ConsumerRecord<?, ?> ignored : records) {
        testing.runWithSpan("processing", () -> {});
      }
  
      testing.waitAndAssertTraces(
          trace ->
              trace.hasSpansSatisfyingExactly(
                  span ->
                      span.hasName(SHARED_TOPIC + " publish")
                          .hasKind(SpanKind.PRODUCER)
                          .hasNoParent()
                          .hasAttributesSatisfyingExactly(sendAttributes(null, message, false))),
          trace ->
              trace.hasSpansSatisfyingExactly(
                  span ->
                      span.hasName(SHARED_TOPIC + " process")
                          .hasKind(SpanKind.CONSUMER)
                          .hasLinks(Collections.emptyList())
                          .hasAttributesSatisfyingExactly(
                              processAttributes(null, message, false, false)),
                  span -> span.hasName("processing").hasParent(trace.getSpan(0))));
    }
  }
  