@PATCH
    @Path("/{connector}/offsets")
    @Operation(summary = "Alter the offsets for the specified connector")
    public Response alterConnectorOffsets(final @Parameter(hidden = true) @QueryParam("forward") Boolean forward,
                                          final @Context HttpHeaders headers, final @PathParam("connector") String connector,
                                          final ConnectorOffsets offsets) throws Throwable {
        if (offsets.offsets() == null || offsets.offsets().isEmpty()) {
            throw new BadRequestException("Partitions / offsets need to be provided for an alter offsets request");
        }

        FutureCallback<Message> cb = new FutureCallback<>();
        herder.alterConnectorOffsets(connector, offsets.toMap(), cb);
        Message msg = requestHandler.completeOrForwardRequest(cb, "/connectors/" + connector + "/offsets", "PATCH", headers, offsets,
                new TypeReference<Message>() { }, new IdentityTranslator<>(), forward);
        return Response.ok().entity(msg).build();
    }
@Test
    public void testAlterOffsetsConnectorNotFound() {
        Map<String, ?> partition = new HashMap<>();
        Map<String, ?> offset = new HashMap<>();
        ConnectorOffset connectorOffset = new ConnectorOffset(partition, offset);
        ConnectorOffsets body = new ConnectorOffsets(Collections.singletonList(connectorOffset));
        final ArgumentCaptor<Callback<Message>> cb = ArgumentCaptor.forClass(Callback.class);
        expectAndCallbackException(cb, new NotFoundException("Connector not found"))
                .when(herder).alterConnectorOffsets(eq(CONNECTOR_NAME), eq(body.toMap()), cb.capture());

        assertThrows(NotFoundException.class, () -> connectorsResource.alterConnectorOffsets(null, NULL_HEADERS, CONNECTOR_NAME, body));
    }