Test: Updated default schema registry compatibilty

Author: rozza

src/integrationTest/java/com/mongodb/kafka/connect/embedded/EmbeddedKafka.java

@@ -62,7 +62,7 @@ public class EmbeddedKafka implements BeforeAllCallback, AfterEachCallback, Afte
   private static final Logger LOGGER = LoggerFactory.getLogger(EmbeddedKafka.class);
   private static final int DEFAULT_BROKER_PORT = 0; // 0 results in a random port being selected
   private static final String KAFKA_SCHEMAS_TOPIC = "_schemas";
-  private static final String AVRO_COMPATIBILITY_TYPE = "NONE";
+  private static final String COMPATIBILITY_LEVEL = "BACKWARD";
 
   private static final String KAFKASTORE_OPERATION_TIMEOUT_MS = "10000";
   private static final String KAFKASTORE_DEBUG = "true";
@@ -198,11 +198,7 @@ public void start() throws Exception {
 
     schemaRegistry =
         new RestApp(
-            0,
-            zookeeperConnect(),
-            KAFKA_SCHEMAS_TOPIC,
-            AVRO_COMPATIBILITY_TYPE,
-            schemaRegistryProps);
+            zookeeperConnect(), KAFKA_SCHEMAS_TOPIC, COMPATIBILITY_LEVEL, schemaRegistryProps);
     schemaRegistry.start();
 
     LOGGER.debug("Starting a Connect standalone instance...");

src/integrationTest/java/com/mongodb/kafka/connect/embedded/RestApp.java

@@ -19,7 +19,7 @@
 
 import java.util.Properties;
 
-import io.confluent.kafka.schemaregistry.avro.AvroCompatibilityLevel;
+import io.confluent.kafka.schemaregistry.CompatibilityLevel;
 import io.confluent.kafka.schemaregistry.client.rest.RestService;
 import io.confluent.kafka.schemaregistry.exceptions.SchemaRegistryException;
 import io.confluent.kafka.schemaregistry.rest.SchemaRegistryConfig;
@@ -37,38 +37,28 @@ public class RestApp {
   public Server restServer;
   public String restConnect;
 
-  public RestApp(int port, String zkConnect, String kafkaTopic) {
-    this(port, zkConnect, kafkaTopic, AvroCompatibilityLevel.NONE.name, null);
+  public RestApp(String zkConnect, String kafkaTopic) {
+    this(zkConnect, kafkaTopic, CompatibilityLevel.NONE.name, null);
   }
 
   public RestApp(
-      int port,
       String zkConnect,
       String kafkaTopic,
       String compatibilityType,
       Properties schemaRegistryProps) {
-    this(port, zkConnect, null, kafkaTopic, compatibilityType, true, schemaRegistryProps);
+    this(zkConnect, null, kafkaTopic, compatibilityType, true, schemaRegistryProps);
   }
 
   public RestApp(
-      int port,
       String zkConnect,
       String kafkaTopic,
       String compatibilityType,
       boolean masterEligibility,
       Properties schemaRegistryProps) {
-    this(
-        port,
-        zkConnect,
-        null,
-        kafkaTopic,
-        compatibilityType,
-        masterEligibility,
-        schemaRegistryProps);
+    this(zkConnect, null, kafkaTopic, compatibilityType, masterEligibility, schemaRegistryProps);
   }
 
   public RestApp(
-      int port,
       String zkConnect,
       String bootstrapBrokers,
       String kafkaTopic,
@@ -79,15 +69,14 @@ public RestApp(
     if (schemaRegistryProps != null) {
       prop.putAll(schemaRegistryProps);
     }
-    prop.setProperty(SchemaRegistryConfig.PORT_CONFIG, ((Integer) port).toString());
     if (zkConnect != null) {
       prop.setProperty(SchemaRegistryConfig.KAFKASTORE_CONNECTION_URL_CONFIG, zkConnect);
     }
     if (bootstrapBrokers != null) {
       prop.setProperty(SchemaRegistryConfig.KAFKASTORE_BOOTSTRAP_SERVERS_CONFIG, bootstrapBrokers);
     }
     prop.put(SchemaRegistryConfig.KAFKASTORE_TOPIC_CONFIG, kafkaTopic);
-    prop.put(SchemaRegistryConfig.COMPATIBILITY_CONFIG, compatibilityType);
+    prop.put(SchemaRegistryConfig.SCHEMA_COMPATIBILITY_CONFIG, compatibilityType);
     prop.put(SchemaRegistryConfig.MASTER_ELIGIBILITY, masterEligibility);
   }