@Override
    public synchronized void restartTask(ConnectorTaskId taskId, Callback<Void> cb) {
        if (!configState.contains(taskId.connector()))
            cb.onCompletion(new NotFoundException("Connector " + taskId.connector() + " not found", null), null);

        Map<String, String> taskConfigProps = configState.taskConfig(taskId);
        if (taskConfigProps == null)
            cb.onCompletion(new NotFoundException("Task " + taskId + " not found", null), null);
        Map<String, String> connConfigProps = configState.connectorConfig(taskId.connector());

        worker.stopAndAwaitTask(taskId);
        if (startTask(taskId, connConfigProps))
            cb.onCompletion(null, null);
        else
            cb.onCompletion(new ConnectException("Failed to start task: " + taskId), null);
    }
@Test
    public void testRestartTask() throws Exception {
        initialize(true);
        ConnectorTaskId taskId = new ConnectorTaskId(CONNECTOR_NAME, 0);
        expectAdd(SourceSink.SOURCE);

        Map<String, String> connectorConfig = connectorConfig(SourceSink.SOURCE);

        expectConfigValidation(SourceSink.SOURCE, connectorConfig);

        doNothing().when(worker).stopAndAwaitTask(taskId);

        ClusterConfigState configState = new ClusterConfigState(
            -1,
            null,
            Collections.singletonMap(CONNECTOR_NAME, 1),
            Collections.singletonMap(CONNECTOR_NAME, connectorConfig),
            Collections.singletonMap(CONNECTOR_NAME, TargetState.STARTED),
            Collections.singletonMap(taskId, taskConfig(SourceSink.SOURCE)),
            Collections.emptyMap(),
            Collections.emptyMap(),
            Collections.singletonMap(CONNECTOR_NAME, new AppliedConnectorConfig(connectorConfig)),
            new HashSet<>(),
            new HashSet<>(),
            transformer);
        when(worker.startSourceTask(taskId, configState, connectorConfig, taskConfig(SourceSink.SOURCE), herder, TargetState.STARTED))
            .thenReturn(true);

        herder.putConnectorConfig(CONNECTOR_NAME, connectorConfig, false, createCallback);
        Herder.Created<ConnectorInfo> connectorInfo = createCallback.get(WAIT_TIME_MS, TimeUnit.MILLISECONDS);
        createCallback.get(WAIT_TIME_MS, TimeUnit.MILLISECONDS);
        assertEquals(createdInfo(SourceSink.SOURCE), connectorInfo.result());

        FutureCallback<Void> restartTaskCallback = new FutureCallback<>();
        herder.restartTask(taskId, restartTaskCallback);
        restartTaskCallback.get(WAIT_TIME_MS, TimeUnit.MILLISECONDS);
    }