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.connect.integration;
  
  import org.apache.kafka.common.utils.Time;
  
  import java.util.List;
  import java.util.concurrent.CopyOnWriteArrayList;
  import java.util.concurrent.atomic.AtomicInteger;
  
  public class StartAndStopCounter {
  
      private final AtomicInteger startCounter = new AtomicInteger(0);
      private final AtomicInteger stopCounter = new AtomicInteger(0);
      private final List<StartAndStopLatch> restartLatches = new CopyOnWriteArrayList<>();
      private final Time clock;
  
      public StartAndStopCounter() {
          this(Time.SYSTEM);
      }
  
      public StartAndStopCounter(Time clock) {
          this.clock = clock != null ? clock : Time.SYSTEM;
      }
  
      /**
       * Record a start.
       */
      public void recordStart() {
          startCounter.incrementAndGet();
          restartLatches.forEach(StartAndStopLatch::recordStart);
      }
  
      /**
       * Record a stop.
       */
      public void recordStop() {
          stopCounter.incrementAndGet();
          restartLatches.forEach(StartAndStopLatch::recordStop);
      }
  
      /**
       * Get the number of starts.
       *
       * @return the number of starts
       */
      public int starts() {
          return startCounter.get();
      }
  
      /**
       * Get the number of stops.
       *
       * @return the number of stops
       */
      public int stops() {
          return stopCounter.get();
      }
  
      public StartsAndStops countsSnapshot() {
          return new StartsAndStops(starts(), stops());
      }
  
      /**
       * Obtain a {@link StartAndStopLatch} that can be used to wait until the expected number of restarts
       * has been completed.
       *
       * @param expectedStarts   the expected number of starts; may be 0
       * @param expectedStops    the expected number of stops; may be 0
       * @return the latch; never null
       */
      public StartAndStopLatch expectedRestarts(int expectedStarts, int expectedStops) {
          return expectedRestarts(expectedStarts, expectedStops, null);
      }
  
      /**
       * Obtain a {@link StartAndStopLatch} that can be used to wait until the expected number of restarts
       * has been completed.
       *
       * @param expectedStarts   the expected number of starts; may be 0
       * @param expectedStops    the expected number of stops; may be 0
       * @param dependents       any dependent latches that must also complete in order for the
       *                         resulting latch to complete
       * @return the latch; never null
       */
      public StartAndStopLatch expectedRestarts(int expectedStarts, int expectedStops, List<StartAndStopLatch> dependents) {
          StartAndStopLatch latch = new StartAndStopLatch(expectedStarts, expectedStops, this::remove, dependents, clock);
          restartLatches.add(latch);
          return latch;
      }
  
      /**
       * Obtain a {@link StartAndStopLatch} that can be used to wait until the expected number of restarts
       * has been completed.
       *
       * @param expectedRestarts the expected number of restarts
       * @return the latch; never null
       */
      public StartAndStopLatch expectedRestarts(int expectedRestarts) {
          return expectedRestarts(expectedRestarts, expectedRestarts);
      }
  
      /**
       * Obtain a {@link StartAndStopLatch} that can be used to wait until the expected number of starts
       * has been completed.
       *
       * @param expectedStarts the expected number of starts
       * @return the latch; never null
       */
      public StartAndStopLatch expectedStarts(int expectedStarts) {
          return expectedRestarts(expectedStarts, 0);
      }
  
      /**
       * Obtain a {@link StartAndStopLatch} that can be used to wait until the expected number of starts
       * has been completed.
       *
       * @param expectedStarts the expected number of starts
       * @param dependents     any dependent latches that must also complete in order for the
       *                       resulting latch to complete
       * @return the latch; never null
       */
      public StartAndStopLatch expectedStarts(int expectedStarts, List<StartAndStopLatch> dependents) {
          return expectedRestarts(expectedStarts, 0, dependents);
      }
  
  
      /**
       * Obtain a {@link StartAndStopLatch} that can be used to wait until the expected number of
       * stops has been completed.
       *
       * @param expectedStops the expected number of stops
       * @return the latch; never null
       */
      public StartAndStopLatch expectedStops(int expectedStops) {
          return expectedRestarts(0, expectedStops);
      }
  
      /**
       * Obtain a {@link StartAndStopLatch} that can be used to wait until the expected number of
       * stops has been completed.
       *
       * @param expectedStops the expected number of stops
       * @param dependents    any dependent latches that must also complete in order for the
       *                      resulting latch to complete
       * @return the latch; never null
       */
      public StartAndStopLatch expectedStops(int expectedStops, List<StartAndStopLatch> dependents) {
          return expectedRestarts(0, expectedStops, dependents);
      }
  
      protected void remove(StartAndStopLatch restartLatch) {
          restartLatches.remove(restartLatch);
      }
  }
  