public void runEarlyPrepareNodes(StartControllers controllers, String operationName, byte[] data,
    String... cohort) throws Exception {
    ZKWatcher watcher = UTIL.getZooKeeperWatcher();
    List<String> expected = Lists.newArrayList(cohort);

    final Subprocedure sub = Mockito.mock(Subprocedure.class);
    Mockito.when(sub.getName()).thenReturn(operationName);

    final CountDownLatch prepared = new CountDownLatch(expected.size());
    final CountDownLatch committed = new CountDownLatch(expected.size());
    ArrayList<byte[]> dataFromMembers = new ArrayList<>();

    // mock out coordinator so we can keep track of zk progress
    ProcedureCoordinator coordinator =
      setupMockCoordinator(operationName, prepared, committed, dataFromMembers);

    ProcedureMember member = Mockito.mock(ProcedureMember.class);
    Procedure p = Mockito.mock(Procedure.class);
    Mockito.when(p.getName()).thenReturn(operationName);

    Pair<ZKProcedureCoordinator, List<ZKProcedureMemberRpcs>> pair = controllers.start(watcher,
      operationName, coordinator, CONTROLLER_NODE_NAME, member, expected);
    ZKProcedureCoordinator controller = pair.getFirst();
    List<ZKProcedureMemberRpcs> cohortControllers = pair.getSecond();

    // post 1/2 the prepare nodes early
    for (int i = 0; i < cohortControllers.size() / 2; i++) {
      cohortControllers.get(i).sendMemberAcquired(sub);
    }

    // start the operation
    controller.sendGlobalBarrierAcquire(p, data, expected);

    // post the prepare node for each expected node
    for (ZKProcedureMemberRpcs cc : cohortControllers) {
      cc.sendMemberAcquired(sub);
    }

    // wait for all the notifications to reach the coordinator
    prepared.await();
    // make sure we got the all the nodes and no more
    Mockito.verify(coordinator, times(expected.size()))
      .memberAcquiredBarrier(Mockito.eq(operationName), Mockito.anyString());

    // kick off the commit phase
    controller.sendGlobalBarrierReached(p, expected);

    // post the committed node for each expected node
    for (ZKProcedureMemberRpcs cc : cohortControllers) {
      cc.sendMemberCompleted(sub, memberData);
    }

    // wait for all commit notifications to reach the coordiantor
    committed.await();
    // make sure we got the all the nodes and no more
    Mockito.verify(coordinator, times(expected.size())).memberFinishedBarrier(
      Mockito.eq(operationName), Mockito.anyString(), Mockito.eq(memberData));

    controller.resetMembers(p);

    // verify all behavior
    verifyZooKeeperClean(operationName, watcher, controller.getZkProcedureUtil());
    verifyCohort(member, cohortControllers.size(), operationName, data);
    verifyCoordinator(operationName, coordinator, expected);
  }
Smaller test to just test the actuation on the cohort member
@throws Exception on failure
hbase-server/src/test/java/org/apache/hadoop/hbase/procedure/TestZKProcedureControllers.java