[LI-HOTFIX] KAFKA-9855 - return cached Structs for Schemas with no fi…

…elds

TICKET =
LI_DESCERIPTION = committing to our fork until upstream picks this up
EXIT_CRITERIA = when KAFKA-9855 is accepted upstream (specifically - apache#8472)

Signed-off-by: radai-rosenblatt <radai.rosenblatt@gmail.com>

clients/src/main/java/org/apache/kafka/common/protocol/types/Schema.java

@@ -25,10 +25,12 @@
  * The schema for a compound record definition
  */
 public class Schema extends Type {
+    private final static Object[] NO_VALUES = new Object[0];
 
     private final BoundField[] fields;
     private final Map<String, BoundField> fieldsByName;
     private final boolean tolerateMissingFieldsWithDefaults;
+    private final Struct cachedStruct;
 
     /**
      * Construct the schema with a given list of its field values
@@ -62,6 +64,9 @@ public Schema(boolean tolerateMissingFieldsWithDefaults, Field... fs) {
             this.fields[i] = new BoundField(def, this, i);
             this.fieldsByName.put(def.name, this.fields[i]);
         }
+        //6 schemas have no fields at the time of this writing (3 versions each of list_groups and api_versions)
+        //for such schemas there's no point in even creating a unique Struct object when deserializing.
+        this.cachedStruct = this.fields.length > 0 ? null : new Struct(this, NO_VALUES);
     }
 
     /**
@@ -90,6 +95,9 @@ public void write(ByteBuffer buffer, Object o) {
      */
     @Override
     public Struct read(ByteBuffer buffer) {
+        if (cachedStruct != null) {
+            return cachedStruct;
+        }
         Object[] objects = new Object[fields.length];
         for (int i = 0; i < fields.length; i++) {
             try {