KAFKA-7503: Connect integration test harness

build.gradle

@@ -1453,6 +1453,8 @@ project(':connect:runtime') {
     testCompile libs.powermockEasymock
 
     testCompile project(':clients').sourceSets.test.output
+    testCompile project(':core')
+    testCompile project(':core').sourceSets.test.output
 
     testRuntime libs.slf4jlog4j
   }

checkstyle/import-control.xml

@@ -347,8 +347,6 @@
       </subpackage>
     </subpackage>
 
-
-
     <subpackage name="cli">
       <allow pkg="org.apache.kafka.connect.runtime" />
       <allow pkg="org.apache.kafka.connect.storage" />
@@ -366,6 +364,18 @@
       <allow pkg="org.reflections.vfs" />
       <!-- for annotations to avoid code duplication -->
       <allow pkg="com.fasterxml.jackson.annotation" />
+      <allow pkg="com.fasterxml.jackson.databind" />
+      <subpackage name="clusters">
+        <allow pkg="kafka.server" />
+        <allow pkg="kafka.zk" />
+        <allow pkg="kafka.utils" />
+        <allow class="javax.servlet.http.HttpServletResponse" />
+      </subpackage>
+    </subpackage>
+
+    <subpackage name="integration">
+      <allow pkg="org.apache.kafka.connect.util.clusters" />
+      <allow pkg="org.apache.kafka.connect" />
     </subpackage>
 
     <subpackage name="json">

connect/runtime/src/main/java/org/apache/kafka/connect/cli/ConnectDistributed.java

@@ -20,6 +20,7 @@
 import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.common.utils.Utils;
 import org.apache.kafka.connect.runtime.Connect;
+import org.apache.kafka.connect.runtime.HerderProvider;
 import org.apache.kafka.connect.runtime.Worker;
 import org.apache.kafka.connect.runtime.WorkerConfigTransformer;
 import org.apache.kafka.connect.runtime.WorkerInfo;
@@ -54,62 +55,26 @@
 public class ConnectDistributed {
     private static final Logger log = LoggerFactory.getLogger(ConnectDistributed.class);
 
+    private final Time time = Time.SYSTEM;
+    private final long initStart = time.hiResClockMs();
+
     public static void main(String[] args) {
+
         if (args.length < 1 || Arrays.asList(args).contains("--help")) {
             log.info("Usage: ConnectDistributed worker.properties");
             Exit.exit(1);
         }
 
         try {
-            Time time = Time.SYSTEM;
-            log.info("Kafka Connect distributed worker initializing ...");
-            long initStart = time.hiResClockMs();
             WorkerInfo initInfo = new WorkerInfo();
             initInfo.logAll();
 
             String workerPropsFile = args[0];
             Map<String, String> workerProps = !workerPropsFile.isEmpty() ?
-                    Utils.propsToStringMap(Utils.loadProps(workerPropsFile)) : Collections.<String, String>emptyMap();
-
-            log.info("Scanning for plugin classes. This might take a moment ...");
-            Plugins plugins = new Plugins(workerProps);
-            plugins.compareAndSwapWithDelegatingLoader();
-            DistributedConfig config = new DistributedConfig(workerProps);
-
-            String kafkaClusterId = ConnectUtils.lookupKafkaClusterId(config);
-            log.debug("Kafka cluster ID: {}", kafkaClusterId);
-
-            RestServer rest = new RestServer(config);
-            URI advertisedUrl = rest.advertisedUrl();
-            String workerId = advertisedUrl.getHost() + ":" + advertisedUrl.getPort();
-
-            KafkaOffsetBackingStore offsetBackingStore = new KafkaOffsetBackingStore();
-            offsetBackingStore.configure(config);
-
-            Worker worker = new Worker(workerId, time, plugins, config, offsetBackingStore);
-            WorkerConfigTransformer configTransformer = worker.configTransformer();
-
-            Converter internalValueConverter = worker.getInternalValueConverter();
-            StatusBackingStore statusBackingStore = new KafkaStatusBackingStore(time, internalValueConverter);
-            statusBackingStore.configure(config);
-
-            ConfigBackingStore configBackingStore = new KafkaConfigBackingStore(
-                    internalValueConverter,
-                    config,
-                    configTransformer);
-
-            DistributedHerder herder = new DistributedHerder(config, time, worker,
-                    kafkaClusterId, statusBackingStore, configBackingStore,
-                    advertisedUrl.toString());
-            final Connect connect = new Connect(herder, rest);
-            log.info("Kafka Connect distributed worker initialization took {}ms", time.hiResClockMs() - initStart);
-            try {
-                connect.start();
-            } catch (Exception e) {
-                log.error("Failed to start Connect", e);
-                connect.stop();
-                Exit.exit(3);
-            }
+                    Utils.propsToStringMap(Utils.loadProps(workerPropsFile)) : Collections.emptyMap();
+
+            ConnectDistributed connectDistributed = new ConnectDistributed();
+            Connect connect = connectDistributed.startConnect(workerProps);
 
             // Shutdown will be triggered by Ctrl-C or via HTTP shutdown request
             connect.awaitStop();
@@ -119,4 +84,55 @@ public static void main(String[] args) {
             Exit.exit(2);
         }
     }
+
+    public Connect startConnect(Map<String, String> workerProps) {
+        log.info("Scanning for plugin classes. This might take a moment ...");
+        Plugins plugins = new Plugins(workerProps);
+        plugins.compareAndSwapWithDelegatingLoader();
+        DistributedConfig config = new DistributedConfig(workerProps);
+
+        String kafkaClusterId = ConnectUtils.lookupKafkaClusterId(config);
+        log.debug("Kafka cluster ID: {}", kafkaClusterId);
+
+        RestServer rest = new RestServer(config);
+        HerderProvider provider = new HerderProvider();
+        rest.start(provider, plugins);
+
+        URI advertisedUrl = rest.advertisedUrl();
+        String workerId = advertisedUrl.getHost() + ":" + advertisedUrl.getPort();
+
+        KafkaOffsetBackingStore offsetBackingStore = new KafkaOffsetBackingStore();
+        offsetBackingStore.configure(config);
+
+        Worker worker = new Worker(workerId, time, plugins, config, offsetBackingStore);
+        WorkerConfigTransformer configTransformer = worker.configTransformer();
+
+        Converter internalValueConverter = worker.getInternalValueConverter();
+        StatusBackingStore statusBackingStore = new KafkaStatusBackingStore(time, internalValueConverter);
+        statusBackingStore.configure(config);
+
+        ConfigBackingStore configBackingStore = new KafkaConfigBackingStore(
+                internalValueConverter,
+                config,
+                configTransformer);
+
+        DistributedHerder herder = new DistributedHerder(config, time, worker,
+                kafkaClusterId, statusBackingStore, configBackingStore,
+                advertisedUrl.toString());
+
+        final Connect connect = new Connect(herder, rest);
+        log.info("Kafka Connect distributed worker initialization took {}ms", time.hiResClockMs() - initStart);
+        try {
+            connect.start();
+            // herder has initialized now, and ready to be used by the RestServer.
+            provider.setHerder(herder);
+        } catch (Exception e) {
+            log.error("Failed to start Connect", e);
+            connect.stop();
+            Exit.exit(3);
+        }
+
+        return connect;
+    }
+
 }

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Connect.java

@@ -20,6 +20,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.net.URI;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicBoolean;
 
@@ -50,7 +51,6 @@ public void start() {
             Runtime.getRuntime().addShutdownHook(shutdownHook);
 
             herder.start();
-            rest.start(herder);
 
             log.info("Kafka Connect started");
         } finally {
@@ -82,6 +82,11 @@ public void awaitStop() {
         }
     }
 
+    // Visible for testing
+    public URI restUrl() {
+        return rest.serverUrl();
+    }
+
     private class ShutdownHook extends Thread {
         @Override
         public void run() {

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/HerderProvider.java

@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.connect.runtime;
+
+import org.apache.kafka.connect.errors.ConnectException;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * A supplier for {@link Herder}s.
+ */
+public class HerderProvider {
+
+    private final CountDownLatch initialized = new CountDownLatch(1);
+    volatile Herder herder = null;
+
+    public HerderProvider() {
+    }
+
+    /**
+     * Create a herder provider with a herder.
+     * @param herder the herder that will be supplied to threads waiting on this provider
+     */
+    public HerderProvider(Herder herder) {
+        this.herder = herder;
+        initialized.countDown();
+    }
+
+    /**
+     * @return the contained herder.
+     * @throws ConnectException if a herder was not available within a duration of calling this method
+     */
+    public Herder get() {
+        try {
+            // wait for herder to be initialized
+            if (!initialized.await(1, TimeUnit.MINUTES)) {
+                throw new ConnectException("Timed out waiting for herder to be initialized.");
+            }
+        } catch (InterruptedException e) {
+            throw new ConnectException("Interrupted while waiting for herder to be initialized.", e);
+        }
+        return herder;
+    }
+
+    /**
+     * @param herder set a herder, and signal to all threads waiting on get().
+     */
+    public void setHerder(Herder herder) {
+        this.herder = herder;
+        initialized.countDown();
+    }
+
+}

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/health/ConnectClusterStateImpl.java

@@ -22,7 +22,7 @@
 import org.apache.kafka.connect.health.ConnectorState;
 import org.apache.kafka.connect.health.ConnectorType;
 import org.apache.kafka.connect.health.TaskState;
-import org.apache.kafka.connect.runtime.Herder;
+import org.apache.kafka.connect.runtime.HerderProvider;
 import org.apache.kafka.connect.runtime.rest.entities.ConnectorStateInfo;
 import org.apache.kafka.connect.util.Callback;
 
@@ -34,16 +34,16 @@
 
 public class ConnectClusterStateImpl implements ConnectClusterState {
 
-    private Herder herder;
+    private HerderProvider herderProvider;
 
-    public ConnectClusterStateImpl(Herder herder) {
-        this.herder = herder;
+    public ConnectClusterStateImpl(HerderProvider herderProvider) {
+        this.herderProvider = herderProvider;
     }
 
     @Override
     public Collection<String> connectors() {
         final Collection<String> connectors = new ArrayList<>();
-        herder.connectors(new Callback<java.util.Collection<String>>() {
+        herderProvider.get().connectors(new Callback<java.util.Collection<String>>() {
             @Override
             public void onCompletion(Throwable error, Collection<String> result) {
                 connectors.addAll(result);
@@ -55,7 +55,7 @@ public void onCompletion(Throwable error, Collection<String> result) {
     @Override
     public ConnectorHealth connectorHealth(String connName) {
 
-        ConnectorStateInfo state = herder.connectorStatus(connName);
+        ConnectorStateInfo state = herderProvider.get().connectorStatus(connName);
         ConnectorState connectorState = new ConnectorState(
             state.connector().state(),
             state.connector().workerId(),