@Override
    public void connectorOffsets(String connName, Callback<ConnectorOffsets> cb) {
        ClusterConfigState configSnapshot = configBackingStore.snapshot();
        try {
            if (!configSnapshot.contains(connName)) {
                cb.onCompletion(new NotFoundException("Connector " + connName + " not found"), null);
                return;
            }
            // The worker asynchronously processes the request and completes the passed callback when done
            worker.connectorOffsets(connName, configSnapshot.connectorConfig(connName), cb);
        } catch (Throwable t) {
            cb.onCompletion(t, null);
        }
    }
@Test
    public void testConnectorOffsets() throws Exception {
        ConnectorOffsets offsets = new ConnectorOffsets(Arrays.asList(
                new ConnectorOffset(Collections.singletonMap("partitionKey", "partitionValue"), Collections.singletonMap("offsetKey", "offsetValue")),
                new ConnectorOffset(Collections.singletonMap("partitionKey", "partitionValue2"), Collections.singletonMap("offsetKey", "offsetValue"))
        ));
        @SuppressWarnings("unchecked")
        ArgumentCaptor<Callback<ConnectorOffsets>> workerCallback = ArgumentCaptor.forClass(Callback.class);
        doAnswer(invocation -> {
            workerCallback.getValue().onCompletion(null, offsets);
            return null;
        }).when(worker).connectorOffsets(eq(CONN1), eq(CONN1_CONFIG), workerCallback.capture());
        AbstractHerder herder = testHerder();
        when(configStore.snapshot()).thenReturn(SNAPSHOT);

        FutureCallback<ConnectorOffsets> cb = new FutureCallback<>();
        herder.connectorOffsets(CONN1, cb);
        assertEquals(offsets, cb.get(1000, TimeUnit.MILLISECONDS));
    }