Merge branch '8.0.x' into 8.1.x by rayokota

Author: semaphore-agent-production[bot]

client/src/main/java/io/confluent/kafka/schemaregistry/client/rest/entities/Schema.java

@@ -259,7 +259,14 @@ public Schema copy(Integer version, Integer id) {
   }
 
   public Schema toHashKey() {
-    return new Schema(subject, null, null, schemaType, references, metadata, ruleSet, schema);
+    // Deep copy the references list if it's not null
+    List<SchemaReference> referencesCopy = references != null
+        ? references.stream()
+        .map(SchemaReference::copy)
+        .collect(Collectors.toList())
+        : null;
+
+    return new Schema(subject, null, null, schemaType, referencesCopy, metadata, ruleSet, schema);
   }
 
   @io.swagger.v3.oas.annotations.media.Schema(description = SUBJECT_DESC, example = SUBJECT_EXAMPLE)

core/src/main/java/io/confluent/kafka/schemaregistry/storage/KafkaSchemaRegistry.java

@@ -2324,14 +2324,24 @@ private CloseableIterator<SchemaRegistryValue> allVersionsFromAllContexts(
     return new DelegatingIterator<>(versions.iterator());
   }
 
-  public void invalidateFromNewSchemaCache(Schema schema) {
-    newSchemaCache.invalidate(new RawSchema(schema, true, false));
-    newSchemaCache.invalidate(new RawSchema(schema, true, true));
+  /**
+   * Invalidate the cached parsed schema for a new schema.
+   *
+   * @param schemaKey a schema key obtained by {@link Schema#toHashKey()}
+   */
+  public void invalidateFromNewSchemaCache(Schema schemaKey) {
+    newSchemaCache.invalidate(new RawSchema(schemaKey, true, false));
+    newSchemaCache.invalidate(new RawSchema(schemaKey, true, true));
   }
 
-  public void invalidateFromOldSchemaCache(Schema schema) {
-    oldSchemaCache.invalidate(new RawSchema(schema, false, false));
-    oldSchemaCache.invalidate(new RawSchema(schema, false, true));
+  /**
+   * Invalidate the cached parsed schema for an old schema.
+   *
+   * @param schemaKey a schema key obtained by {@link Schema#toHashKey()}
+   */
+  public void invalidateFromOldSchemaCache(Schema schemaKey) {
+    oldSchemaCache.invalidate(new RawSchema(schemaKey, false, false));
+    oldSchemaCache.invalidate(new RawSchema(schemaKey, false, true));
   }
 
   public void clearNewSchemaCache() {