### Question:
KafkaMetricsSet implements MetricSet { public Boolean connectionToKafkaTopicsIsSuccess() { if (nonNull(metricTopics) && nonNull(connectionTimeoutTopic)) { StopWatch executionTime = StopWatch.createStarted(); DescribeTopicsOptions describeTopicsOptions = new DescribeTopicsOptions().timeoutMs( connectionTimeoutTopic); try (AdminClient adminClient = AdminClient.create(kafkaAdmin.getConfig())) { try { DescribeTopicsResult describeTopicsResult = adminClient.describeTopics( metricTopics, describeTopicsOptions); Map<String, TopicDescription> topicDescriptionMap = describeTopicsResult.all().get(); boolean monitoringResult = nonNull(topicDescriptionMap); log.info("Connection to Kafka topics is {}, time: {}", monitoringResult, executionTime.getTime()); return monitoringResult; } catch (Exception e) { log.warn("Exception when try connect to kafka topics: {}, exception: {}, time: {}", metricTopics, e.getMessage(), executionTime.getTime()); return false; } } } log.warn("metricTopics or connectionTimeoutTopic not found: {}, {}", metricTopics, connectionTimeoutTopic); return null; } KafkaMetricsSet(KafkaAdmin kafkaAdmin, Integer connectionTimeoutTopic, List<String> metricTopics); @Override Map<String, Metric> getMetrics(); Boolean connectionToKafkaTopicsIsSuccess();  }### Answer:
@Test public void connectionToKafkaTopicsIsSuccess() { KafkaMetricsSet kafkaMetricsSet = initKafkaMetricSet(); assertTrue(kafkaMetricsSet.connectionToKafkaTopicsIsSuccess()); }
@Test @SneakyThrows public void connectionToKafkaTopicsIsNotSuccess() { KafkaMetricsSet kafkaMetricsSet = initKafkaMetricSet(); kafkaEmbedded.destroy(); assertFalse(kafkaMetricsSet.connectionToKafkaTopicsIsSuccess()); }
@Test public void connectionToKafkaIsNotSuccessWithWrongTopic() { KafkaMetricsSet kafkaMetricsSet = initNotExistTopic(); assertFalse(kafkaMetricsSet.connectionToKafkaTopicsIsSuccess()); }