Merge remote-tracking branch 'origin/7.0.x'

* origin/7.0.x:
  Remove ZooKeeperClient/KafkaZkClient usage from tests (#893)

Author: ijuma

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

@@ -44,7 +44,6 @@
 import org.junit.Before;
 import org.junit.Test;
 import scala.Option;
-import scala.collection.JavaConverters;
 
 public class AuthorizationErrorTest
     extends AbstractProducerTest<BinaryTopicProduceRequest, BinaryPartitionProduceRequest> {
@@ -71,8 +70,7 @@ public class AuthorizationErrorTest
   @Before
   public void setUp() throws Exception {
     super.setUp();
-    kafka.utils.TestUtils.createTopic(
-        zkClient, TOPIC_NAME, 1, 1, JavaConverters.asScalaBuffer(this.servers), new Properties());
+    createTopic(TOPIC_NAME, 1, (short) 1);
   }
 
   @Override

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

@@ -38,7 +38,6 @@
 import org.apache.avro.Schema;
 import org.junit.Before;
 import org.junit.Test;
-import scala.collection.JavaConverters;
 
 // This test is much lighter than the Binary one which exercises all variants. Since binary
 // covers most code paths well, this just tries to exercise Schema-specific parts.
@@ -112,14 +111,8 @@ public AvroProducerTest() {
   public void setUp() throws Exception {
     super.setUp();
     final int numPartitions = 3;
-    final int replicationFactor = 1;
-    kafka.utils.TestUtils.createTopic(
-        zkClient,
-        topicName,
-        numPartitions,
-        replicationFactor,
-        JavaConverters.asScalaBuffer(this.servers),
-        new Properties());
+    final short replicationFactor = 1;
+    createTopic(topicName, numPartitions, replicationFactor);
 
     deserializerProps = new Properties();
     deserializerProps.setProperty("schema.registry.url", schemaRegConnect);

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

@@ -54,8 +54,6 @@
 import kafka.utils.MockTime;
 import kafka.utils.TestUtils;
 import kafka.zk.EmbeddedZookeeper;
-import kafka.zk.KafkaZkClient;
-import kafka.zookeeper.ZooKeeperClient;
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.admin.AdminClientConfig;
 import org.apache.kafka.clients.admin.AlterConfigOp;
@@ -72,10 +70,8 @@
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.config.ConfigResource;
-import org.apache.kafka.common.security.JaasUtils;
 import org.apache.kafka.common.security.auth.SecurityProtocol;
 import org.apache.kafka.common.serialization.ByteArraySerializer;
-import org.apache.kafka.common.utils.Time;
 import org.eclipse.jetty.server.Server;
 import org.junit.After;
 import org.junit.Assert;
@@ -122,7 +118,6 @@ public static int choosePort() {
   // ZK Config
   protected String zkConnect;
   protected EmbeddedZookeeper zookeeper;
-  protected KafkaZkClient zkClient;
   protected int zkConnectionTimeout = 10000;
   protected int zkSessionTimeout = 6000;
 
@@ -169,21 +164,6 @@ public Properties overrideSchemaRegistryProps(Properties props) {
   @Before
   public void setUp() throws Exception {
     zookeeper = new EmbeddedZookeeper();
-    zkConnect = String.format("127.0.0.1:%d", zookeeper.port());
-    Time time = Time.SYSTEM;
-    zkClient =
-        new KafkaZkClient(
-            new ZooKeeperClient(
-                zkConnect,
-                zkSessionTimeout,
-                zkConnectionTimeout,
-                Integer.MAX_VALUE,
-                time,
-                "testMetricGroup",
-                "testMetricGroupType"),
-            JaasUtils.isZkSaslEnabled(),
-            time);
-
     // start brokers concurrently
     startBrokersConcurrently(numBrokers);
 
@@ -327,7 +307,6 @@ public void tearDown() throws Exception {
       CoreUtils.delete(server.config().logDirs());
     }
 
-    zkClient.close();
     zookeeper.shutdown();
   }
 

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

@@ -26,10 +26,8 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Properties;
 import org.junit.Before;
 import org.junit.Test;
-import scala.collection.JavaConverters;
 
 public class JsonProducerTest
     extends AbstractProducerTest<JsonTopicProduceRequest, JsonPartitionProduceRequest> {
@@ -41,14 +39,8 @@ public class JsonProducerTest
   public void setUp() throws Exception {
     super.setUp();
     final int numPartitions = 3;
-    final int replicationFactor = 1;
-    kafka.utils.TestUtils.createTopic(
-        zkClient,
-        topicName,
-        numPartitions,
-        replicationFactor,
-        JavaConverters.asScalaBuffer(this.servers),
-        new Properties());
+    final short replicationFactor = 1;
+    createTopic(topicName, numPartitions, replicationFactor);
   }
 
   @Override

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

@@ -36,7 +36,6 @@
 import javax.ws.rs.core.Response;
 import org.junit.Before;
 import org.junit.Test;
-import scala.collection.JavaConverters;
 
 /**
  * Tests metadata access against a real cluster. This isn't exhaustive since the unit tests cover
@@ -80,7 +79,7 @@ public class MetadataAPITest extends ClusterTestHarness {
     topic2Configs.setProperty("cleanup.policy", "delete");
   }
 
-  private static final int numReplicas = 2;
+  private static final short numReplicas = 2;
 
   public MetadataAPITest() {
     super(2, false);
@@ -90,20 +89,8 @@ public MetadataAPITest() {
   @Override
   public void setUp() throws Exception {
     super.setUp();
-    kafka.utils.TestUtils.createTopic(
-        zkClient,
-        topic1Name,
-        topic1Partitions.size(),
-        numReplicas,
-        JavaConverters.asScalaBuffer(this.servers),
-        new Properties());
-    kafka.utils.TestUtils.createTopic(
-        zkClient,
-        topic2Name,
-        topic2Partitions.size(),
-        numReplicas,
-        JavaConverters.asScalaBuffer(this.servers),
-        topic2Configs);
+    createTopic(topic1Name, topic1Partitions.size(), numReplicas);
+    createTopic(topic2Name, topic2Partitions.size(), numReplicas);
   }
 
   @Test

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

@@ -24,13 +24,11 @@
 import io.confluent.kafkarest.entities.v2.PartitionOffset;
 import java.util.Arrays;
 import java.util.List;
-import java.util.Properties;
 import javax.ws.rs.client.Entity;
 import javax.ws.rs.core.Response;
 import org.apache.kafka.common.serialization.ByteArrayDeserializer;
 import org.junit.Before;
 import org.junit.Test;
-import scala.collection.JavaConverters;
 
 public class ProducerTest
     extends AbstractProducerTest<BinaryTopicProduceRequest, BinaryPartitionProduceRequest> {
@@ -108,14 +106,8 @@ public class ProducerTest
   public void setUp() throws Exception {
     super.setUp();
     final int numPartitions = 3;
-    final int replicationFactor = 1;
-    kafka.utils.TestUtils.createTopic(
-        zkClient,
-        topicName,
-        numPartitions,
-        replicationFactor,
-        JavaConverters.asScalaBuffer(this.servers),
-        new Properties());
+    final short replicationFactor = 1;
+    createTopic(topicName, numPartitions, replicationFactor);
   }
 
   @Test