@DELETE
    @Path("/{connector}/offsets")
    @Operation(summary = "Reset the offsets for the specified connector")
    public Response resetConnectorOffsets(final @Parameter(hidden = true) @QueryParam("forward") Boolean forward,
                                          final @Context HttpHeaders headers, final @PathParam("connector") String connector) throws Throwable {
        FutureCallback<Message> cb = new FutureCallback<>();
        herder.resetConnectorOffsets(connector, cb);
        Message msg = requestHandler.completeOrForwardRequest(cb, "/connectors/" + connector + "/offsets", "DELETE", headers, null,
                new TypeReference<Message>() { }, new IdentityTranslator<>(), forward);
        return Response.ok().entity(msg).build();
    }
@Test
    public void testResetOffsets() throws Throwable {
        final ArgumentCaptor<Callback<Message>> cb = ArgumentCaptor.forClass(Callback.class);
        Message msg = new Message("The offsets for this connector have been reset successfully");
        doAnswer(invocation -> {
            cb.getValue().onCompletion(null, msg);
            return null;
        }).when(herder).resetConnectorOffsets(eq(CONNECTOR_NAME), cb.capture());
        Response response = connectorsResource.resetConnectorOffsets(null, NULL_HEADERS, CONNECTOR_NAME);
        assertEquals(200, response.getStatus());
        assertEquals(msg, response.getEntity());
    }