Refactor ClusterTestHarness to extend generic SR test harness (#4007)

Author: GunalKupta

client-encryption/src/test/java/io/confluent/kafka/schemaregistry/encryption/RestApiFieldEncryptionTest.java

@@ -82,7 +82,7 @@ public RestApiFieldEncryptionTest() {
   protected abstract EncryptionProperties getFieldEncryptionProperties(List<String> ruleNames);
 
   @Override
-  protected Properties getSchemaRegistryProperties() throws Exception {
+  public Properties getSchemaRegistryProperties() throws Exception {
     Properties props = new Properties();
     props.setProperty("resource.extension.class", DekRegistryResourceExtension.class.getName());
     Object testClient = getFieldEncryptionProperties(null).getTestClient();
@@ -93,7 +93,7 @@ protected Properties getSchemaRegistryProperties() throws Exception {
   }
 
   @Override
-  protected void setUp() throws Exception {
+  public void setUp() throws Exception {
     super.setUp();
     restApp.schemaRegistry().setRuleSetHandler(new RuleSetHandler() {
       public void handle(String subject, ConfigUpdateRequest request) {

client-encryption/src/test/java/io/confluent/kafka/schemaregistry/rules/WidgetBytesProto.java

@@ -77,7 +77,7 @@
  */
 @Tag("IntegrationTest")
 @DisplayNameGeneration(AddKraftQuorum.class)
-public abstract class ClusterTestHarness {
+public abstract class ClusterTestHarness implements SchemaRegistryTestHarness {
 
   private static final Logger log = LoggerFactory.getLogger(ClusterTestHarness.class);
 
@@ -109,9 +109,10 @@ public static int[] choosePorts(int count) {
   /**
    * Choose an available port
    */
-  public static int choosePort() {
-    return choosePorts(1)[0];
-  }
+ @Override
+ public int choosePort() {
+   return choosePorts(1)[0];
+ }
 
   private final int numBrokers;
   private final boolean setupRestApp;
@@ -156,7 +157,7 @@ public void setUpTest(TestInfo testInfo) throws Exception {
     log.info("Completed setup of {}", getClass().getSimpleName());
   }
 
-  protected void setUp() throws Exception {
+  public void setUp() throws Exception {
     checkState(testInfo != null);
     log.info("Starting controller of {}", getClass().getSimpleName());
     // start controller
@@ -175,8 +176,7 @@ protected void setUp() throws Exception {
     setupAcls();
 
     if (setupRestApp) {
-      if (schemaRegistryPort == null)
-        schemaRegistryPort = choosePort();
+      schemaRegistryPort = getSchemaRegistryPort();
       Properties schemaRegistryProps = getSchemaRegistryProperties();
       if (!schemaRegistryProps.containsKey(SchemaRegistryConfig.LISTENERS_CONFIG)) {
         schemaRegistryProps.put(SchemaRegistryConfig.LISTENERS_CONFIG, getSchemaRegistryProtocol() +
@@ -189,17 +189,18 @@ protected void setUp() throws Exception {
     }
   }
 
-  protected void setupRestApp(Properties schemaRegistryProps) throws Exception {
+  public void setupRestApp(Properties schemaRegistryProps) throws Exception {
     restApp = new RestApp(schemaRegistryPort, null, brokerList, KAFKASTORE_TOPIC,
                           compatibilityType, true, schemaRegistryProps);
     restApp.start();
   }
 
-  protected Properties getSchemaRegistryProperties() throws Exception {
+  public Properties getSchemaRegistryProperties() throws Exception {
     return new Properties();
   }
 
-  protected void injectProperties(Properties props) {
+  @Override
+  public void injectProperties(Properties props) {
     // Make sure that broker only role is "broker"
     props.setProperty("process.roles", "broker");
     props.setProperty("message.max.bytes", String.valueOf(MAX_MESSAGE_SIZE));
@@ -236,18 +237,36 @@ protected KafkaConfig getKafkaConfig(int brokerId) {
     return KafkaConfig.fromProps(props);
   }
 
-  protected SecurityProtocol getBrokerSecurityProtocol() {
+  public SecurityProtocol getBrokerSecurityProtocol() {
     return SecurityProtocol.PLAINTEXT;
   }
 
-  protected String getSchemaRegistryProtocol() {
+  @Override
+  public String getSchemaRegistryProtocol() {
     return SchemaRegistryConfig.HTTP;
   }
 
-  protected Time brokerTime(int brokerId) {
+  public Time brokerTime(int brokerId) {
     return Time.SYSTEM;
   }
 
+  @Override
+  public String getBrokerList() {
+    return brokerList;
+  }
+
+  @Override
+  public RestApp getRestApp() {
+    return restApp;
+  }
+  
+  @Override
+  public Integer getSchemaRegistryPort() {
+    if (schemaRegistryPort == null)
+        schemaRegistryPort = choosePort();
+    return schemaRegistryPort;
+  }
+
   private void startBrokersConcurrently(int numBrokers) {
     log.info("Starting concurrently {} brokers for {}", numBrokers, getClass().getSimpleName());
     configs =

client-encryption/src/test/java/io/confluent/kafka/schemaregistry/rules/WidgetProto.java

@@ -51,12 +51,12 @@ public SASLClusterTestHarness() {
   }
 
   @Override
-  protected SecurityProtocol getBrokerSecurityProtocol() {
+  public SecurityProtocol getBrokerSecurityProtocol() {
     return SecurityProtocol.SASL_PLAINTEXT;
   }
 
   @Override
-  protected void setUp() throws Exception {
+  public void setUp() throws Exception {
     // Important if tests leak consumers, producers or brokers.
     LoginManager.closeAll();
 

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

@@ -36,7 +36,7 @@ public SSLClusterTestHarness() {
   }
 
   @Override
-  protected SecurityProtocol getBrokerSecurityProtocol() {
+  public SecurityProtocol getBrokerSecurityProtocol() {
     return SecurityProtocol.SSL;
   }