add Kafka connection failure handling

Author: vincentfree

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

@@ -72,7 +72,12 @@ public CompletableResultCode export(@Nonnull Collection<SpanData> spans) {
                     result.fail();
                   }
                 }),
-        executorService);
+        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;
@@ -155,6 +158,31 @@ 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();