Repository: open-telemetry/opentelemetry-java
File: sdk_contrib/async_processor/src/main/java/io/opentelemetry/sdk/contrib/trace/export/DisruptorEventQueue.java (Java)

Code before:
    synchronized (this) {
      if (isShutdown) {
        return;
      }
      enqueue(null, EventType.ON_SHUTDOWN);
      isShutdown = true;
      try {
        shutdownCounter.await();
      } catch (InterruptedException e) {
        // Preserve the interruption.
        Thread.currentThread().interrupt();
        logger.warning("Thread interrupted, shutdown may not finished.");
      }
    }
  }

  // An event in the {@link EventQueue}. Just holds a reference to an EventQueue.Entry.
  private static final class DisruptorEvent {
    @Nullable private ReadableSpan readableSpan = null;
    @Nullable private EventType eventType = null;

    // Sets the EventQueueEntry associated with this DisruptorEvent.
    void setEntry(@Nullable ReadableSpan readableSpan, @Nullable EventType eventType) {
      this.readableSpan = readableSpan;
      this.eventType = eventType;
    }

    @Nullable
    ReadableSpan getReadableSpan() {
      return readableSpan;
    }

    @Nullable
    EventType getEventType() {
      return eventType;
    }
  }

  private static final class DisruptorEventHandler implements EventHandler<DisruptorEvent> {
    private final SpanProcessor spanProcessor;
    private final CountDownLatch shutdownCounter;

    private DisruptorEventHandler(SpanProcessor spanProcessor, CountDownLatch shutdownCounter) {
      this.spanProcessor = spanProcessor;
      this.shutdownCounter = shutdownCounter;
    }

    @Override
    public void onEvent(DisruptorEvent event, long sequence, boolean endOfBatch) {
      final ReadableSpan readableSpan = event.getReadableSpan();
      final EventType eventType = event.getEventType();

Review comment (suggestion):
This is not waiting for the processor thread to actually call the flush maybe just document it for the moment and add a TODO to wait for flush to be called. Or simply create an issue.

Code after:
    }
    synchronized (this) {
      if (isShutdown) {
        return;
      }
      enqueue(null, EventType.ON_SHUTDOWN);
      isShutdown = true;
      try {
        shutdownCounter.await();
      } catch (InterruptedException e) {
        // Preserve the interruption.
        Thread.currentThread().interrupt();
        logger.warning("Thread interrupted, shutdown may not finished.");
      }
    }
  }

  void forceFlush() {
    if (isShutdown) {
      return;
    }
    synchronized (this) {
      if (isShutdown) {
        return;
      }
      enqueue(null, EventType.ON_FORCE_FLUSH);
    }
  }

  // An event in the {@link EventQueue}. Just holds a reference to an EventQueue.Entry.
  private static final class DisruptorEvent {
    @Nullable private ReadableSpan readableSpan = null;
    @Nullable private EventType eventType = null;

    // Sets the EventQueueEntry associated with this DisruptorEvent.
    void setEntry(@Nullable ReadableSpan readableSpan, @Nullable EventType eventType) {
      this.readableSpan = readableSpan;
      this.eventType = eventType;
    }

    @Nullable
    ReadableSpan getReadableSpan() {
      return readableSpan;
    }

    @Nullable
    EventType getEventType() {
      return eventType;
    }
  }