From d616c769b36616aa5ad536e134a2fa55237a1cb1 Mon Sep 17 00:00:00 2001
From: Robert Yokota <rayokota@gmail.com>
Date: Mon, 10 Oct 2022 15:49:18 -0700
Subject: [PATCH] DGS-5084 Ignore compat check in IMPORT mode (#2399)

---
 .../storage/KafkaSchemaRegistry.java          |  2 +-
 .../schemaregistry/rest/RestApiModeTest.java  | 49 +++++++++++++++++++
 2 files changed, 50 insertions(+), 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
index fa8493ef4d3..80efaca0d90 100644
--- a/core/src/main/java/io/confluent/kafka/schemaregistry/storage/KafkaSchemaRegistry.java
+++ b/core/src/main/java/io/confluent/kafka/schemaregistry/storage/KafkaSchemaRegistry.java
@@ -475,7 +475,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
index 82d00ef6d69..41584a0db66 100644
--- 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());
+  }
 }