REVERT: DGS-20986 - (Breaking API Change) (#3922) (#3932)

Author: Nitss10

core/src/main/java/io/confluent/kafka/schemaregistry/rest/resources/SubjectsResource.java

@@ -121,6 +121,7 @@ public void lookUpSchemaUnderSubject(
              subject, lookupDeletedSchema, request.getSchemaType());
 
     subject = QualifiedSubject.normalize(schemaRegistry.tenant(), subject);
+
     // returns version if the schema exists. Otherwise returns 404
     Schema schema = new Schema(subject, request);
     io.confluent.kafka.schemaregistry.client.rest.entities.Schema matchingSchema;
@@ -130,15 +131,6 @@ public void lookUpSchemaUnderSubject(
       }
       matchingSchema = schemaRegistry.lookUpSchemaUnderSubjectUsingContexts(
           subject, schema, normalize, lookupDeletedSchema);
-
-      // If first attempt failed with normalize=false, try again with normalize=true
-      if (matchingSchema == null && !normalize) {
-        log.debug("No matching schema found with normalize = false,"
-                + " retrying with normalize = true");
-        matchingSchema = schemaRegistry.lookUpSchemaUnderSubjectUsingContexts(
-              subject, schema, true, lookupDeletedSchema);
-      }
-
       if (matchingSchema == null) {
         if (!schemaRegistry.hasSubjects(subject, lookupDeletedSchema)) {
           throw Errors.subjectNotFoundException(subject);

core/src/test/java/io/confluent/kafka/schemaregistry/rest/RestApiTest.java

@@ -1423,39 +1423,6 @@ public void testSchemaNormalization() throws Exception {
     assertTrue(!messages.isEmpty() && messages.get(0).contains("Invalid schema"));
   }
 
-  @Test
-  public void testLookUpSchemaWithNormalizationRetry() throws Exception {
-      String subject = "testSubject";
-
-      String schemaString1 = "{\"type\":\"record\",\"name\":\"User\",\"fields\":[{\"name\":\"id\",\"type\":\"int\"},{\"name\":\"email4\",\"type\":\"string\"}]}";
-  
-      // Register the original schema
-      TestUtils.registerAndVerifySchema(restApp.restClient, schemaString1, 1, subject);
-  
-      // Same schema with different field ordering (semantically equivalent)
-      String schemaString2 = "{\"type\":\"record\",\"name\":\"User\",\"fields\":[{\"type\":\"int\",\"name\":\"id\"},{\"type\":\"string\",\"name\":\"email4\"}]}";
-  
-      RegisterSchemaRequest request = new RegisterSchemaRequest();
-      request.setSchema(schemaString2);
-
-      io.confluent.kafka.schemaregistry.client.rest.entities.Schema schema = 
-          restApp.restClient.lookUpSubjectVersion(request, subject, false, false);
-      assertNotNull(schema);
-      assertEquals(1, schema.getVersion().intValue());
-
-      // Different schema with different field name (not semantically equivalent)
-      String invalidSchema = "{\"type\":\"record\",\"name\":\"User\",\"fields\":[{\"type\":\"int\",\"name\":\"id\"},{\"type\":\"string\",\"name\":\"email6\"}]}";
-      request.setSchema(invalidSchema);
-  
-      // Should fail since schemas are actually different
-      try {
-          restApp.restClient.lookUpSubjectVersion(request, subject, true, false);
-          fail("Should fail as schemas are not semantically equivalent");
-      } catch (RestClientException e) {
-          assertEquals(Errors.SCHEMA_NOT_FOUND_ERROR_CODE, e.getErrorCode());
-      }
-  }
-
   @Test
   public void testBad() throws Exception {
     String subject1 = "testTopic1";