@Test
public void testLeaderPerformAssignment1() throws Exception {

    // Since all the protocol responses are mocked, the other tests validate doSync runs, but don't validate its
    // output. So we test it directly here.
    EasyMock.expect(configStorage.snapshot()).andReturn(configState1);
    PowerMock.replayAll();
    // Prime the current configuration state
    coordinator.metadata();
    Map<String, ByteBuffer> configs = new HashMap<>();
    // Mark everyone as in sync with configState1
    configs.put("leader", ConnectProtocol.serializeMetadata(new ConnectProtocol.WorkerState(LEADER_URL, 1L)));
    configs.put("member", ConnectProtocol.serializeMetadata(new ConnectProtocol.WorkerState(MEMBER_URL, 1L)));
    Map<String, ByteBuffer> result = Whitebox.invokeMethod(coordinator, "performAssignment", "leader", WorkerCoordinator.DEFAULT_SUBPROTOCOL, configs);
    // configState1 has 1 connector, 1 task
    ConnectProtocol.Assignment leaderAssignment = ConnectProtocol.deserializeAssignment(result.get("leader"));
    assertEquals(false, leaderAssignment.failed());
    assertEquals("leader", leaderAssignment.leader());
    assertEquals(1, leaderAssignment.offset());
    assertEquals(Collections.singletonList(connectorId), leaderAssignment.connectors());
    assertEquals(Collections.emptyList(), leaderAssignment.tasks());
    ConnectProtocol.Assignment memberAssignment = ConnectProtocol.deserializeAssignment(result.get("member"));
    assertEquals(false, memberAssignment.failed());
    assertEquals("leader", memberAssignment.leader());
    assertEquals(1, memberAssignment.offset());
    assertEquals(Collections.emptyList(), memberAssignment.connectors());
    assertEquals(Collections.singletonList(taskId0), memberAssignment.tasks());
    PowerMock.verifyAll();
}