KAFKA-8304: Fix registration of Connect REST extensions

checkstyle/import-control.xml

@@ -382,6 +382,7 @@
       <allow pkg="org.apache.kafka.connect.util.clusters" />
       <allow pkg="org.apache.kafka.connect" />
       <allow pkg="org.apache.kafka.tools" />
+      <allow pkg="javax.ws.rs" />
     </subpackage>
 
     <subpackage name="json">

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

@@ -20,7 +20,6 @@
 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;
@@ -95,8 +94,7 @@ public Connect startConnect(Map<String, String> workerProps) {
         log.debug("Kafka cluster ID: {}", kafkaClusterId);
 
         RestServer rest = new RestServer(config);
-        HerderProvider provider = new HerderProvider();
-        rest.start(provider, plugins);
+        rest.initializeServer();
 
         URI advertisedUrl = rest.advertisedUrl();
         String workerId = advertisedUrl.getHost() + ":" + advertisedUrl.getPort();
@@ -124,8 +122,6 @@ public Connect startConnect(Map<String, String> workerProps) {
         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();

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

@@ -22,7 +22,6 @@
 import org.apache.kafka.connect.runtime.Connect;
 import org.apache.kafka.connect.runtime.ConnectorConfig;
 import org.apache.kafka.connect.runtime.Herder;
-import org.apache.kafka.connect.runtime.HerderProvider;
 import org.apache.kafka.connect.runtime.Worker;
 import org.apache.kafka.connect.runtime.WorkerInfo;
 import org.apache.kafka.connect.runtime.isolation.Plugins;
@@ -83,8 +82,7 @@ public static void main(String[] args) {
             log.debug("Kafka cluster ID: {}", kafkaClusterId);
 
             RestServer rest = new RestServer(config);
-            HerderProvider provider = new HerderProvider();
-            rest.start(provider, plugins);
+            rest.initializeServer();
 
             URI advertisedUrl = rest.advertisedUrl();
             String workerId = advertisedUrl.getHost() + ":" + advertisedUrl.getPort();
@@ -97,8 +95,6 @@ public static void main(String[] args) {
 
             try {
                 connect.start();
-                // herder has initialized now, and ready to be used by the RestServer.
-                provider.setHerder(herder);
                 for (final String connectorPropsFile : Arrays.copyOfRange(args, 1, args.length)) {
                     Map<String, String> connectorProps = Utils.propsToStringMap(Utils.loadProps(connectorPropsFile));
                     FutureCallback<Herder.Created<ConnectorInfo>> cb = new FutureCallback<>(new Callback<Herder.Created<ConnectorInfo>>() {

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

@@ -51,6 +51,7 @@ public void start() {
             Runtime.getRuntime().addShutdownHook(shutdownHook);
 
             herder.start();
+            rest.initializeResources(herder);
 
             log.info("Kafka Connect started");
         } finally {

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

@@ -23,7 +23,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.HerderProvider;
+import org.apache.kafka.connect.runtime.Herder;
 import org.apache.kafka.connect.runtime.rest.entities.ConnectorStateInfo;
 import org.apache.kafka.connect.util.FutureCallback;
 
@@ -38,17 +38,17 @@
 public class ConnectClusterStateImpl implements ConnectClusterState {
 
     private final long herderRequestTimeoutMs;
-    private final HerderProvider herderProvider;
+    private final Herder herder;
 
-    public ConnectClusterStateImpl(long connectorsTimeoutMs, HerderProvider herderProvider) {
+    public ConnectClusterStateImpl(long connectorsTimeoutMs, Herder herder) {
         this.herderRequestTimeoutMs = connectorsTimeoutMs;
-        this.herderProvider = herderProvider;
+        this.herder = herder;
     }
 
     @Override
     public Collection<String> connectors() {
         FutureCallback<Collection<String>> connectorsCallback = new FutureCallback<>();
-        herderProvider.get().connectors(connectorsCallback);
+        herder.connectors(connectorsCallback);
         try {
             return connectorsCallback.get(herderRequestTimeoutMs, TimeUnit.MILLISECONDS);
         } catch (InterruptedException | ExecutionException | TimeoutException e) {
@@ -59,7 +59,7 @@ public Collection<String> connectors() {
     @Override
     public ConnectorHealth connectorHealth(String connName) {
 
-        ConnectorStateInfo state = herderProvider.get().connectorStatus(connName);
+        ConnectorStateInfo state = herder.connectorStatus(connName);
         ConnectorState connectorState = new ConnectorState(
             state.connector().state(),
             state.connector().workerId(),

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/RestServer.java

@@ -21,10 +21,9 @@
 import org.apache.kafka.connect.errors.ConnectException;
 import org.apache.kafka.connect.rest.ConnectRestExtension;
 import org.apache.kafka.connect.rest.ConnectRestExtensionContext;
-import org.apache.kafka.connect.runtime.HerderProvider;
+import org.apache.kafka.connect.runtime.Herder;
 import org.apache.kafka.connect.runtime.WorkerConfig;
 import org.apache.kafka.connect.runtime.health.ConnectClusterStateImpl;
-import org.apache.kafka.connect.runtime.isolation.Plugins;
 import org.apache.kafka.connect.runtime.rest.errors.ConnectExceptionMapper;
 import org.apache.kafka.connect.runtime.rest.resources.ConnectorPluginsResource;
 import org.apache.kafka.connect.runtime.rest.resources.ConnectorsResource;
@@ -34,8 +33,8 @@
 import org.eclipse.jetty.server.Handler;
 import org.eclipse.jetty.server.Server;
 import org.eclipse.jetty.server.ServerConnector;
+import org.eclipse.jetty.server.handler.ContextHandlerCollection;
 import org.eclipse.jetty.server.handler.DefaultHandler;
-import org.eclipse.jetty.server.handler.HandlerCollection;
 import org.eclipse.jetty.server.handler.RequestLogHandler;
 import org.eclipse.jetty.server.handler.StatisticsHandler;
 import org.eclipse.jetty.servlet.FilterHolder;
@@ -74,6 +73,7 @@ public class RestServer {
     private static final String PROTOCOL_HTTPS = "https";
 
     private final WorkerConfig config;
+    private ContextHandlerCollection handlers;
     private Server jettyServer;
 
     private List<ConnectRestExtension> connectRestExtensions = Collections.emptyList();
@@ -87,6 +87,7 @@ public RestServer(WorkerConfig config) {
         List<String> listeners = parseListeners();
 
         jettyServer = new Server();
+        handlers = new ContextHandlerCollection();
 
         createConnectors(listeners);
     }
@@ -159,21 +160,40 @@ public Connector createConnector(String listener) {
         return connector;
     }
 
+    public void initializeServer() {
+        log.info("Initializing REST server");
+
+        /* Needed for graceful shutdown as per `setStopTimeout` documentation */
+        StatisticsHandler statsHandler = new StatisticsHandler();
+        statsHandler.setHandler(handlers);
+        jettyServer.setHandler(statsHandler);
+        jettyServer.setStopTimeout(GRACEFUL_SHUTDOWN_TIMEOUT_MS);
+        jettyServer.setStopAtShutdown(true);
+
+        try {
+            jettyServer.start();
+        } catch (Exception e) {
+            throw new ConnectException("Unable to initialize REST server", e);
+        }
+
+        log.info("REST server listening at " + jettyServer.getURI() + ", advertising URL " + advertisedUrl());
+    }
+
     @SuppressWarnings("deprecation")
-    public void start(HerderProvider herderProvider, Plugins plugins) {
-        log.info("Starting REST server");
+    public void initializeResources(Herder herder) {
+        log.info("Initializing REST resources");
 
         ResourceConfig resourceConfig = new ResourceConfig();
         resourceConfig.register(new JacksonJsonProvider());
 
-        resourceConfig.register(new RootResource(herderProvider));
-        resourceConfig.register(new ConnectorsResource(herderProvider, config));
-        resourceConfig.register(new ConnectorPluginsResource(herderProvider));
+        resourceConfig.register(new RootResource(herder));
+        resourceConfig.register(new ConnectorsResource(herder, config));
+        resourceConfig.register(new ConnectorPluginsResource(herder));
 
         resourceConfig.register(ConnectExceptionMapper.class);
         resourceConfig.property(ServerProperties.WADL_FEATURE_DISABLE, true);
 
-        registerRestExtensions(herderProvider, plugins, resourceConfig);
+        registerRestExtensions(herder, resourceConfig);
 
         ServletContainer servletContainer = new ServletContainer(resourceConfig);
         ServletHolder servletHolder = new ServletHolder(servletContainer);
@@ -201,23 +221,14 @@ public void start(HerderProvider herderProvider, Plugins plugins) {
         requestLog.setLogLatency(true);
         requestLogHandler.setRequestLog(requestLog);
 
-        HandlerCollection handlers = new HandlerCollection();
         handlers.setHandlers(new Handler[]{context, new DefaultHandler(), requestLogHandler});
-
-        /* Needed for graceful shutdown as per `setStopTimeout` documentation */
-        StatisticsHandler statsHandler = new StatisticsHandler();
-        statsHandler.setHandler(handlers);
-        jettyServer.setHandler(statsHandler);
-        jettyServer.setStopTimeout(GRACEFUL_SHUTDOWN_TIMEOUT_MS);
-        jettyServer.setStopAtShutdown(true);
-
         try {
-            jettyServer.start();
+            context.start();
         } catch (Exception e) {
-            throw new ConnectException("Unable to start REST server", e);
+            throw new ConnectException("Unable to initialize REST resources", e);
         }
 
-        log.info("REST server listening at " + jettyServer.getURI() + ", advertising URL " + advertisedUrl());
+        log.info("REST resources initialized; server is started and ready to handle requests");
     }
 
     public URI serverUrl() {
@@ -238,17 +249,17 @@ public void stop() {
             jettyServer.stop();
             jettyServer.join();
         } catch (Exception e) {
-            throw new ConnectException("Unable to stop REST server", e);
-        } finally {
             jettyServer.destroy();
+            throw new ConnectException("Unable to stop REST server", e);
         }
 
         log.info("REST server stopped");
     }
 
     /**
      * Get the URL to advertise to other workers and clients. This uses the default connector from the embedded Jetty
-     * server, unless overrides for advertised hostname and/or port are provided via configs.
+     * server, unless overrides for advertised hostname and/or port are provided via configs. {@link #initializeServer()}
+     * must be invoked successfully before calling this method.
      */
     public URI advertisedUrl() {
         UriBuilder builder = UriBuilder.fromUri(jettyServer.getURI());
@@ -304,8 +315,8 @@ ServerConnector findConnector(String protocol) {
         return null;
     }
 
-    void registerRestExtensions(HerderProvider provider, Plugins plugins, ResourceConfig resourceConfig) {
-        connectRestExtensions = plugins.newPlugins(
+    void registerRestExtensions(Herder herder, ResourceConfig resourceConfig) {
+        connectRestExtensions = herder.plugins().newPlugins(
             config.getList(WorkerConfig.REST_EXTENSION_CLASSES_CONFIG),
             config, ConnectRestExtension.class);
 
@@ -320,7 +331,7 @@ void registerRestExtensions(HerderProvider provider, Plugins plugins, ResourceCo
         ConnectRestExtensionContext connectRestExtensionContext =
             new ConnectRestExtensionContextImpl(
                 new ConnectRestConfigurable(resourceConfig),
-                new ConnectClusterStateImpl(herderRequestTimeoutMs, provider)
+                new ConnectClusterStateImpl(herderRequestTimeoutMs, herder)
             );
         for (ConnectRestExtension connectRestExtension : connectRestExtensions) {
             connectRestExtension.register(connectRestExtensionContext);

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorPluginsResource.java

@@ -18,7 +18,7 @@
 
 import org.apache.kafka.connect.connector.Connector;
 import org.apache.kafka.connect.runtime.ConnectorConfig;
-import org.apache.kafka.connect.runtime.HerderProvider;
+import org.apache.kafka.connect.runtime.Herder;
 import org.apache.kafka.connect.runtime.isolation.PluginDesc;
 import org.apache.kafka.connect.runtime.rest.entities.ConfigInfos;
 import org.apache.kafka.connect.runtime.rest.entities.ConnectorPluginInfo;
@@ -49,7 +49,7 @@
 public class ConnectorPluginsResource {
 
     private static final String ALIAS_SUFFIX = "Connector";
-    private final HerderProvider herderProvider;
+    private final Herder herder;
     private final List<ConnectorPluginInfo> connectorPlugins;
 
     private static final List<Class<? extends Connector>> CONNECTOR_EXCLUDES = Arrays.asList(
@@ -58,8 +58,8 @@ public class ConnectorPluginsResource {
             SchemaSourceConnector.class
     );
 
-    public ConnectorPluginsResource(HerderProvider herderProvider) {
-        this.herderProvider = herderProvider;
+    public ConnectorPluginsResource(Herder herder) {
+        this.herder = herder;
         this.connectorPlugins = new ArrayList<>();
     }
 
@@ -78,7 +78,7 @@ public ConfigInfos validateConfigs(
             );
         }
 
-        return herderProvider.get().validateConnectorConfig(connectorConfig);
+        return herder.validateConnectorConfig(connectorConfig);
     }
 
     @GET
@@ -90,7 +90,7 @@ public List<ConnectorPluginInfo> listConnectorPlugins() {
     // TODO: improve once plugins are allowed to be added/removed during runtime.
     private synchronized List<ConnectorPluginInfo> getConnectorPlugins() {
         if (connectorPlugins.isEmpty()) {
-            for (PluginDesc<Connector> plugin : herderProvider.get().plugins().connectors()) {
+            for (PluginDesc<Connector> plugin : herder.plugins().connectors()) {
                 if (!CONNECTOR_EXCLUDES.contains(plugin.pluginClass())) {
                     connectorPlugins.add(new ConnectorPluginInfo(plugin));
                 }