Merge branch '8.1.x' into master by rayokota

Author: semaphore-agent-production[bot]

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

@@ -258,6 +258,10 @@ public Schema copy(Integer version, Integer id) {
         ruleSet, schema, schemaTags, timestamp, deleted);
   }
 
+  public Schema toHashKey() {
+    return new Schema(subject, null, null, schemaType, references, metadata, ruleSet, schema);
+  }
+
   @io.swagger.v3.oas.annotations.media.Schema(description = SUBJECT_DESC, example = SUBJECT_EXAMPLE)
   @JsonProperty("subject")
   public String getSubject() {

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

@@ -1714,8 +1714,8 @@ public ParsedSchema parseSchema(
           boolean normalize) throws InvalidSchemaException {
     try {
       ParsedSchema parsedSchema = isNew
-          ? newSchemaCache.get(new RawSchema(schema.copy(), isNew, normalize))
-          : oldSchemaCache.get(new RawSchema(schema.copy(), isNew, normalize));
+          ? newSchemaCache.get(new RawSchema(schema.toHashKey(), isNew, normalize))
+          : oldSchemaCache.get(new RawSchema(schema.toHashKey(), isNew, normalize));
       if (schema.getVersion() != null) {
         parsedSchema = parsedSchema.copy(schema.getVersion());
       }
@@ -2323,6 +2323,16 @@ 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));
+  }
+
+  public void invalidateFromOldSchemaCache(Schema schema) {
+    oldSchemaCache.invalidate(new RawSchema(schema, false, false));
+    oldSchemaCache.invalidate(new RawSchema(schema, false, true));
+  }
+
   public void clearNewSchemaCache() {
     newSchemaCache.invalidateAll();
   }

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

@@ -156,12 +156,12 @@ private void handleDeleteSubject(DeleteSubjectValue deleteSubjectValue) {
           schemaValue.setDeleted(true);
           SchemaValue oldSchemaValue = (SchemaValue) lookupCache.put(schemaKey, schemaValue);
           lookupCache.schemaDeleted(schemaKey, schemaValue, oldSchemaValue);
+          schemaRegistry.invalidateFromNewSchemaCache(schemaValue.toHashKey());
         }
       } catch (StoreException e) {
         log.error("Failed to delete subject {} in the local cache", subject, e);
       }
     }
-    schemaRegistry.clearNewSchemaCache();
   }
 
   private void handleClearSubject(ClearSubjectValue clearSubjectValue) {
@@ -185,7 +185,7 @@ private void handleSchemaUpdate(SchemaKey schemaKey,
 
       if (schemaValue.isDeleted()) {
         lookupCache.schemaDeleted(schemaKey, schemaValue, oldSchemaValue);
-        schemaRegistry.clearNewSchemaCache();
+        schemaRegistry.invalidateFromNewSchemaCache(schemaValue.toHashKey());
         updateMetrics(metricsContainer.getSchemasDeleted(),
                       metricsContainer.getSchemasDeleted(getSchemaType(schemaValue)));
       } else {
@@ -195,6 +195,7 @@ private void handleSchemaUpdate(SchemaKey schemaKey,
       }
     } else {
       lookupCache.schemaTombstoned(schemaKey, oldSchemaValue);
+      // Need to clear entire cache until we can prevent hard deleting referenced schemas
       schemaRegistry.clearOldSchemaCache();
     }
   }

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

@@ -333,4 +333,19 @@ public Schema toSchemaEntity() {
         isDeleted()
     );
   }
+
+  public Schema toHashKey() {
+    return new Schema(
+        getSubject(),
+        null,
+        null,
+        getSchemaType(),
+        getReferences() == null ? null : getReferences().stream()
+            .map(SchemaReference::toRefEntity)
+            .collect(Collectors.toList()),
+        getMetadata() == null ? null : getMetadata().toMetadataEntity(),
+        getRuleSet() == null ? null : getRuleSet().toRuleSetEntity(),
+        getSchema()
+    );
+  }
 }