Merge branch '5.5.x' into 6.0.x

confluentinc · Oct 10, 2022 · fd4f10d · fd4f10d
2 parents b2a9360 + d616c76
commit fd4f10d
Show file tree

Hide file tree

Showing 2 changed files with 50 additions and 1 deletion.
diff --git a/core/src/main/java/io/confluent/kafka/schemaregistry/storage/KafkaSchemaRegistry.java b/core/src/main/java/io/confluent/kafka/schemaregistry/storage/KafkaSchemaRegistry.java
@@ -473,7 +473,7 @@ public int register(String subject,
       schema.setSchema(parsedSchema.canonicalString());
       schema.setReferences(parsedSchema.references());
 
-      if (isCompatible) {
+      if (isCompatible || getModeInScope(subject) == Mode.IMPORT) {
         // assign a guid and put the schema in the kafka store
         if (schema.getVersion() <= 0) {
           schema.setVersion(newVersion);

diff --git a/core/src/test/java/io/confluent/kafka/schemaregistry/rest/RestApiModeTest.java b/core/src/test/java/io/confluent/kafka/schemaregistry/rest/RestApiModeTest.java
@@ -36,6 +36,13 @@ public class RestApiModeTest extends ClusterTestHarness {
           + "[{\"type\":\"string\",\"name\":\"f1\"}]}")
       .canonicalString();
 
+  private static String SCHEMA2_STRING = AvroUtils.parseSchema(
+          "{\"type\":\"record\","
+              + "\"name\":\"myrecord\","
+              + "\"fields\":"
+              + "[{\"type\":\"int\",\"name\":\"f1\"}]}")
+      .canonicalString();
+
   public RestApiModeTest() {
     super(1, true, CompatibilityLevel.BACKWARD.name);
   }
@@ -226,4 +233,46 @@ public void testRegisterSchemaWithSameIdAfterImport() throws Exception {
             SCHEMA_STRING,
             restApp.restClient.getVersion(subject, 1).getSchema());
   }
+
+  @Test
+  public void testRegisterIncompatibleSchemaDuringImport() throws Exception {
+    String subject = "testSubject";
+    String mode = "READWRITE";
+
+    // set mode to read write
+    assertEquals(
+        mode,
+        restApp.restClient.setMode(mode).getMode());
+
+    int expectedIdSchema1 = 1;
+    assertEquals("Registering without id should succeed",
+        expectedIdSchema1,
+        restApp.restClient.registerSchema(SCHEMA_STRING, subject));
+
+    // delete subject so we can switch to import mode
+    restApp.restClient.deleteSubject(Collections.emptyMap(), subject);
+
+    mode = "IMPORT";
+
+    // set mode to import
+    assertEquals(
+        mode,
+        restApp.restClient.setMode(mode).getMode());
+
+    // register same schema with same id
+    expectedIdSchema1 = 1;
+    assertEquals("Registering with id should succeed",
+        expectedIdSchema1,
+        restApp.restClient.registerSchema(SCHEMA_STRING, subject, 1, expectedIdSchema1));
+
+    // register same schema with same id
+    expectedIdSchema1 = 2;
+    assertEquals("Registering with id should succeed",
+        expectedIdSchema1,
+        restApp.restClient.registerSchema(SCHEMA2_STRING, subject, 2, expectedIdSchema1));
+
+    assertEquals("Getting schema by id should succeed",
+        SCHEMA2_STRING,
+        restApp.restClient.getVersion(subject, 2).getSchema());
+  }
 }