Merge branch '7.2.x' into 7.3.x

confluentinc · Oct 10, 2022 · 9cfcc68 · 9cfcc68
2 parents 9d7e8d3 + 5261b6d
commit 9cfcc68
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
@@ -529,7 +529,7 @@ public int register(String subject,
       schema.setSchema(parsedSchema.canonicalString());
       schema.setReferences(parsedSchema.references());
 
-      if (isCompatible) {
+      if (isCompatible || getModeInScope(subject) == Mode.IMPORT) {
         // save the context key
         QualifiedSubject qs = QualifiedSubject.create(tenant(), subject);
         if (qs != null && !DEFAULT_CONTEXT.equals(qs.getContext())) {

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();
+
   private static String SCHEMA_WITH_DECIMAL = AvroUtils.parseSchema(
           "{\n"
               + "    \"type\": \"record\",\n"
@@ -340,4 +347,46 @@ public void testImportModeWithSameSchemaDifferentId() throws Exception {
         SCHEMA_WITH_DECIMAL,
         restApp.restClient.getVersion(subject, 2).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());
+  }
 }