### Question:
KafkaResource { @POST @Path("/clusters/{clusterId}/topics") public Response addKafkaTopic(@PathParam("clusterId") String clusterId, KafkaTopic topic) { try { kafkaService.addTopic(clusterId, topic); return Response.ok().build(); } catch (Exception ex) { return Response.serverError().entity("Exception occured while adding the topic: " + ex.getMessage()).build(); } } @Inject  KafkaResource(KafkaService kafkaService); @GET @Path("/clusters") @Produces(MediaType.APPLICATION_JSON) List<KafkaCluster> listKafkaClusters(); @POST @Path("/clusters") @Consumes(MediaType.APPLICATION_JSON) @Produces(MediaType.APPLICATION_JSON) KafkaCluster addKafkaCluster(KafkaCluster kafkaCluster); @GET @Path("/clusters/{clusterId}") @Produces(MediaType.APPLICATION_JSON) KafkaCluster getKafkaCluster(@PathParam("clusterId") String clusterId); @DELETE @Path("/clusters/{clusterId}") Response removeKafkaCluster(@PathParam("clusterId") String clusterId); @GET @Path("/clusters/{clusterId}/topics") @Produces(MediaType.APPLICATION_JSON) List<KafkaTopic> listKafkaTopics(@PathParam("clusterId") String clusterId); @POST @Path("/clusters/{clusterId}/topics") Response addKafkaTopic(@PathParam("clusterId") String clusterId, KafkaTopic topic); @DELETE @Path("/clusters/{clusterId}/topics/{topicName}") Response deleteKafkaTopic(@PathParam("clusterId") String clusterId, 
            @PathParam("topicName") String topicName);  }### Answer:
@Test public void addKafkaTopic() { final String requestClusterId = "CLUSTER_TO_GET"; final KafkaTopic requestTopic = new KafkaTopic(); requestTopic.setName("Kafka Topic 1"); doAnswer(new Answer<Void>() { @Override public Void answer(InvocationOnMock invocation) throws Throwable { String argClusterId = (String) invocation.getArguments()[0]; KafkaTopic argTopic = (KafkaTopic) invocation.getArguments()[1]; assertEquals("Service kafka cluster ID should match the requested kafka cluster ID", argClusterId, requestClusterId); assertEquals("Service kafka topic should equal the requested kafka topic", requestTopic, argTopic); return null; } }).when(kafkaServiceMock).addTopic(anyString(), any(KafkaTopic.class)); ClientResponse clientResponse = resource() .path("/api/kafka/clusters/" + requestClusterId + "/topics") .type(MediaType.APPLICATION_JSON) .post(ClientResponse.class, requestTopic); assertEquals("Response HTTP status code should be 200 (OK)", 200, clientResponse.getStatus()); verify(kafkaServiceMock).addTopic(anyString(), any(KafkaTopic.class)); }