@Test
    public void testAlterSourceConnectorOffsetsInvalidRequestBody() throws Exception {
        // Create a source connector and stop it
        connect.configureConnector(connectorName, baseSourceConnectorConfigs());
        connect.assertions().assertConnectorAndAtLeastNumTasksAreRunning(connectorName, NUM_TASKS,
                "Connector tasks did not start in time.");
        connect.stopConnector(connectorName);
        connect.assertions().assertConnectorIsStopped(
                connectorName,
                "Connector did not stop in time"
        );
        String url = connect.endpointForResource(String.format("connectors/%s/offsets", connectorName));

        String content = "[]";
        try (Response response = connect.requestPatch(url, content)) {
            assertEquals(500, response.getStatus());
            assertThat(response.getEntity().toString(), containsString("Cannot deserialize value"));
        }

        content = "{}";
        try (Response response = connect.requestPatch(url, content)) {
            assertEquals(400, response.getStatus());
            assertThat(response.getEntity().toString(), containsString("Partitions / offsets need to be provided for an alter offsets request"));
        }

        content = "{\"key\": []}";
        try (Response response = connect.requestPatch(url, content)) {
            assertEquals(500, response.getStatus());
            assertThat(response.getEntity().toString(), containsString("Unrecognized field"));
        }

        content = "{\"offsets\": []}";
        try (Response response = connect.requestPatch(url, content)) {
            assertEquals(400, response.getStatus());
            assertThat(response.getEntity().toString(), containsString("Partitions / offsets need to be provided for an alter offsets request"));
        }

        content = "{\"offsets\": {}}";
        try (Response response = connect.requestPatch(url, content)) {
            assertEquals(500, response.getStatus());
            assertThat(response.getEntity().toString(), containsString("Cannot deserialize value"));
        }

        content = "{\"offsets\": [123]}";
        try (Response response = connect.requestPatch(url, content)) {
            assertEquals(500, response.getStatus());
            assertThat(response.getEntity().toString(), containsString("Cannot construct instance"));
        }

        content = "{\"offsets\": [{\"key\": \"val\"}]}";
        try (Response response = connect.requestPatch(url, content)) {
            assertEquals(500, response.getStatus());
            assertThat(response.getEntity().toString(), containsString("Unrecognized field"));
        }

        content = "{\"offsets\": [{\"partition\": []]}]}";
        try (Response response = connect.requestPatch(url, content)) {
            assertEquals(500, response.getStatus());
            assertThat(response.getEntity().toString(), containsString("Cannot deserialize value"));
        }
    }
Integration tests for Kafka Connect's connector offset management REST APIs
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java