content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements. See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License. You may obtain a copy of the License at
   *
   *    http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.kafka.coordinator.common.runtime;
  
  import org.apache.kafka.common.TopicPartition;
  import org.apache.kafka.common.utils.LogContext;
  import org.apache.kafka.common.utils.Time;
  
  import org.slf4j.Logger;
  
  import java.util.List;
  import java.util.Objects;
  import java.util.concurrent.RejectedExecutionException;
  import java.util.concurrent.TimeUnit;
  import java.util.stream.Collectors;
  import java.util.stream.IntStream;
  
  /**
   * A multithreaded {{@link CoordinatorEvent}} processor which uses a {{@link EventAccumulator}}
   * which guarantees that events sharing a partition key are not processed concurrently.
   */
  public final class MultiThreadedEventProcessor implements CoordinatorEventProcessor {
  
      /**
       * The poll timeout to wait for an event by the EventProcessorThread.
       */
      private static final long POLL_TIMEOUT_MS = 300L;
  
      /**
       * The logger.
       */
      private final Logger log;
  
      /**
       * The accumulator.
       */
      private final EventAccumulator<TopicPartition, CoordinatorEvent> accumulator;
  
      /**
       * The processing threads.
       */
      private final List<EventProcessorThread> threads;
  
      /**
       * A boolean indicated whether the event processor is shutting down.
       */
      private volatile boolean shuttingDown;
  
      /**
       * The coordinator runtime metrics.
       */
      private final CoordinatorRuntimeMetrics metrics;
  
      /**
       * The time.
       */
      private final Time time;
  
      public MultiThreadedEventProcessor(
          LogContext logContext,
          String threadPrefix,
          int numThreads,
          Time time,
          CoordinatorRuntimeMetrics metrics
      ) {
          this(logContext, threadPrefix, numThreads, time, metrics, new EventAccumulator<>());
      }
  
      /**
       * Constructor.
       *
       * @param logContext        The log context.
       * @param threadPrefix      The thread prefix.
       * @param numThreads        The number of threads.
       * @param metrics           The coordinator runtime metrics.
       * @param time              The time.
       * @param eventAccumulator  The event accumulator.
       */
      public MultiThreadedEventProcessor(
          LogContext logContext,
          String threadPrefix,
          int numThreads,
          Time time,
          CoordinatorRuntimeMetrics metrics,
          EventAccumulator<TopicPartition, CoordinatorEvent> eventAccumulator
      ) {
          this.log = logContext.logger(MultiThreadedEventProcessor.class);
          this.shuttingDown = false;
          this.accumulator = eventAccumulator;
          this.time = Objects.requireNonNull(time);
          this.metrics = Objects.requireNonNull(metrics);
          this.metrics.registerEventQueueSizeGauge(accumulator::size);
          this.threads = IntStream.range(0, numThreads).mapToObj(threadId ->
              new EventProcessorThread(
                  threadPrefix + threadId
              )
          ).collect(Collectors.toList());
          this.threads.forEach(EventProcessorThread::start);
      }
  
      /**
       * The event processor thread. The thread pulls events from the
       * accumulator and runs them.
       */
      private class EventProcessorThread extends Thread {
          private final Logger log;
  
          EventProcessorThread(
              String name
          ) {
              super(name);
              log = new LogContext("[" + name + "]: ").logger(EventProcessorThread.class);
              setDaemon(false);
          }
  
          private void handleEvents() {
              while (!shuttingDown) {
                  // We use a single meter for aggregate idle percentage for the thread pool.
                  // Since meter is calculated as total_recorded_value / time_window and
                  // time_window is independent of the number of threads, each recorded idle
                  // time should be discounted by # threads.
  
                  long idleStartTimeMs = time.milliseconds();
                  CoordinatorEvent event = accumulator.poll(POLL_TIMEOUT_MS, TimeUnit.MILLISECONDS);
                  long idleEndTimeMs = time.milliseconds();
                  long idleTimeMs = idleEndTimeMs - idleStartTimeMs;
                  metrics.recordThreadIdleTime((double) idleTimeMs / (double) threads.size());
                  if (event != null) {
                      try {
                          log.debug("Executing event: {}.", event);
                          long dequeuedTimeMs = time.milliseconds();
                          metrics.recordEventQueueTime(dequeuedTimeMs - event.createdTimeMs());
                          event.run();
                          metrics.recordEventProcessingTime(time.milliseconds() - dequeuedTimeMs);
                      } catch (Throwable t) {
                          log.error("Failed to run event {} due to: {}.", event, t.getMessage(), t);
                          event.complete(t);
                      } finally {
                          accumulator.done(event);
                      }
                  }
              }
          }
  
          private void drainEvents() {
              CoordinatorEvent event;
              while ((event = accumulator.poll()) != null) {
                  try {
                      log.debug("Draining event: {}.", event);
                      metrics.recordEventQueueTime(time.milliseconds() - event.createdTimeMs());
                      event.complete(new RejectedExecutionException("EventProcessor is closed."));
                  } catch (Throwable t) {
                      log.error("Failed to reject event {} due to: {}.", event, t.getMessage(), t);
                  } finally {
                      accumulator.done(event);
                  }
              }
          }
  
          @Override
          public void run() {
              log.info("Starting");
  
              try {
                  handleEvents();
              } catch (Throwable t) {
                  log.error("Exiting with exception.", t);
              }
  
              // The accumulator is drained and all the pending events are rejected
              // when the event processor is shutdown.
              if (shuttingDown) {
                  log.info("Shutting down. Draining the remaining events.");
                  try {
                      drainEvents();
                  } catch (Throwable t) {
                      log.error("Draining threw exception.", t);
                  }
                  log.info("Shutdown completed");
              }
          }
      }
  
      /**
       * Enqueues a new {{@link CoordinatorEvent}} at the end of the processor.
       *
       * @param event The event.
       * @throws RejectedExecutionException If the event processor is closed.
       */
      @Override
      public void enqueueLast(CoordinatorEvent event) throws RejectedExecutionException {
          accumulator.addLast(event);
      }
  
      /**
       * Enqueues a new {{@link CoordinatorEvent}} at the front of the processor.
       *
       * @param event The event.
       * @throws RejectedExecutionException If the event processor is closed.
       */
      @Override
      public void enqueueFirst(CoordinatorEvent event) throws RejectedExecutionException {
          accumulator.addFirst(event);
      }
  
      /**
       * Begins the shutdown of the event processor.
       */
      public synchronized void beginShutdown() {
          if (shuttingDown) {
              log.debug("Event processor is already shutting down.");
              return;
          }
  
          log.info("Shutting down event processor.");
          // The accumulator must be closed first to ensure that new events are
          // rejected before threads are notified to shutdown and start to drain
          // the accumulator.
          accumulator.close();
          shuttingDown = true;
      }
  
      /**
       * Closes the event processor.
       */
      @Override
      public void close() throws InterruptedException {
          beginShutdown();
          for (Thread t : threads) {
              t.join();
          }
          log.info("Event processor closed.");
      }
  }
  