Merge branch '7.0.x'

Author: dimitarndimitrov

kafka-rest/src/test/java/io/confluent/kafkarest/controllers/TopicManagerImplTest.java

@@ -19,8 +19,8 @@
 import static java.util.Collections.singletonList;
 import static java.util.Collections.singletonMap;
 import static java.util.concurrent.CompletableFuture.completedFuture;
-import static org.easymock.EasyMock.anyObject;
 import static org.easymock.EasyMock.expect;
+import static org.easymock.EasyMock.isA;
 import static org.easymock.EasyMock.replay;
 import static org.easymock.EasyMock.verify;
 import static org.junit.Assert.assertEquals;
@@ -36,6 +36,7 @@
 import io.confluent.kafkarest.entities.Topic;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -411,7 +412,7 @@ public void listTopics_existingCluster_returnsTopics() throws Exception {
     expect(clusterManager.getCluster(CLUSTER_ID)).andReturn(completedFuture(Optional.of(CLUSTER)));
     expect(adminClient.listTopics()).andReturn(listTopicsResult);
     expect(listTopicsResult.listings()).andReturn(KafkaFuture.completedFuture(TOPIC_LISTINGS));
-    expect(adminClient.describeTopics(anyObject())).andReturn(describeTopicResult);
+    expect(adminClient.describeTopics(isA(Collection.class))).andReturn(describeTopicResult);
     expect(describeTopicResult.all())
         .andReturn(
             KafkaFuture.completedFuture(
@@ -444,7 +445,7 @@ public void listLocalTopics_returnsTopics() throws Exception {
     expect(clusterManager.getLocalCluster()).andReturn(completedFuture(CLUSTER));
     expect(adminClient.listTopics()).andReturn(listTopicsResult);
     expect(listTopicsResult.listings()).andReturn(KafkaFuture.completedFuture(TOPIC_LISTINGS));
-    expect(adminClient.describeTopics(anyObject())).andReturn(describeTopicResult);
+    expect(adminClient.describeTopics(isA(Collection.class))).andReturn(describeTopicResult);
     expect(describeTopicResult.all())
         .andReturn(
             KafkaFuture.completedFuture(
@@ -473,7 +474,7 @@ public void listTopic_nonExistingCluster_throwsNotFoundException() throws Except
   @Test
   public void getTopic_existingTopic_returnsTopic() throws Exception {
     expect(clusterManager.getCluster(CLUSTER_ID)).andReturn(completedFuture(Optional.of(CLUSTER)));
-    expect(adminClient.describeTopics(anyObject())).andReturn(describeTopicResult);
+    expect(adminClient.describeTopics(isA(Collection.class))).andReturn(describeTopicResult);
     expect(describeTopicResult.all())
         .andReturn(KafkaFuture.completedFuture(createTopicDescriptionMap(TOPIC_DESCRIPTION_1)));
     replay(clusterManager, adminClient, describeTopicResult);
@@ -486,7 +487,7 @@ public void getTopic_existingTopic_returnsTopic() throws Exception {
   @Test
   public void getTopic_nonExistingCluster_throwsNotFoundException() throws Exception {
     expect(clusterManager.getCluster(CLUSTER_ID)).andReturn(completedFuture(Optional.empty()));
-    expect(adminClient.describeTopics(anyObject())).andReturn(describeTopicResult);
+    expect(adminClient.describeTopics(isA(Collection.class))).andReturn(describeTopicResult);
     replay(clusterManager);
 
     try {
@@ -500,7 +501,7 @@ public void getTopic_nonExistingCluster_throwsNotFoundException() throws Excepti
   @Test
   public void getTopic_nonExistingTopic_returnsEmpty() throws Exception {
     expect(clusterManager.getCluster(CLUSTER_ID)).andReturn(completedFuture(Optional.of(CLUSTER)));
-    expect(adminClient.describeTopics(anyObject())).andReturn(describeTopicResult);
+    expect(adminClient.describeTopics(isA(Collection.class))).andReturn(describeTopicResult);
     expect(describeTopicResult.all()).andReturn(KafkaFuture.completedFuture(new HashMap<>()));
     replay(clusterManager, adminClient, describeTopicResult);
 
@@ -512,7 +513,7 @@ public void getTopic_nonExistingTopic_returnsEmpty() throws Exception {
   @Test
   public void getLocalTopic_existingTopic_returnsTopic() throws Exception {
     expect(clusterManager.getLocalCluster()).andReturn(completedFuture(CLUSTER));
-    expect(adminClient.describeTopics(anyObject())).andReturn(describeTopicResult);
+    expect(adminClient.describeTopics(isA(Collection.class))).andReturn(describeTopicResult);
     expect(describeTopicResult.all())
         .andReturn(KafkaFuture.completedFuture(createTopicDescriptionMap(TOPIC_DESCRIPTION_1)));
     replay(clusterManager, adminClient, describeTopicResult);
@@ -525,7 +526,7 @@ public void getLocalTopic_existingTopic_returnsTopic() throws Exception {
   @Test
   public void getLocalTopic_nonExistingTopic_returnsEmpty() throws Exception {
     expect(clusterManager.getLocalCluster()).andReturn(completedFuture(CLUSTER));
-    expect(adminClient.describeTopics(anyObject())).andReturn(describeTopicResult);
+    expect(adminClient.describeTopics(isA(Collection.class))).andReturn(describeTopicResult);
     expect(describeTopicResult.all()).andReturn(KafkaFuture.completedFuture(new HashMap<>()));
     replay(clusterManager, adminClient, describeTopicResult);
 

kafka-rest/src/test/java/io/confluent/kafkarest/integration/ClusterTestHarness.java

@@ -164,6 +164,7 @@ public Properties overrideSchemaRegistryProps(Properties props) {
   @Before
   public void setUp() throws Exception {
     zookeeper = new EmbeddedZookeeper();
+    zkConnect = String.format("127.0.0.1:%d", zookeeper.port());
     // start brokers concurrently
     startBrokersConcurrently(numBrokers);