@Override
    public ConnectorStateInfo connectorStatus(String connName) {
        ConnectorStatus connector = statusBackingStore.get(connName);
        if (connector == null)
            throw new NotFoundException("No status found for connector " + connName);

        Collection<TaskStatus> tasks = statusBackingStore.getAll(connName);

        ConnectorStateInfo.ConnectorState connectorState = new ConnectorStateInfo.ConnectorState(
                connector.state().toString(), connector.workerId(), connector.trace());
        List<ConnectorStateInfo.TaskState> taskStates = new ArrayList<>();

        for (TaskStatus status : tasks) {
            taskStates.add(new ConnectorStateInfo.TaskState(status.id().task(),
                    status.state().toString(), status.workerId(), status.trace()));
        }

        Collections.sort(taskStates);

        Map<String, String> conf = rawConfig(connName);
        return new ConnectorStateInfo(connName, connectorState, taskStates, connectorType(conf));
    }
@Test
    public void testConnectorStatus() {
        ConnectorTaskId taskId = new ConnectorTaskId(connectorName, 0);

        AbstractHerder herder = testHerder();

        when(plugins.newConnector(anyString())).thenReturn(new SampleSourceConnector());
        when(herder.plugins()).thenReturn(plugins);

        when(herder.rawConfig(connectorName)).thenReturn(Collections.singletonMap(
                ConnectorConfig.CONNECTOR_CLASS_CONFIG, SampleSourceConnector.class.getName()
        ));

        when(statusStore.get(connectorName))
                .thenReturn(new ConnectorStatus(connectorName, AbstractStatus.State.RUNNING, workerId, generation));

        when(statusStore.getAll(connectorName))
                .thenReturn(Collections.singletonList(
                        new TaskStatus(taskId, AbstractStatus.State.UNASSIGNED, workerId, generation)));

        ConnectorStateInfo state = herder.connectorStatus(connectorName);

        assertEquals(connectorName, state.name());
        assertEquals(ConnectorType.SOURCE, state.type());
        assertEquals("RUNNING", state.connector().state());
        assertEquals(1, state.tasks().size());
        assertEquals(workerId, state.connector().workerId());

        ConnectorStateInfo.TaskState taskState = state.tasks().get(0);
        assertEquals(0, taskState.id());
        assertEquals("UNASSIGNED", taskState.state());
        assertEquals(workerId, taskState.workerId());
    }