### Question:
KafkaResource { @POST @Path("/clusters") @Consumes(MediaType.APPLICATION_JSON) @Produces(MediaType.APPLICATION_JSON) public KafkaCluster addKafkaCluster(KafkaCluster kafkaCluster) { return kafkaService.addCluster(kafkaCluster); } @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 addKafkaCluster() { final KafkaCluster requestCluster = new KafkaCluster(); requestCluster.setId("kafkaCluster-1"); requestCluster.setName("Kafka Cluster 1"); requestCluster.setZookeeperUri("zookeeper: when(kafkaServiceMock.addCluster(any(KafkaCluster.class))).then(new Answer<KafkaCluster>() { @Override public KafkaCluster answer(InvocationOnMock invocation) { KafkaCluster argCluster = (KafkaCluster) invocation.getArguments()[0]; assertEquals("Service kafka cluster should equal the requested kafka cluster", requestCluster, argCluster); return argCluster; } }); KafkaCluster responseCluster = resource().path("/api/kafka/clusters") .accept(MediaType.APPLICATION_JSON).type(MediaType.APPLICATION_JSON) .post(KafkaCluster.class, requestCluster); assertEquals("Response kafka cluster should match the request kafka cluster", requestCluster, responseCluster); verify(kafkaServiceMock).addCluster(any(KafkaCluster.class)); }