diff --git a/sdk/logs/src/main/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessor.java b/sdk/logs/src/main/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessor.java index 317f79f324d..374d49e7abc 100644 --- a/sdk/logs/src/main/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessor.java +++ b/sdk/logs/src/main/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessor.java @@ -11,6 +11,7 @@ import io.opentelemetry.sdk.common.InternalTelemetryVersion; import io.opentelemetry.sdk.common.internal.ComponentId; import io.opentelemetry.sdk.common.internal.DaemonThreadFactory; +import io.opentelemetry.sdk.common.internal.ThrowableUtil; import io.opentelemetry.sdk.logs.LogRecordProcessor; import io.opentelemetry.sdk.logs.ReadWriteLogRecord; import io.opentelemetry.sdk.logs.data.LogRecordData; @@ -301,9 +302,10 @@ private void exportCurrentBatch() { error = "export_failed"; } } - } catch (RuntimeException e) { - logger.log(Level.WARNING, "Exporter threw an Exception", e); - error = e.getClass().getName(); + } catch (Throwable t) { + ThrowableUtil.propagateIfFatal(t); + logger.log(Level.WARNING, "Exporter threw an Exception", t); + error = t.getClass().getName(); } finally { logProcessorInstrumentation.finishLogs(batch.size(), error); batch.clear(); diff --git a/sdk/logs/src/test/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessorTest.java b/sdk/logs/src/test/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessorTest.java index 144ab7c35f6..579d6d2bc02 100644 --- a/sdk/logs/src/test/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessorTest.java +++ b/sdk/logs/src/test/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessorTest.java @@ -6,13 +6,13 @@ package io.opentelemetry.sdk.logs.export; import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.assertThat; -import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.assertj.core.api.AssertionsForClassTypes.assertThatCode; import static org.awaitility.Awaitility.await; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyList; import static org.mockito.ArgumentMatchers.argThat; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.reset; import static org.mockito.Mockito.times; @@ -32,6 +32,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import javax.annotation.Nullable; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -376,6 +377,35 @@ void exporterThrowsException() { .satisfiesExactly(logRecordData -> assertThat(logRecordData).hasBody(LOG_MESSAGE_2)); } + @Test + @Timeout(5) + @SuppressLogger(BatchLogRecordProcessor.class) + @SuppressLogger(loggerName = "io.opentelemetry.sdk.logs.export.BatchLogRecordProcessor$Worker") + void exporterThrowsNonRuntimeException() { + AtomicInteger exportCount = new AtomicInteger(); + doAnswer( + invocation -> { + if (exportCount.incrementAndGet() == 1) { + throw new Exception("simulated non-RuntimeException"); + } + return CompletableResultCode.ofSuccess(); + }) + .when(mockLogRecordExporter) + .export(anyList()); + BatchLogRecordProcessor processor = + BatchLogRecordProcessor.builder(mockLogRecordExporter) + .setScheduleDelay(MAX_SCHEDULE_DELAY_MILLIS, TimeUnit.MILLISECONDS) + .build(); + SdkLoggerProvider loggerProvider = + SdkLoggerProvider.builder().addLogRecordProcessor(processor).build(); + + emitLog(loggerProvider, LOG_MESSAGE_1); + await().untilAsserted(() -> assertThat(exportCount.get()).isGreaterThanOrEqualTo(1)); + // Confirm worker is still alive after the non-RuntimeException + emitLog(loggerProvider, LOG_MESSAGE_2); + await().untilAsserted(() -> assertThat(exportCount.get()).isGreaterThanOrEqualTo(2)); + } + @Test @Timeout(5) public void continuesIfExporterTimesOut() throws InterruptedException {