BatchLogRecordProcessor enforced to end of processor pipeline

Author: onurkybsi

sdk/logs/src/main/java/io/opentelemetry/sdk/logs/LogRecordProcessor.java

@@ -9,6 +9,7 @@
 import io.opentelemetry.api.logs.Logger;
 import io.opentelemetry.context.Context;
 import io.opentelemetry.sdk.common.CompletableResultCode;
+import io.opentelemetry.sdk.logs.export.BatchLogRecordProcessor;
 import java.io.Closeable;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -39,15 +40,26 @@ static LogRecordProcessor composite(LogRecordProcessor... processors) {
    */
   static LogRecordProcessor composite(Iterable<LogRecordProcessor> processors) {
     List<LogRecordProcessor> processorList = new ArrayList<>();
+    BatchLogRecordProcessor batchProcessor = null;
     for (LogRecordProcessor processor : processors) {
-      processorList.add(processor);
+      if (!BatchLogRecordProcessor.class.equals(processor.getClass())) {
+        processorList.add(processor);
+      } else {
+        batchProcessor = (BatchLogRecordProcessor) processor;
+      }
     }
+    if (batchProcessor != null) {
+      processorList.add(batchProcessor);
+    }
+
     if (processorList.isEmpty()) {
       return NoopLogRecordProcessor.getInstance();
     }
+
     if (processorList.size() == 1) {
       return processorList.get(0);
     }
+
     return MultiLogRecordProcessor.create(processorList);
   }
 

sdk/logs/src/test/java/io/opentelemetry/sdk/logs/SdkLoggerProviderTest.java

@@ -29,6 +29,7 @@
 import io.opentelemetry.sdk.common.CompletableResultCode;
 import io.opentelemetry.sdk.common.InstrumentationScopeInfo;
 import io.opentelemetry.sdk.logs.data.LogRecordData;
+import io.opentelemetry.sdk.logs.export.BatchLogRecordProcessor;
 import io.opentelemetry.sdk.resources.Resource;
 import java.util.Optional;
 import java.util.concurrent.TimeUnit;
@@ -142,6 +143,29 @@ void builder_multipleProcessors() {
             });
   }
 
+  @Test
+  void builder_multipleProcessorsWithBatchProcessor() {
+    assertThat(
+            SdkLoggerProvider.builder()
+                .addLogRecordProcessor(mock(BatchLogRecordProcessor.class))
+                .addLogRecordProcessor(logRecordProcessor)
+                .addLogRecordProcessor(logRecordProcessor)
+                .build())
+        .extracting("sharedState", as(InstanceOfAssertFactories.type(LoggerSharedState.class)))
+        .extracting(LoggerSharedState::getLogRecordProcessor)
+        .satisfies(
+            activeLogRecordProcessor -> {
+              assertThat(activeLogRecordProcessor).isInstanceOf(MultiLogRecordProcessor.class);
+              assertThat(activeLogRecordProcessor)
+                  .extracting(
+                      "logRecordProcessors",
+                      as(InstanceOfAssertFactories.list(LogRecordProcessor.class)))
+                  .hasSize(3)
+                  .last()
+                  .isInstanceOf(BatchLogRecordProcessor.class);
+            });
+  }
+
   @Test
   void loggerBuilder_SameName() {
     assertThat(sdkLoggerProvider.loggerBuilder("test").build())