### Question:
KafkaAdminClient implements Closeable { public AdminClient.ConsumerGroupSummary getConsumerGroupSummary(String consumerGroup) { if (StringUtils.isBlank(consumerGroup)) throw new IllegalArgumentException("consumerGroup cannot be null, empty or blank"); try { return getAdminClient().describeConsumerGroup(consumerGroup, operationTimeout); } catch (KafkaException e) { throw new AdminOperationException("Unable to retrieve summary for consumer group: " + consumerGroup, e); } } KafkaAdminClient(Properties properties); Set<String> getTopics(); Set<TopicAndPartition> getPartitions(); Set<TopicAndPartition> getPartitions(String topic); Authorizer getAuthorizer(); Map<Resource, Set<Acl>> getAcls(); Map<Resource, Set<Acl>> getAcls(KafkaPrincipal principal); Set<Acl> getAcls(Resource resource); void addAcls(Set<Acl> acls, Resource resource); void removeAcls(Set<Acl> acls, Resource resource); void createTopic(String topic, int partitions, int replicationFactor); void createTopic(String topic, int partitions, int replicationFactor, Properties topicConfig); void deleteTopic(String topic); Properties getTopicConfig(String topic); void updateTopicConfig(String topic, Properties properties); int getTopicReplicationFactor(String topic); int getTopicPartitions(String topic); void addTopicPartitions(String topic, int partitions); AdminClient.ConsumerGroupSummary getConsumerGroupSummary(String consumerGroup); Collection<AdminClient.ConsumerSummary> getConsumerGroupSummaries(String consumerGroup); Map<TopicPartition, String> getConsumerGroupAssignments(String consumerGroup); @Override void close(); static final String ZOOKEEPER_SECURE; static final String DEFAULT_ZOOKEEPER_SECURE; static final String OPERATION_TIMEOUT_MS; static final String DEFAULT_OPERATION_TIMEOUT_MS; static final String OPERATION_SLEEP_MS; static final String DEFAULT_OPERATION_SLEEP_MS; }### Answer:
@Test (expected = IllegalArgumentException.class) public void getConsumerGroupSummary_nullConsumerGroup() { client.getConsumerGroupSummary(null); }
@Test (expected = IllegalArgumentException.class) public void getConsumerGroupSummary_emptyConsumerGroup() { client.getConsumerGroupSummary(""); }
@Test (expected = IllegalArgumentException.class) public void getConsumerGroupSummary_blankConsumerGroup() { client.getConsumerGroupSummary(" "); }
@Test public void getConsumerGroupSummary() { client.createTopic(testName.getMethodName(), 1, 1); Properties properties = new Properties(); properties.putAll(KafkaTests.getProps()); properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, BytesDeserializer.class.getName()); properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, BytesDeserializer.class.getName()); properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, testName.getMethodName()); properties.setProperty(ConsumerConfig.CLIENT_ID_CONFIG, testName.getMethodName() + "-client-id"); try (Consumer<Object, Object> consumer = new KafkaConsumer<>(properties)) { consumer.subscribe(Arrays.asList(testName.getMethodName())); consumer.poll(Duration.ofSeconds(5L)); AdminClient.ConsumerGroupSummary summary = client.getConsumerGroupSummary(testName.getMethodName()); assertThat("Expected only 1 consumer summary when getConsumerGroupSummaries(" + testName.getMethodName() + ")", convertToJavaSet(summary.consumers().get().iterator()).size(), is(1)); assertThat(summary.state(), is(notNullValue())); assertThat(summary.coordinator(), is(notNullValue())); assertThat(summary.assignmentStrategy(), is(notNullValue())); } }
@Test public void getConsumerGroupSummary_doesNotExist() { AdminClient.ConsumerGroupSummary summary = failureClient.getConsumerGroupSummary("consumer-group-does-not-exist"); assertThat(summary.state(), is(ConsumerGroupState.DEAD.toString())); }