### Question:
KafkaResource { @GET @Path("/clusters/{clusterId}/topics") @Produces(MediaType.APPLICATION_JSON) public List<KafkaTopic> listKafkaTopics(@PathParam("clusterId") String clusterId) { return kafkaService.listTopics(clusterId); } @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 listKafkaTopics() { String requestClusterId = "CLUSTER_TO_GET"; KafkaTopic topic1 = new KafkaTopic(); topic1.setName("Kafka Topic 1"); topic1.setNumPartitions(5); topic1.setReplicationFactor(2); KafkaTopic topic2 = new KafkaTopic(); topic2.setName("Kafka Topic 2"); topic2.setNumPartitions(10); topic2.setReplicationFactor(3); List<KafkaTopic> mockedTopics = new ArrayList<KafkaTopic>(); mockedTopics.add(topic1); mockedTopics.add(topic2); when(kafkaServiceMock.listTopics(anyString())).thenReturn(mockedTopics); List<KafkaTopic> responseTopics = resource() .path("/api/kafka/clusters/" + requestClusterId + "/topics") .accept(MediaType.APPLICATION_JSON).get(new GenericType<List<KafkaTopic>>(){}); assertEquals("Response kafka topics should be equal to the mocked kafka topics", mockedTopics, responseTopics); verify(kafkaServiceMock).listTopics(anyString()); }