Demonstrate strangeness

Author: jack-berg

sdk/logs/src/test/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessorTest.java

@@ -10,17 +10,21 @@
 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.doThrow;
 import static org.mockito.Mockito.reset;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import io.opentelemetry.api.internal.GuardedBy;
 import io.opentelemetry.internal.testing.slf4j.SuppressLogger;
 import io.opentelemetry.sdk.common.CompletableResultCode;
 import io.opentelemetry.sdk.logs.SdkLoggerProvider;
 import io.opentelemetry.sdk.logs.data.LogRecordData;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -139,6 +143,32 @@ void builderAdjustMaxBatchSize() {
         .isEqualTo(BatchLogRecordProcessorBuilder.DEFAULT_MAX_EXPORT_BATCH_SIZE);
   }
 
+  @Test
+  void maxExportBatchSizeExceedsQueueSize() throws InterruptedException {
+    // Given a processor configured with a maxExportBatchSize > maxQueueSize, ensure that after n =
+    // maxQueueSize logs are emitted, export is triggered and that the queue is fully drained and
+    // exported.
+    int maxQueueSize = 2048;
+    when(mockLogRecordExporter.export(any())).thenReturn(CompletableResultCode.ofSuccess());
+    SdkLoggerProvider sdkLoggerProvider =
+        SdkLoggerProvider.builder()
+            .addLogRecordProcessor(
+                BatchLogRecordProcessor.builder(mockLogRecordExporter)
+                    .setScheduleDelay(Duration.ofSeconds(Integer.MAX_VALUE))
+                    .setMaxExportBatchSize(2049)
+                    .setMaxQueueSize(maxQueueSize)
+                    .build())
+            .build();
+
+    for (int i = 0; i < maxQueueSize; i++) {
+      emitLog(sdkLoggerProvider, "log " + i);
+    }
+
+    Thread.sleep(10);
+
+    verify(mockLogRecordExporter, times(1)).export(any());
+  }
+
   @Test
   void emitMultipleLogs() {
     WaitingLogRecordExporter waitingLogRecordExporter =

sdk/trace/src/test/java/io/opentelemetry/sdk/trace/export/BatchSpanProcessorTest.java

@@ -14,6 +14,8 @@
 import static org.mockito.ArgumentMatchers.argThat;
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.reset;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import io.opentelemetry.api.internal.GuardedBy;
@@ -26,6 +28,7 @@
 import io.opentelemetry.sdk.trace.data.SpanData;
 import io.opentelemetry.sdk.trace.samplers.Sampler;
 import io.opentelemetry.sdk.trace.samplers.SamplingResult;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -144,6 +147,32 @@ void builderAdjustMaxBatchSize() {
         .isEqualTo(BatchSpanProcessorBuilder.DEFAULT_MAX_EXPORT_BATCH_SIZE);
   }
 
+  @Test
+  void maxExportBatchSizeExceedsQueueSize() throws InterruptedException {
+    // Given a processor configured with a maxExportBatchSize > maxQueueSize, ensure that after n =
+    // maxQueueSize spans are ended, export is triggered and that the queue is fully drained and
+    // exported.
+    int maxQueueSize = 2048;
+    when(mockSpanExporter.export(any())).thenReturn(CompletableResultCode.ofSuccess());
+    sdkTracerProvider =
+        SdkTracerProvider.builder()
+            .addSpanProcessor(
+                BatchSpanProcessor.builder(mockSpanExporter)
+                    .setScheduleDelay(Duration.ofSeconds(Integer.MAX_VALUE))
+                    .setMaxExportBatchSize(2049)
+                    .setMaxQueueSize(maxQueueSize)
+                    .build())
+            .build();
+
+    for (int i = 0; i < maxQueueSize; i++) {
+      createEndedSpan("span " + i);
+    }
+
+    Thread.sleep(10);
+
+    verify(mockSpanExporter, times(1)).export(any());
+  }
+
   @Test
   void startEndRequirements() {
     BatchSpanProcessor spansProcessor =