Fix compile

Remove unnecessary endpoints

checkstyle/suppressions.xml

@@ -14,7 +14,7 @@
     <!-- TODO after add offset monitoring -->
     <suppress
             checks="(ClassDataAbstractionCoupling)"
-            files="(AdminClientWrapper|MetadataObserver|GroupMetadataObserver).java"
+            files="(AdminClientWrapper|GroupMetadataObserver).java"
     />
 
     <suppress

kafka-rest-scala-consumer/src/main/java/io/confluent/kafkarest/MetadataObserver.java

@@ -15,14 +15,10 @@
 
 package io.confluent.kafkarest;
 
-import io.confluent.kafkarest.entities.EndPointEntity;
-import io.confluent.kafkarest.entities.BrokerEntity;
-import kafka.cluster.EndPoint;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.Arrays;
-import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
 import java.util.Properties;
@@ -40,7 +36,6 @@
 import kafka.utils.ZkUtils;
 import scala.Option;
 import scala.collection.JavaConversions;
-import scala.collection.JavaConverters;
 import scala.collection.Map;
 import scala.collection.Seq;
 import scala.math.Ordering;
@@ -83,29 +78,6 @@ public boolean topicExists(String topicName) {
     return false;
   }
 
-  /**
-   * <p>Get broker metadata</p>
-   *
-   * @param brokerId - broker ID
-   * @return metadata about broker by ID or null if broker not found
-   */
-  public BrokerEntity getBroker(final int brokerId) {
-    try {
-      Broker broker = getBrokerById(brokerId);
-      List<EndPointEntity> endpoints = new ArrayList<>();
-      for (EndPoint endPoint :
-              JavaConverters.seqAsJavaListConverter(broker.endPoints()).asJava()) {
-        endpoints.add(new EndPointEntity(endPoint.securityProtocol().name,
-                endPoint.host(),
-                endPoint.port()));
-      }
-      return new BrokerEntity(broker.id(), endpoints);
-    } catch (RestNotFoundException e) {
-      log.warn("Broker with id = {} not found. Return empty broker information.", brokerId);
-      return BrokerEntity.empty();
-    }
-  }
-
   private Collection<String> getTopicNames() {
     Seq<String> topicNames = zkUtils.getAllTopics().sorted(Ordering.String$.MODULE$);
     return JavaConversions.asJavaCollection(topicNames);

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

@@ -33,53 +33,28 @@
 import java.util.Vector;
 import java.util.concurrent.TimeUnit;
 
-import io.confluent.kafkarest.entities.NodeState;
 import io.confluent.kafkarest.entities.Partition;
 import io.confluent.kafkarest.entities.PartitionReplica;
 import io.confluent.kafkarest.entities.Topic;
 
 public class AdminClientWrapper {
 
-  private AdminClient adminClient;
-  private int initTimeOut;
+  private final AdminClient adminClient;
+  private final int initTimeOut;
 
   public AdminClientWrapper(KafkaRestConfig kafkaRestConfig, AdminClient adminClient) {
     this.adminClient = adminClient;
     this.initTimeOut = kafkaRestConfig.getInt(KafkaRestConfig.KAFKACLIENT_INIT_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,
         RestConfigUtils.bootstrapBrokers(kafkaRestConfig));
     return properties;
   }
 
-  /**
-   * <p>Check if broker is available</p>
-   *
-   * @param brokerId - broker ID for check
-   * @return true if brokerInfo by ID not null
-   */
-  public NodeState getBrokerState(Integer brokerId) {
-    DescribeClusterResult clusterResults = adminClient.describeCluster();
-    try {
-      Collection<Node> nodeCollection =
-          clusterResults.nodes().get(initTimeOut, TimeUnit.MILLISECONDS);
-      for (Node eachNode: nodeCollection) {
-        if (brokerId != null && brokerId.equals(eachNode.id())) {
-          return new NodeState(true);
-        }
-      }
-      return new NodeState(false);
-    } catch (InterruptedException | ExecutionException | TimeoutException e) {
-      throw new RestServerErrorException(Errors.KAFKA_ERROR_MESSAGE,
-              Errors.KAFKA_ERROR_ERROR_CODE, e
-      );
-    }
-  }
-
   public List<Integer> getBrokerIds() throws Exception {
     List<Integer> brokerIds = new Vector<>();
     DescribeClusterResult clusterResults = adminClient.describeCluster();
@@ -92,31 +67,27 @@ 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 = buildPartitionsData(topicDescription.partitions(), null);
-    return partitions;
+    return buildPartitionsData(topicDescription.partitions(), null);
   }
 
   public Partition getTopicPartition(String topicName, int partition) throws Exception {
@@ -138,14 +109,13 @@ private Topic buildTopic(String topicName, TopicDescription topicDescription) th
 
     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> buildPartitionsData(
@@ -178,7 +148,7 @@ private List<Partition> buildPartitionsData(
   }
 
   private TopicDescription getTopicDescription(String topicName) throws Exception {
-    return adminClient.describeTopics(Collections.unmodifiableList(Arrays.asList(topicName)))
+    return adminClient.describeTopics(Collections.unmodifiableList(Collections.singletonList(topicName)))
         .values().get(topicName).get(initTimeOut, TimeUnit.MILLISECONDS);
   }
 

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

@@ -30,7 +30,6 @@ public class DefaultKafkaRestContext implements KafkaRestContext {
   private ProducerPool producerPool;
   private KafkaConsumerManager kafkaConsumerManager;
   private AdminClientWrapper adminClientWrapper;
-  private final ClusterInformationObserver clusterInformationObserver;
   private final GroupMetadataObserver groupMetadataObserver;
 
 
@@ -39,7 +38,6 @@ public DefaultKafkaRestContext(
       ProducerPool producerPool,
       KafkaConsumerManager kafkaConsumerManager,
       AdminClientWrapper adminClientWrapper,
-      ClusterInformationObserver clusterInformationObserver,
       GroupMetadataObserver groupMetadataObserver,
       ScalaConsumersContext scalaConsumersContext
   ) {
@@ -48,7 +46,6 @@ public DefaultKafkaRestContext(
     this.producerPool = producerPool;
     this.kafkaConsumerManager = kafkaConsumerManager;
     this.adminClientWrapper = adminClientWrapper;
-    this.clusterInformationObserver = clusterInformationObserver;
     this.groupMetadataObserver = groupMetadataObserver;
     this.scalaConsumersContext = scalaConsumersContext;
   }
@@ -99,11 +96,6 @@ public AdminClientWrapper getAdminClientWrapper() {
     return adminClientWrapper;
   }
 
-  @Override
-  public ClusterInformationObserver getClusterInformationObserver() {
-    return clusterInformationObserver;
-  }
-
   @Override
   public GroupMetadataObserver getGroupMetadataObserver() {
     return groupMetadataObserver;