KAFKA-6342; Remove unused workaround for JSON parsing of non-escaped strings (#8591)

Previously we had fallback logic when parsing ACLs to handle older entries which may contain non-escaped characters. This code became dead after 1.1 since it was no longer used in the parsing of ACLs. This patch removes the fallback logic.

Reviewers: Ismael Juma <ismael@juma.me.uk>, Jason Gustafson <jason@confluent.io>

core/src/main/scala/kafka/utils/Json.scala

@@ -35,16 +35,7 @@ object Json {
    */
   def parseFull(input: String): Option[JsonValue] =
     try Option(mapper.readTree(input)).map(JsonValue(_))
-    catch {
-      case _: JsonProcessingException =>
-        // Before 1.0.1, Json#encode did not escape backslash or any other special characters. SSL principals
-        // stored in ACLs may contain backslash as an escape char, making the JSON generated in earlier versions invalid.
-        // Escape backslash and retry to handle these strings which may have been persisted in ZK.
-        // Note that this does not handle all special characters (e.g. non-escaped double quotes are not supported)
-        val escapedInput = input.replaceAll("\\\\", "\\\\\\\\")
-        try Option(mapper.readTree(escapedInput)).map(JsonValue(_))
-        catch { case _: JsonProcessingException => None }
-    }
+    catch { case _: JsonProcessingException => None }
 
   /**
    * Parse a JSON string into either a generic type T, or a JsonProcessingException in the case of

core/src/test/scala/unit/kafka/security/authorizer/AclEntryTest.scala

@@ -23,15 +23,14 @@ import org.apache.kafka.common.acl.AclOperation.READ
 import org.apache.kafka.common.acl.AclPermissionType.{ALLOW, DENY}
 import org.apache.kafka.common.security.auth.KafkaPrincipal
 import org.junit.{Assert, Test}
-import org.scalatestplus.junit.JUnitSuite
 
 import scala.jdk.CollectionConverters._
 
-class AclEntryTest extends JUnitSuite {
+class AclEntryTest {
 
-  val AclJson = "{\"version\": 1, \"acls\": [{\"host\": \"host1\",\"permissionType\": \"Deny\",\"operation\": \"READ\", \"principal\": \"User:alice\"  },  " +
-    "{  \"host\":  \"*\" ,  \"permissionType\": \"Allow\",  \"operation\":  \"Read\", \"principal\": \"User:bob\"  },  " +
-    "{  \"host\": \"host1\",  \"permissionType\": \"Deny\",  \"operation\":   \"Read\" ,  \"principal\": \"User:bob\"}  ]}"
+  val AclJson = """{"version": 1, "acls": [{"host": "host1","permissionType": "Deny","operation": "READ", "principal": "User:alice"  },
+    {  "host":  "*" ,  "permissionType": "Allow",  "operation":  "Read", "principal": "User:bob"  },
+    {  "host": "host1",  "permissionType": "Deny",  "operation":   "Read" ,  "principal": "User:bob"}]}"""
 
   @Test
   def testAclJsonConversion(): Unit = {
@@ -40,10 +39,8 @@ class AclEntryTest extends JUnitSuite {
     val acl3 = AclEntry(new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "bob"), DENY, "host1", READ)
 
     val acls = Set[AclEntry](acl1, acl2, acl3)
-    val jsonAcls = Json.encodeAsBytes(AclEntry.toJsonCompatibleMap(acls).asJava)
 
-    Assert.assertEquals(acls, AclEntry.fromBytes(jsonAcls))
+    Assert.assertEquals(acls, AclEntry.fromBytes(Json.encodeAsBytes(AclEntry.toJsonCompatibleMap(acls).asJava)))
     Assert.assertEquals(acls, AclEntry.fromBytes(AclJson.getBytes(UTF_8)))
   }
-
 }

core/src/test/scala/unit/kafka/utils/JsonTest.scala

@@ -59,10 +59,6 @@ class JsonTest {
     val encoded = Json.legacyEncodeAsString(map)
     val decoded = Json.parseFull(encoded)
     assertEquals(Json.parseFull("""{"foo1":"bar1\\,bar2", "foo2":"\\bar"}"""), decoded)
-
-    // Test strings with non-escaped backslash and quotes. This is to verify that ACLs
-    // containing non-escaped chars persisted using 1.0 can be parsed.
-    assertEquals(decoded, Json.parseFull("""{"foo1":"bar1\,bar2", "foo2":"\bar"}"""))
   }
 
   @Test