[kafka-exporter] Add Kafka connectivity error handling

kafka-exporter/src/main/java/io/opentelemetry/contrib/kafka/KafkaSpanExporter.java

@@ -59,20 +59,28 @@ public CompletableResultCode export(@Nonnull Collection<SpanData> spans) {
 
  CompletableResultCode result = new CompletableResultCode();
  CompletableFuture.runAsync(
- () ->
- producer.send(
- producerRecord,
- (metadata, exception) -> {
- if (exception == null) {
- result.succeed();
- } else {
- logger.error(
- String.format("Error while sending spans to Kafka topic %s", topicName),
- exception);
- result.fail();
- }
- }),
- executorService);
+ () ->
+ producer.send(
+ producerRecord,
+ (metadata, exception) -> {
+ if (exception == null) {
+ result.succeed();
+ } else {
+ logger.error(
+ String.format("Error while sending spans to Kafka topic %s", topicName),
+ exception);
+ result.fail();
+ }
+ }),
+ executorService)
+ .whenComplete(
+ (ignore, exception) -> {
+ if (exception != null) {
+ logger.error(
+ "Executor task failed while sending to Kafka topic {}", topicName, exception);
+ result.fail();
+ }
+ });
  return result;
  }
 

kafka-exporter/src/test/java/io/opentelemetry/contrib/kafka/KafkaSpanExporterIntegrationTest.java

@@ -19,6 +19,7 @@
 import io.opentelemetry.sdk.common.CompletableResultCode;
 import io.opentelemetry.sdk.trace.data.SpanData;
 import java.time.Duration;
+import java.util.Collection;
 import java.util.List;
 import java.util.UUID;
 import java.util.concurrent.TimeUnit;
@@ -29,7 +30,9 @@
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.MockProducer;
 import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.errors.ApiException;
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.kafka.common.serialization.StringSerializer;
@@ -46,7 +49,7 @@
 @TestInstance(TestInstance.Lifecycle.PER_CLASS)
 class KafkaSpanExporterIntegrationTest {
  private static final DockerImageName KAFKA_TEST_IMAGE =
- DockerImageName.parse("apache/kafka:3.8.1");
+ DockerImageName.parse("apache/kafka:3.9.1");
  private static final String TOPIC = "span_topic";
  private KafkaContainer kafka;
  private KafkaConsumer<String, ExportTraceServiceRequest> consumer;
@@ -155,6 +158,28 @@ void exportWhenProducerInError() {
  testSubject.shutdown();
  }
 
+ @Test
+ void exportWhenProducerFailsToSend() {
+ var mockProducer = new MockProducer<String, Collection<SpanData>>();
+ mockProducer.sendException = new KafkaException("Simulated kafka exception");
+ var testSubjectWithMockProducer =
+ KafkaSpanExporter.newBuilder().setTopicName(TOPIC).setProducer(mockProducer).build();
+
+ ImmutableList<SpanData> spans =
+ ImmutableList.of(makeBasicSpan("span-1"), makeBasicSpan("span-2"));
+
+ CompletableResultCode actual = testSubjectWithMockProducer.export(spans);
+
+ await()
+ .untilAsserted(
+ () -> {
+ assertThat(actual.isSuccess()).isFalse();
+ assertThat(actual.isDone()).isTrue();
+ });
+
+ testSubjectWithMockProducer.shutdown();
+ }
+
  @Test
  void flush() {
  CompletableResultCode actual = testSubject.flush();