public void sanityCheckDryRun(boolean dryRun, boolean stopOngoingExecution) {
    if (dryRun) {
      return;
    }
    if (hasOngoingExecution()) {
      if (!stopOngoingExecution) {
        throw new IllegalStateException(String.format("Cannot start a new execution while there is an ongoing execution. "
                                                      + "Please use %s=true to stop ongoing execution and start a new one.",
                                                      STOP_ONGOING_EXECUTION_PARAM));
      }
    } else {
      Set<TopicPartition> partitionsBeingReassigned;
      try {
        partitionsBeingReassigned = _executor.listPartitionsBeingReassigned();
      } catch (TimeoutException | InterruptedException | ExecutionException e) {
        // This may indicate transient (e.g. network) issues.
        throw new IllegalStateException("Cannot execute new proposals due to failure to retrieve whether the Kafka cluster has "
                                        + "an already ongoing partition reassignment.", e);
      }
      if (!partitionsBeingReassigned.isEmpty()) {
        if (_config.getBoolean(ExecutorConfig.AUTO_STOP_EXTERNAL_AGENT_CONFIG)) {
          // Stop the external agent reassignment.
          if (_executor.maybeStopExternalAgent()) {
            LOG.info("External agent is reassigning partitions. "
                     + "The request to stop it is submitted successfully: {}", partitionsBeingReassigned);
          }
        } else {
          throw new IllegalStateException(String.format("Cannot execute new proposals while there are ongoing partition reassignments "
                                                        + "initiated by external agent: %s", partitionsBeingReassigned));
        }
      }
    }
  }
@Test
  public void testSanityCheckDryRun() throws InterruptedException, ExecutionException, TimeoutException {
    Time time = EasyMock.mock(Time.class);
    AnomalyDetectorManager anomalyDetectorManager = EasyMock.mock(AnomalyDetectorManager.class);
    Executor executor = EasyMock.strictMock(Executor.class);
    LoadMonitor loadMonitor = EasyMock.mock(LoadMonitor.class);
    ExecutorService goalOptimizerExecutor = EasyMock.mock(ExecutorService.class);
    GoalOptimizer goalOptimizer = EasyMock.mock(GoalOptimizer.class);

    // For sanityCheckDryRun(false, true) and sanityCheckDryRun(false, false) (see #1 and #2 below).
    EasyMock.expect(executor.hasOngoingExecution()).andReturn(true).times(2);
    // For sanityCheckDryRun(false, XXX) (see #3 below)
    EasyMock.expect(executor.hasOngoingExecution()).andReturn(false).once();
    EasyMock.expect(executor.listPartitionsBeingReassigned()).andReturn(DUMMY_ONGOING_PARTITION_REASSIGNMENTS);
    EasyMock.expect(executor.maybeStopExternalAgent()).andReturn(true);
    // For sanityCheckDryRun(false, XXX) (see #4 below)
    EasyMock.expect(executor.hasOngoingExecution()).andReturn(false).once();
    EasyMock.expect(executor.listPartitionsBeingReassigned()).andReturn(Collections.emptySet());
    // For sanityCheckDryRun(false, XXX) (see #5 below)
    EasyMock.expect(executor.hasOngoingExecution()).andReturn(false).once();
    EasyMock.expect(executor.listPartitionsBeingReassigned()).andThrow(new TimeoutException()).once();

    EasyMock.replay(time, anomalyDetectorManager, executor, loadMonitor, goalOptimizerExecutor, goalOptimizer);
    KafkaCruiseControl kafkaCruiseControl = new KafkaCruiseControl(_config, time, anomalyDetectorManager, executor,
                                                                   loadMonitor, goalOptimizerExecutor, goalOptimizer, new NoopProvisioner());

    // Expect no failure (dryrun = true) regardless of ongoing executions.
    kafkaCruiseControl.sanityCheckDryRun(true, false);
    kafkaCruiseControl.sanityCheckDryRun(true, true);
    // 1. Expect no failure (dryrun = false), if there is ongoing execution started by CC, it must be requested to stop.
    kafkaCruiseControl.sanityCheckDryRun(false, true);
    // 2. Expect failure (dryrun = false), if there is ongoing execution started by CC, not requested to stop.
    assertThrows(IllegalStateException.class, () -> kafkaCruiseControl.sanityCheckDryRun(false, false));
    // 3. Expect no failure (dryrun = false), there is no execution started by CC, but ongoing replica reassignment, request to stop is irrelevant.
    kafkaCruiseControl.sanityCheckDryRun(false, false);
    // 4. Expect no failure (dryrun = false), there is no execution started by CC or other tools, request to stop is irrelevant.
    kafkaCruiseControl.sanityCheckDryRun(false, false);
    // 5. Expect failure (dryrun = false), there is no execution started by CC, but checking ongoing executions started
    // by other tools timed out, request to stop is irrelevant.
    assertThrows(IllegalStateException.class, () -> kafkaCruiseControl.sanityCheckDryRun(false, false));

    EasyMock.verify(time, anomalyDetectorManager, executor, loadMonitor, goalOptimizerExecutor, goalOptimizer);

    // Verify initialization and functioning of Admin Client
    AdminClient adminClient = kafkaCruiseControl.adminClient();
    assertNotNull(adminClient);
    assertEquals(clusterSize(), adminClient.describeCluster().nodes().get(CLIENT_REQUEST_TIMEOUT_MS, TimeUnit.MILLISECONDS).size());
  }