@Override
    public void connectorOffsets(String connName, Callback<ConnectorOffsets> cb) {
        log.trace("Submitting offset fetch request for connector: {}", connName);
        // Handle this in the tick thread to avoid concurrent calls to the Worker
        addRequest(
                () -> {
                    super.connectorOffsets(connName, cb);
                    return null;
                },
                forwardErrorAndTickThreadStages(cb)
        );
    }
@Test
    public void testConnectorOffsets() throws Exception {
        when(member.memberId()).thenReturn("leader");
        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
        expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true);
        expectConfigRefreshAndSnapshot(SNAPSHOT);

        when(statusBackingStore.connectors()).thenReturn(Collections.emptySet());
        expectMemberPoll();

        herder.tick();

        when(configBackingStore.snapshot()).thenReturn(SNAPSHOT);
        ConnectorOffsets offsets = new ConnectorOffsets(Collections.singletonList(new ConnectorOffset(
                Collections.singletonMap("partitionKey", "partitionValue"),
                Collections.singletonMap("offsetKey", "offsetValue"))));

        ArgumentCaptor<Callback<ConnectorOffsets>> callbackCapture = ArgumentCaptor.forClass(Callback.class);
        doAnswer(invocation -> {
            callbackCapture.getValue().onCompletion(null, offsets);
            return null;
        }).when(worker).connectorOffsets(eq(CONN1), eq(CONN1_CONFIG), callbackCapture.capture());

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

        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
    }