@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 testConnectorOffsetsConnectorNotFound() {
        when(configStore.snapshot()).thenReturn(SNAPSHOT);
        AbstractHerder herder = testHerder();
        FutureCallback<ConnectorOffsets> cb = new FutureCallback<>();
        herder.connectorOffsets("unknown-connector", cb);
        ExecutionException e = assertThrows(ExecutionException.class, () -> cb.get(1000, TimeUnit.MILLISECONDS));
        assertEquals(NotFoundException.class, e.getCause().getClass());
    }