Add feature to request consumer group information

checkstyle/suppressions.xml

@@ -18,7 +18,7 @@
 
     <suppress
             checks="(ClassDataAbstractionCoupling)"
-            files="(AvroConverter|ConsumerManager|KafkaConsumerManager|KafkaRestApplication|SimpleConsumerManager|KafkaRestContextProvider|KafkaConsumerState).java"
+            files="(AdminClientWrapper|AvroConverter|ConsumerManager|KafkaConsumerManager|KafkaRestApplication|SimpleConsumerManager|KafkaRestContextProvider|KafkaConsumerState).java"
     />
 
     <suppress

config/kafka-rest.properties

@@ -15,7 +15,6 @@
 
 #id=kafka-rest-test-server
 #schema.registry.url=http://localhost:8081
-#zookeeper.connect=localhost:2181
 bootstrap.servers=PLAINTEXT://localhost:9092
 #
 # Configure interceptor classes for sending consumer and producer metrics to Confluent Control Center

kafka-rest-common/src/main/java/io/confluent/kafkarest/KafkaRestConfig.java

@@ -196,6 +196,10 @@ public class KafkaRestConfig extends RestConfig {
    * <code>client.init.timeout.ms</code>
    */
   public static final String KAFKACLIENT_INIT_TIMEOUT_CONFIG = "client.init.timeout.ms";
+  /**
+   * <code>client.request.timeout.ms</code>
+   */
+  public static final String KAFKACLIENT_REQUEST_TIMEOUT_CONFIG = "client.request.timeout.ms";
 
   public static final String ZOOKEEPER_SET_ACL_CONFIG = "zookeeper.set.acl";
   public static final String KAFKACLIENT_SECURITY_PROTOCOL_CONFIG =
@@ -247,6 +251,9 @@ public class KafkaRestConfig extends RestConfig {
   protected static final String KAFKACLIENT_INIT_TIMEOUT_DOC =
       "The timeout for initialization of the Kafka store, including creation of the Kafka topic "
       + "that stores schema data.";
+  protected static final String KAFKACLIENT_REQUEST_TIMEOUT_DOC =
+      "The timeout for sending any admin-client request to Kafka cluster including waiting for"
+      + " the response on client side.";
   protected static final String KAFKACLIENT_TIMEOUT_DOC =
       "The timeout for an operation on the Kafka store";
   protected static final String
@@ -450,6 +457,14 @@ protected static ConfigDef baseKafkaRestConfigDef() {
         Importance.MEDIUM,
         KAFKACLIENT_INIT_TIMEOUT_DOC
     )
+    .define(
+        KAFKACLIENT_REQUEST_TIMEOUT_CONFIG,
+        Type.INT,
+        60000,
+        Range.atLeast(0),
+        Importance.MEDIUM,
+        KAFKACLIENT_REQUEST_TIMEOUT_DOC
+    )
     .define(
         KAFKACLIENT_TIMEOUT_CONFIG,
         Type.INT,

kafka-rest-common/src/main/java/io/confluent/kafkarest/entities/Topic.java

@@ -15,6 +15,7 @@
 
 package io.confluent.kafkarest.entities;
 
+import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.annotation.JsonProperty;
 
 import org.hibernate.validator.constraints.NotEmpty;
@@ -23,17 +24,14 @@
 import java.util.Objects;
 import java.util.Properties;
 
-import javax.validation.constraints.NotNull;
-
+@JsonInclude(JsonInclude.Include.NON_NULL)
 public class Topic {
 
   @NotEmpty
   private String name;
 
-  @NotNull
   private Properties configs;
 
-  @NotEmpty
   private List<Partition> partitions;
 
   public Topic(

kafka-rest/src/main/java/io/confluent/kafkarest/AdminClientWrapper.java

@@ -16,18 +16,22 @@
 package io.confluent.kafkarest;
 
 import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.ConsumerGroupListing;
+import org.apache.kafka.clients.admin.ConsumerGroupDescription;
 import org.apache.kafka.clients.admin.Config;
 import org.apache.kafka.clients.admin.ConfigEntry;
 import org.apache.kafka.clients.admin.DescribeClusterResult;
 import org.apache.kafka.clients.admin.TopicDescription;
+import org.apache.kafka.clients.admin.ListConsumerGroupsOptions;
+import org.apache.kafka.clients.admin.DescribeConsumerGroupsOptions;
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.TopicPartitionInfo;
 import org.apache.kafka.common.config.ConfigResource;
 
-import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.Properties;
 import java.util.TreeSet;
 import java.util.Vector;
@@ -39,15 +43,18 @@
 
 public class AdminClientWrapper {
 
-  private AdminClient adminClient;
-  private int initTimeOut;
+  private final AdminClient adminClient;
+  private final int initTimeOut;
+  private final int requestTimeOut;
 
   public AdminClientWrapper(KafkaRestConfig kafkaRestConfig, AdminClient adminClient) {
     this.adminClient = adminClient;
     this.initTimeOut = kafkaRestConfig.getInt(KafkaRestConfig.KAFKACLIENT_INIT_TIMEOUT_CONFIG);
+    this.requestTimeOut = kafkaRestConfig.getInt(
+        KafkaRestConfig.KAFKACLIENT_REQUEST_TIMEOUT_CONFIG);
   }
 
-  public static Properties adminProperties(KafkaRestConfig kafkaRestConfig) {
+  static Properties adminProperties(KafkaRestConfig kafkaRestConfig) {
     Properties properties = new Properties();
     properties.putAll(kafkaRestConfig.getAdminProperties());
     properties.put(KafkaRestConfig.BOOTSTRAP_SERVERS_CONFIG,
@@ -67,36 +74,32 @@ public List<Integer> getBrokerIds() throws Exception {
   }
 
   public Collection<String> getTopicNames() throws Exception {
-    Collection<String> allTopics = null;
-    allTopics = new TreeSet<>(
+    return new TreeSet<>(
         adminClient.listTopics().names().get(initTimeOut, TimeUnit.MILLISECONDS));
-    return allTopics;
   }
 
   public boolean topicExists(String topic) throws Exception {
-    Collection<String> allTopics = getTopicNames();
-    return allTopics.contains(topic);
+    return getTopicNames().contains(topic);
   }
 
   public Topic getTopic(String topicName) throws Exception {
-    Topic topic = null;
     if (topicExists(topicName)) {
       TopicDescription topicDescription = getTopicDescription(topicName);
 
-      topic = buildTopic(topicName, topicDescription);
+      return buildTopic(topicName, topicDescription);
+    } else {
+      return null;
     }
-    return topic;
   }
 
   public List<Partition> getTopicPartitions(String topicName) throws Exception {
     TopicDescription topicDescription = getTopicDescription(topicName);
-    List<Partition> partitions = buildPartitonsData(topicDescription.partitions(), null);
-    return partitions;
+    return buildPartitionsData(topicDescription.partitions(), null);
   }
 
   public Partition getTopicPartition(String topicName, int partition) throws Exception {
     TopicDescription topicDescription = getTopicDescription(topicName);
-    List<Partition> partitions = buildPartitonsData(topicDescription.partitions(), partition);
+    List<Partition> partitions = buildPartitionsData(topicDescription.partitions(), partition);
     if (partitions.isEmpty()) {
       return null;
     }
@@ -108,22 +111,33 @@ public boolean partitionExists(String topicName, int partition) throws Exception
     return (partition >= 0 && partition < topic.getPartitions().size());
   }
 
+  public Collection<ConsumerGroupListing> listConsumerGroups() throws Exception {
+    return adminClient.listConsumerGroups(new ListConsumerGroupsOptions()
+        .timeoutMs(requestTimeOut)).all().get(requestTimeOut, TimeUnit.MILLISECONDS);
+  }
+
+  public Map<String, ConsumerGroupDescription> describeConsumerGroups(
+          Collection<String> groupIds) throws Exception {
+    return adminClient.describeConsumerGroups(groupIds,
+        new DescribeConsumerGroupsOptions().timeoutMs(requestTimeOut))
+        .all().get(requestTimeOut, TimeUnit.MILLISECONDS);
+  }
+
   private Topic buildTopic(String topicName, TopicDescription topicDescription) throws Exception {
-    List<Partition> partitions = buildPartitonsData(topicDescription.partitions(), null);
+    List<Partition> partitions = buildPartitionsData(topicDescription.partitions(), null);
 
     ConfigResource topicResource = new ConfigResource(ConfigResource.Type.TOPIC, topicName);
     Config config = adminClient.describeConfigs(
-        Collections.unmodifiableList(Arrays.asList(topicResource))
+        Collections.unmodifiableList(Collections.singletonList(topicResource))
     ).values().get(topicResource).get();
     Properties topicProps = new Properties();
     for (ConfigEntry configEntry : config.entries()) {
       topicProps.put(configEntry.name(), configEntry.value());
     }
-    Topic topic = new Topic(topicName, topicProps, partitions);
-    return topic;
+    return new Topic(topicName, topicProps, partitions);
   }
 
-  private List<Partition> buildPartitonsData(
+  private List<Partition> buildPartitionsData(
       List<TopicPartitionInfo> partitions,
       Integer partitionsFilter
   ) {
@@ -153,7 +167,7 @@ private List<Partition> buildPartitonsData(
   }
 
   private TopicDescription getTopicDescription(String topicName) throws Exception {
-    return adminClient.describeTopics(Collections.unmodifiableList(Arrays.asList(topicName)))
+    return adminClient.describeTopics(Collections.singletonList(topicName))
         .values().get(topicName).get(initTimeOut, TimeUnit.MILLISECONDS);
   }
 

kafka-rest/src/main/java/io/confluent/kafkarest/DefaultKafkaRestContext.java

@@ -30,20 +30,23 @@ public class DefaultKafkaRestContext implements KafkaRestContext {
   private ProducerPool producerPool;
   private KafkaConsumerManager kafkaConsumerManager;
   private AdminClientWrapper adminClientWrapper;
+  private GroupMetadataObserver groupMetadataObserver;
 
 
   public DefaultKafkaRestContext(
       KafkaRestConfig config,
       ProducerPool producerPool,
       KafkaConsumerManager kafkaConsumerManager,
       AdminClientWrapper adminClientWrapper,
+      GroupMetadataObserver groupMetadataObserver,
       ScalaConsumersContext scalaConsumersContext
   ) {
 
     this.config = config;
     this.producerPool = producerPool;
     this.kafkaConsumerManager = kafkaConsumerManager;
     this.adminClientWrapper = adminClientWrapper;
+    this.groupMetadataObserver = groupMetadataObserver;
     this.scalaConsumersContext = scalaConsumersContext;
   }
 
@@ -54,7 +57,7 @@ public KafkaRestConfig getConfig() {
   }
 
   @Override
-  public ProducerPool getProducerPool() {
+  public synchronized ProducerPool getProducerPool() {
     if (producerPool == null) {
       producerPool = new ProducerPool(config);
     }
@@ -77,22 +80,30 @@ public SimpleConsumerManager getSimpleConsumerManager() {
   }
 
   @Override
-  public KafkaConsumerManager getKafkaConsumerManager() {
+  public synchronized KafkaConsumerManager getKafkaConsumerManager() {
     if (kafkaConsumerManager == null) {
       kafkaConsumerManager = new KafkaConsumerManager(config);
     }
     return kafkaConsumerManager;
   }
 
   @Override
-  public AdminClientWrapper getAdminClientWrapper() {
+  public synchronized AdminClientWrapper getAdminClientWrapper() {
     if (adminClientWrapper == null) {
       adminClientWrapper = new AdminClientWrapper(config,
           AdminClient.create(AdminClientWrapper.adminProperties(config)));
     }
     return adminClientWrapper;
   }
 
+  @Override
+  public synchronized GroupMetadataObserver getGroupMetadataObserver() {
+    if (groupMetadataObserver == null) {
+      groupMetadataObserver = new GroupMetadataObserver(config, getAdminClientWrapper());
+    }
+    return groupMetadataObserver;
+  }
+
   @Override
   public void shutdown() {
     if (kafkaConsumerManager != null) {