public StartAndStopLatch expectedStops(int expectedStops, Map<String, Integer> expectedTasksStops, boolean includeTasks) {
        List<StartAndStopLatch> taskLatches = includeTasks
                ? taskHandles.values().stream()
                .map(task -> task.expectedStops(expectedTasksStops.get(task.taskId())))
                .collect(Collectors.toList())
                : Collections.emptyList();
        return startAndStopCounter.expectedStops(expectedStops, taskLatches);
    }
Obtain a {@link StartAndStopLatch} that can be used to wait until the connector using this handle
and optionally all tasks using {@link TaskHandle} have completed the minimum number of
stops, starting the counts at the time this method is called.

<p>A test can call this method, specifying the number of times the connector and tasks
will each be stopped from that point (typically {@code expectedStops(1)}).
The test should then change the connector or otherwise cause the connector to stop (or
restart) one or more times, and then can call
{@link StartAndStopLatch#await(long, TimeUnit)} to wait up to a specified duration for the
connector and all tasks to be started at least the specified number of times.

<p>This method does not track the number of times the connector and tasks are started, and
only tracks the number of times the connector and tasks are <em>stopped</em>.

@param expectedStops the minimum number of starts that are expected once this method is
                     called
@param includeTasks  true if the latch should also wait for the tasks to be stopped the
                     specified minimum number of times
@return the latch that can be used to wait for the starts to complete; never null
connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectorHandle.java