protected void waitForConnectorState(
            String connectorName,
            AbstractStatus.State connectorState,
            Predicate<Integer> expectedNumTasks,
            Integer numTasksInTasksState,
            AbstractStatus.State tasksState,
            String conditionMessage,
            String detailMessage,
            long maxWaitMs
    ) throws InterruptedException {
        AtomicReference<ConnectorStateInfo> lastInfo = new AtomicReference<>();
        AtomicReference<Exception> lastInfoError = new AtomicReference<>();
        try {
            waitForCondition(
                    () -> {
                        try {
                            ConnectorStateInfo info = connect.connectorStatus(connectorName);
                            lastInfo.set(info);
                            lastInfoError.set(null);

                            if (info == null)
                                return false;

                            int numTasks = info.tasks().size();
                            int expectedTasksInState = Optional.ofNullable(numTasksInTasksState).orElse(numTasks);
                            return expectedNumTasks.test(info.tasks().size())
                                    && info.connector().state().equals(connectorState.toString())
                                    && info.tasks().stream().filter(s -> s.state().equals(tasksState.toString())).count() == expectedTasksInState;
                        } catch (Exception e) {
                            log.error("Could not check connector state info.", e);
                            lastInfo.set(null);
                            lastInfoError.set(e);
                            return false;
                        }
                    },
                    maxWaitMs,
                    () -> {
                        String result = conditionMessage;
                        if (lastInfo.get() != null) {
                            return result + " When last checked, " + stateSummary(lastInfo.get());
                        } else if (lastInfoError.get() != null) {
                            result +=  " The last attempt to check the connector state failed: " + lastInfoError.get().getClass();
                            String exceptionMessage = lastInfoError.get().getMessage();
                            if (exceptionMessage != null) {
                                result += ": " + exceptionMessage;
                            }
                            return result;
                        } else {
                            return result;
                        }
                    }
            );
        } catch (AssertionError e) {
            throw new AssertionError(detailMessage, e);
        }
    }
Check whether the given connector state matches the current state of the connector and
whether it has at least the given number of tasks, with some number of tasks matching the given
task state.
@param connectorName the connector
@param connectorState
@param expectedNumTasks the expected number of tasks
@param tasksState
connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/ConnectAssertions.java