Merge branch '8.1.x' into master by GunalKupta

Author: semaphore-agent-production[bot]

core/src/test/java/io/confluent/kafka/schemaregistry/ClusterTestHarness.java

@@ -37,9 +37,6 @@
 import org.apache.kafka.common.utils.Java;
 import org.apache.kafka.common.utils.Time;
 
-import java.io.IOException;
-import java.net.InetAddress;
-import java.net.ServerSocket;
 import java.util.List;
 import java.util.Properties;
 
@@ -77,7 +74,7 @@
  */
 @Tag("IntegrationTest")
 @DisplayNameGeneration(AddKraftQuorum.class)
-public abstract class ClusterTestHarness implements SchemaRegistryTestHarness {
+public class ClusterTestHarness implements SchemaRegistryTestHarness {
 
   private static final Logger log = LoggerFactory.getLogger(ClusterTestHarness.class);
 
@@ -86,38 +83,12 @@ public abstract class ClusterTestHarness implements SchemaRegistryTestHarness {
   public static final String KAFKASTORE_TOPIC = SchemaRegistryConfig.DEFAULT_KAFKASTORE_TOPIC;
   protected static final Optional<Properties> EMPTY_SASL_PROPERTIES = Optional.empty();
 
-  /**
-   * Choose a number of random available ports
-   */
-  public static int[] choosePorts(int count) {
-    try {
-      ServerSocket[] sockets = new ServerSocket[count];
-      int[] ports = new int[count];
-      for (int i = 0; i < count; i++) {
-        sockets[i] = new ServerSocket(0, 0, InetAddress.getByName("0.0.0.0"));
-        ports[i] = sockets[i].getLocalPort();
-      }
-      for (int i = 0; i < count; i++) {
-        sockets[i].close();
-      }
-      return ports;
-    } catch (IOException e) {
-      throw new RuntimeException(e);
-    }
-  }
-
-  /**
-   * Choose an available port
-   */
- @Override
- public int choosePort() {
-   return choosePorts(1)[0];
- }
-
   private final int numBrokers;
   private final boolean setupRestApp;
   protected String compatibilityType;
 
+  protected Properties schemaRegistryProperties;
+
   // Quorum controller
   private TestInfo testInfo;
   private QuorumTestHarness quorumTestHarness;
@@ -147,6 +118,7 @@ public ClusterTestHarness(int numBrokers, boolean setupRestApp, String compatibi
     this.numBrokers = numBrokers;
     this.setupRestApp = setupRestApp;
     this.compatibilityType = compatibilityType;
+    this.schemaRegistryProperties = new Properties();
   }
 
   @BeforeEach
@@ -195,11 +167,28 @@ public void setupRestApp(Properties schemaRegistryProps) throws Exception {
     restApp.start();
   }
 
+  /**
+   * Subclasses can override this method to provide custom schema registry properties
+   * or add properties via {@link #injectSchemaRegistryProperties(Properties)}.
+   * @return schema registry properties
+   * @throws Exception if an error occurs
+   */
   public Properties getSchemaRegistryProperties() throws Exception {
-    return new Properties();
+    return schemaRegistryProperties;
   }
 
-  @Override
+  /**
+   * Inject properties into schema registry configuration.
+   * @param props properties to inject
+   */
+  public void injectSchemaRegistryProperties(Properties props) {
+    this.schemaRegistryProperties.putAll(props);
+  }
+
+  /**
+   * Inject properties into broker configuration.
+   * @param props properties to inject
+   */
   public void injectProperties(Properties props) {
     // Make sure that broker only role is "broker"
     props.setProperty("process.roles", "broker");
@@ -250,7 +239,10 @@ public Time brokerTime(int brokerId) {
     return Time.SYSTEM;
   }
 
-  @Override
+  /**
+   * Gets the broker list for Kafka connections.
+   * @return broker list string, or null if not applicable
+   */
   public String getBrokerList() {
     return brokerList;
   }

core/src/test/java/io/confluent/kafka/schemaregistry/SchemaRegistryTestHarness.java

@@ -15,20 +15,15 @@
 
 package io.confluent.kafka.schemaregistry;
 
-import java.util.Properties;
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.ServerSocket;
 
 /**
  * Interface defining common operations needed by schema registry integration tests.
- * This allows tests to be written against non-Kafka-based implementations.
+ * This allows tests to be written and run against an arbitrary test harness.
  */
 public interface SchemaRegistryTestHarness {
-  
-  /**
-   * Gets the broker list for Kafka connections.
-   * @return broker list string, or null if not applicable
-   */
-  String getBrokerList();
-  
   /**
    * Gets the REST application instance for making schema registry API calls.
    * @return RestApp instance
@@ -40,23 +35,40 @@ public interface SchemaRegistryTestHarness {
    * @return schema registry port number
    */
   Integer getSchemaRegistryPort();
-  
+
   /**
    * Chooses an available port for schema registry or other services.
    * @return available port number
    */
-  int choosePort();
+  default int choosePort() {
+    return choosePorts(1)[0];
+  }
+
+
+  /**
+   * Choose a number of random available ports
+   */
+  static int[] choosePorts(int count) {
+    try {
+      ServerSocket[] sockets = new ServerSocket[count];
+      int[] ports = new int[count];
+      for (int i = 0; i < count; i++) {
+        sockets[i] = new ServerSocket(0, 0, InetAddress.getByName("0.0.0.0"));
+        ports[i] = sockets[i].getLocalPort();
+      }
+      for (int i = 0; i < count; i++) {
+        sockets[i].close();
+      }
+      return ports;
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+  }
 
   /**
    * Gets the schema registry protocol (http or https).
    * @return protocol string
    */
   String getSchemaRegistryProtocol();
-  
-  /**
-   * Inject properties into broker configuration.
-   * @param props properties to inject
-   */
-  void injectProperties(Properties props);
 }
 

core/src/test/java/io/confluent/kafka/schemaregistry/rest/RestApiClusterTest.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2018 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License (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.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.kafka.schemaregistry.rest;
+
+import io.confluent.kafka.schemaregistry.ClusterTestHarness;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestInfo;
+
+import java.util.Properties;
+
+/**
+ * ClusterTestHarness implementation of REST API integration tests.
+ */
+public class RestApiClusterTest extends RestApiTest {
+
+  protected ClusterTestHarness harness;
+
+  public RestApiClusterTest() {
+    this.harness = new ClusterTestHarness(1, true);
+    this.harness.injectSchemaRegistryProperties(getSchemaRegistryProperties());
+  }
+
+  @BeforeEach
+  public void setUpTest(TestInfo testInfo) throws Exception {
+    harness.setUpTest(testInfo);
+    setRestApp(harness.getRestApp());
+  }
+
+  @AfterEach
+  public void tearDown() throws Exception {
+    harness.tearDown();
+  }
+
+  public Properties getSchemaRegistryProperties() {
+    Properties schemaRegistryProps = new Properties();
+    schemaRegistryProps.put("response.http.headers.config",
+        "add X-XSS-Protection: 1; mode=block, \"add Cache-Control: no-cache, no-store, must-revalidate\"");
+    schemaRegistryProps.put("schema.providers.avro.validate.defaults", "true");
+    return schemaRegistryProps;
+  }
+}