Merge branch '5.5.x' into 6.0.x

confluentinc · Oct 11, 2021 · f74b3a5 · f74b3a5
2 parents 794f955 + 6f9129b
commit f74b3a5
Show file tree

Hide file tree

Showing 2 changed files with 47 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
@@ -504,7 +504,8 @@ public int register(String subject,
           }
         }
         for (SchemaValue schemaValue : deletedVersions) {
-          if (schemaValue.getId().equals(schema.getId())) {
+          if (schemaValue.getId().equals(schema.getId())
+                  && schemaValue.getVersion().compareTo(schema.getVersion()) < 0) {
             // Tombstone previous version with the same ID
             SchemaKey key = new SchemaKey(schemaValue.getSubject(), schemaValue.getVersion());
             kafkaStore.delete(key);

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
@@ -181,4 +181,49 @@ public void testRegisterSchemaWithDifferentIdAfterImport() throws Exception {
         expectedIdSchema1,
         restApp.restClient.registerSchema(SCHEMA_STRING, subject, 1, expectedIdSchema1));
   }
+
+  @Test
+  public void testRegisterSchemaWithSameIdAfterImport() 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));
+
+    // delete subject again
+    restApp.restClient.deleteSubject(Collections.emptyMap(), subject);
+
+    // register same schema with same id
+    expectedIdSchema1 = 1;
+    assertEquals("Registering with id should succeed",
+            expectedIdSchema1,
+            restApp.restClient.registerSchema(SCHEMA_STRING, subject, 1, expectedIdSchema1));
+
+    assertEquals("Getting schema by id should succeed",
+            SCHEMA_STRING,
+            restApp.restClient.getVersion(subject, 1).getSchema());
+  }
 }