Handler getKRaftMigrationHandler() {
        return new AbstractHandler() {
            @Override
            public void handle(String s, Request baseRequest, HttpServletRequest request, HttpServletResponse response) throws IOException {
                response.setContentType("application/json");
                response.setCharacterEncoding("UTF-8");
                baseRequest.setHandled(true);

                if (zkMigrationState != null) {
                    Map<String, Object> migrationResponse = new HashMap<>();
                    migrationResponse.put("state", zkMigrationState.value());
                    response.setStatus(HttpServletResponse.SC_OK);
                    String json = new ObjectMapper().writeValueAsString(migrationResponse);
                    response.getWriter().print(json);
                } else {
                    response.setStatus(HttpServletResponse.SC_NOT_FOUND);
                    response.getWriter().print("ZooKeeper migration state metric not found");
                }
            }
        };
    }
@Test
    public void testZkMigrationMetricNotFound() throws Exception {
        KafkaAgent agent = new KafkaAgent(null, null, null, null);
        context.setHandler(agent.getKRaftMigrationHandler());
        server.setHandler(context);
        server.start();

        HttpResponse<String> response = HttpClient.newBuilder()
                .build()
                .send(req, HttpResponse.BodyHandlers.ofString());
        assertThat(HttpServletResponse.SC_NOT_FOUND, is(response.statusCode()));
    }