@Override
    protected synchronized void modifyConnectorOffsets(String connName, Map<Map<String, ?>, Map<String, ?>> offsets, Callback<Message> cb) {
        if (!modifyConnectorOffsetsChecks(connName, cb)) {
            return;
        }

        worker.modifyConnectorOffsets(connName, configState.connectorConfig(connName), offsets, cb);
    }
@Test
    public void testResetConnectorOffsets() throws Exception {
        initialize(false);
        ArgumentCaptor<Callback<Message>> workerCallbackCapture = ArgumentCaptor.forClass(Callback.class);
        Message msg = new Message("The offsets for this connector have been reset successfully");

        doAnswer(invocation -> {
            workerCallbackCapture.getValue().onCompletion(null, msg);
            return null;
        }).when(worker).modifyConnectorOffsets(eq(CONNECTOR_NAME), eq(connectorConfig(SourceSink.SOURCE)), isNull(), workerCallbackCapture.capture());

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

        herder.configState = new ClusterConfigState(
            10,
            null,
            Collections.singletonMap(CONNECTOR_NAME, 0),
            Collections.singletonMap(CONNECTOR_NAME, connectorConfig(SourceSink.SOURCE)),
            Collections.singletonMap(CONNECTOR_NAME, TargetState.STOPPED),
            Collections.emptyMap(),
            Collections.emptyMap(),
            Collections.emptyMap(),
            Collections.singletonMap(CONNECTOR_NAME, new AppliedConnectorConfig(connectorConfig)),
            Collections.emptySet(),
            Collections.emptySet()
        );
        FutureCallback<Message> resetOffsetsCallback = new FutureCallback<>();
        herder.resetConnectorOffsets(CONNECTOR_NAME, resetOffsetsCallback);
        assertEquals(msg, resetOffsetsCallback.get(1000, TimeUnit.MILLISECONDS));
    }