Merge branch '7.8.x' into 7.9.x by trnguyencflt

Author: ConfluentSemaphore

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

@@ -33,6 +33,7 @@
 import io.confluent.kafkarest.KafkaRestApplication;
 import io.confluent.kafkarest.KafkaRestConfig;
 import io.confluent.kafkarest.common.CompletableFutures;
+import io.confluent.rest.RestConfig;
 import java.io.IOException;
 import java.net.ServerSocket;
 import java.net.URI;
@@ -78,6 +79,7 @@
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.config.ConfigResource;
+import org.apache.kafka.common.network.ListenerName;
 import org.apache.kafka.common.security.auth.SecurityProtocol;
 import org.apache.kafka.common.serialization.ByteArraySerializer;
 import org.apache.kafka.common.serialization.Deserializer;
@@ -222,49 +224,57 @@ private void setupMethod() throws Exception {
 
     setupAcls();
     if (withSchemaRegistry) {
-      int schemaRegPort = choosePort();
-      schemaRegProperties.put(
-          SchemaRegistryConfig.LISTENERS_CONFIG,
-          String.format("http://127.0.0.1:%d", schemaRegPort));
-      schemaRegProperties.put(
-          SchemaRegistryConfig.KAFKASTORE_TOPIC_CONFIG,
-          SchemaRegistryConfig.DEFAULT_KAFKASTORE_TOPIC);
-      schemaRegProperties.put(
-          SchemaRegistryConfig.SCHEMA_COMPATIBILITY_CONFIG, schemaRegCompatibility);
-      String broker =
-          SecurityProtocol.PLAINTEXT.name
-              + "://"
-              + TestUtils.getBrokerListStrFromServers(
-                  JavaConverters.asScalaBuffer(servers), SecurityProtocol.PLAINTEXT);
-      schemaRegProperties.put(SchemaRegistryConfig.KAFKASTORE_BOOTSTRAP_SERVERS_CONFIG, broker);
-      schemaRegConnect = String.format("http://localhost:%d", schemaRegPort);
-
-      schemaRegProperties = overrideSchemaRegistryProps(schemaRegProperties);
-
-      schemaRegApp =
-          new SchemaRegistryRestApplication(new SchemaRegistryConfig(schemaRegProperties));
-      schemaRegServer = schemaRegApp.createServer();
-      schemaRegServer.start();
-      schemaRegApp.postServerStart();
+      doStartSchemaRegistry();
     }
-
     if (manageRest) {
-      startRest(null, null);
+      startRest(brokerList, null, null);
     }
   }
 
+  private void doStartSchemaRegistry() throws Exception {
+    int schemaRegPort = choosePort();
+    schemaRegProperties.put(
+        SchemaRegistryConfig.LISTENERS_CONFIG, String.format("http://127.0.0.1:%d", schemaRegPort));
+    schemaRegProperties.put(
+        SchemaRegistryConfig.KAFKASTORE_TOPIC_CONFIG,
+        SchemaRegistryConfig.DEFAULT_KAFKASTORE_TOPIC);
+    schemaRegProperties.put(
+        SchemaRegistryConfig.SCHEMA_COMPATIBILITY_CONFIG, schemaRegCompatibility);
+    String broker =
+        SecurityProtocol.PLAINTEXT.name
+            + "://"
+            + TestUtils.getBrokerListStrFromServers(
+                JavaConverters.asScalaBuffer(servers), SecurityProtocol.PLAINTEXT);
+    schemaRegProperties.put(SchemaRegistryConfig.KAFKASTORE_BOOTSTRAP_SERVERS_CONFIG, broker);
+    schemaRegConnect = String.format("http://localhost:%d", schemaRegPort);
+
+    schemaRegProperties = overrideSchemaRegistryProps(schemaRegProperties);
+
+    schemaRegApp = new SchemaRegistryRestApplication(new SchemaRegistryConfig(schemaRegProperties));
+    schemaRegServer = schemaRegApp.createServer();
+    schemaRegServer.start();
+    schemaRegApp.postServerStart();
+  }
+
   protected void startRest(RequestLog.Writer requestLogWriter, String requestLogFormat)
       throws Exception {
+    startRest(brokerList, requestLogWriter, requestLogFormat);
+  }
+
+  protected void startRest(
+      String bootstrapServers, RequestLog.Writer requestLogWriter, String requestLogFormat)
+      throws Exception {
+    if (restServer != null && restServer.isRunning()) {
+      log.warn("Rest server already started, skipping start");
+      return;
+    }
     log.info("Setting up REST.");
-    int restPort = choosePort();
-    restProperties.put(KafkaRestConfig.PORT_CONFIG, ((Integer) restPort).toString());
-    restProperties.put(KafkaRestConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
+    restProperties.put(KafkaRestConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
     overrideKafkaRestConfigs(restProperties);
-    if (withSchemaRegistry) {
+    if (withSchemaRegistry && schemaRegConnect != null) {
       restProperties.put(KafkaRestConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegConnect);
     }
-    restConnect = getRestConnectString(restPort);
-    restProperties.put("listeners", restConnect);
+    restProperties.put(RestConfig.LISTENERS_CONFIG, getRestConnectString(0));
 
     // Reduce the metadata fetch timeout so requests for topics that don't exist timeout much
     // faster than the default
@@ -273,22 +283,49 @@ protected void startRest(RequestLog.Writer requestLogWriter, String requestLogFo
         "producer." + ProducerConfig.MAX_REQUEST_SIZE_CONFIG, String.valueOf((2 << 20) * 10));
 
     restConfig = new KafkaRestConfig(restProperties);
-    restApp = new KafkaRestApplication(restConfig, "", null, requestLogWriter, requestLogFormat);
 
     try {
-      restServer = restApp.createServer();
-      restServer.start();
-    } catch (IOException e1) { // sometimes we get an address already in use exception
-      log.warn("IOException when attempting to start rest, trying again", e1);
+      doStartRest(requestLogWriter, requestLogFormat);
+    } catch (IOException e) { // sometimes we get an address already in use exception
+      log.warn("IOException when attempting to start rest, trying again", e);
       stopRest();
       Thread.sleep(Duration.ofSeconds(1).toMillis());
       try {
-        startRest(null, null);
+        doStartRest(requestLogWriter, requestLogFormat);
       } catch (IOException e2) {
         log.error("Restart of rest server failed", e2);
         throw e2;
       }
     }
+    restConnect = getRestConnectString(restServer.getURI().getPort());
+  }
+
+  /**
+   * Return the bootstrap servers string for the given security protocol.
+   *
+   * @param securityProtocol security protocol
+   * @return bootstrap servers string
+   */
+  public String getBootstrapServers(SecurityProtocol securityProtocol) {
+    return TestUtils.getBrokerListStrFromServers(
+        JavaConverters.asScalaBuffer(servers), securityProtocol);
+  }
+
+  /**
+   * Return the bootstrap servers string for the given listener name.
+   *
+   * @param listenerName listener name
+   * @return bootstrap servers string
+   */
+  public String getBootstrapServers(ListenerName listenerName) {
+    return TestUtils.bootstrapServers(JavaConverters.asScalaBuffer(servers), listenerName);
+  }
+
+  private void doStartRest(RequestLog.Writer requestLogWriter, String requestLogFormat)
+      throws Exception {
+    restApp = new KafkaRestApplication(restConfig, "", null, requestLogWriter, requestLogFormat);
+    restServer = restApp.createServer();
+    restServer.start();
   }
 
   protected void stopRest() throws Exception {