### Question:
KafkaResource { @DELETE @Path("/clusters/{clusterId}/topics/{topicName}") public Response deleteKafkaTopic(@PathParam("clusterId") String clusterId, @PathParam("topicName") String topicName) { try { kafkaService.deleteTopic(clusterId, topicName); return Response.ok().build(); } catch (Exception ex) { return Response.serverError().entity("Exception occured while deleting 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 deleteKafkaTopic() { final String requestClusterId = "CLUSTER_TO_GET"; final String requestTopicName = "TOPIC_TO_DELETE"; doAnswer(new Answer<Void>() { @Override public Void answer(InvocationOnMock invocation) throws Throwable { String argClusterId = (String) invocation.getArguments()[0]; String argTopicName = (String) invocation.getArguments()[1]; assertEquals("Service kafka cluster id should match the requested kafka cluster id", argClusterId, requestClusterId); assertEquals("Service kafka topic name should match the requested kafka topic name", argTopicName, requestTopicName); return null; } }).when(kafkaServiceMock).deleteTopic(anyString(), anyString()); ClientResponse clientResponse = resource() .path("/api/kafka/clusters/" + requestClusterId + "/topics/" + requestTopicName) .delete(ClientResponse.class); assertEquals("Response HTTP status code should be 200 (OK)", 200, clientResponse.getStatus()); verify(kafkaServiceMock).deleteTopic(anyString(), anyString()); }