SimpleLogRecordProcessor/SimpleSpanProcessor.forceFlush() result never completes when an export completes while ofAll() counts the pending set #8995

import io.opentelemetry.api.logs.Logger;
import io.opentelemetry.sdk.common.CompletableResultCode;
import io.opentelemetry.sdk.logs.SdkLoggerProvider;
import io.opentelemetry.sdk.logs.data.LogRecordData;
import io.opentelemetry.sdk.logs.export.LogRecordExporter;
import io.opentelemetry.sdk.logs.export.SimpleLogRecordProcessor;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/** SimpleLogRecordProcessor.forceFlush() may never complete when an export completes while ofAll() counts the pending set. */
public class ForceFlushRace {
  public static void main(String[] args) throws Exception {
    int rounds = args.length > 0 ? Integer.parseInt(args[0]) : 2000;
    int exportsPerRound = args.length > 1 ? Integer.parseInt(args[1]) : 5;
    List<CompletableResultCode> exports = Collections.synchronizedList(new ArrayList<>());
    LogRecordExporter exporter = new LogRecordExporter() {
      @Override public CompletableResultCode export(Collection<LogRecordData> logs) {
        CompletableResultCode result = new CompletableResultCode(); // completed later, on another thread
        exports.add(result);
        return result;
      }
      @Override public CompletableResultCode flush() { return CompletableResultCode.ofSuccess(); }
      @Override public CompletableResultCode shutdown() { return CompletableResultCode.ofSuccess(); }
    };
    SdkLoggerProvider provider = SdkLoggerProvider.builder()
        .addLogRecordProcessor(SimpleLogRecordProcessor.create(exporter)).build();
    Logger logger = provider.get("probe");
    ExecutorService completers = Executors.newFixedThreadPool(exportsPerRound);
    int neverCompleted = 0;
    for (int round = 0; round < rounds; round++) {
      exports.clear();
      for (int i = 0; i < exportsPerRound; i++) logger.logRecordBuilder().setBody("line " + i).emit();
      CountDownLatch go = new CountDownLatch(1);
      for (CompletableResultCode export : new ArrayList<>(exports)) {
        completers.execute(() -> { try { go.await(); export.succeed(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } });
      }
      go.countDown(); // the exports complete while forceFlush() counts and iterates the pending set
      CompletableResultCode flush = provider.forceFlush();
      flush.join(1, TimeUnit.SECONDS);
      if (!flush.isDone()) {
        neverCompleted++;
        // every export is done and a fresh forceFlush() says so at once; only the first result is stuck
        if (!exports.stream().allMatch(CompletableResultCode::isSuccess) || !provider.forceFlush().isSuccess()) throw new AssertionError();
      }
    }
    System.out.printf("%d of %d forceFlush() results never completed although every export had completed (%d exports per round)%n",
        neverCompleted, rounds, exportsPerRound);
    completers.shutdownNow();
    provider.close();
    System.exit(0);
  }
}

// SimpleLogRecordProcessor and SimpleSpanProcessor
public CompletableResultCode forceFlush() {
  return CompletableResultCode.ofAll(new ArrayList<>(pendingExports));
}