DGS-19689 Add support for schema guids that are stable across contexts (#3551)

Author: rayokota

.gitignore

@@ -23,3 +23,6 @@ target/
 # Log file from schema registry daemon
 logs/
 nohup.out
+
+# jenv version
+.java-version

client/src/main/java/io/confluent/kafka/schemaregistry/client/SchemaMetadata.java

@@ -24,12 +24,14 @@
 
 import io.confluent.kafka.schemaregistry.avro.AvroSchema;
 import io.confluent.kafka.schemaregistry.client.rest.entities.SchemaReference;
+import java.util.Objects;
 
 public class SchemaMetadata {
 
   private String subject;
   private int id;
   private int version;
+  private String guid;
   private String schemaType;
   private String schema;
   private List<SchemaReference> references;
@@ -59,7 +61,9 @@ public SchemaMetadata(int id,
   public SchemaMetadata(Schema schema) {
     this.subject = schema.getSubject();
     this.id = schema.getId();
+    this.guid = schema.getGuid();
     this.version = schema.getVersion();
+    this.guid = schema.getGuid();
     this.schemaType = schema.getSchemaType();
     this.schema = schema.getSchema();
     this.references = schema.getReferences();
@@ -79,6 +83,10 @@ public int getVersion() {
     return version;
   }
 
+  public String getGuid() {
+    return guid;
+  }
+
   public String getSchemaType() {
     return schemaType;
   }
@@ -98,4 +106,27 @@ public Metadata getMetadata() {
   public RuleSet getRuleSet() {
     return this.ruleSet;
   }
+
+  @Override
+  public boolean equals(Object o) {
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    SchemaMetadata that = (SchemaMetadata) o;
+    return Objects.equals(subject, that.subject)
+        && id == that.id
+        && version == that.version
+        && Objects.equals(guid, that.guid)
+        && Objects.equals(schemaType, that.schemaType)
+        && Objects.equals(schema, that.schema)
+        && Objects.equals(references, that.references)
+        && Objects.equals(metadata, that.metadata)
+        && Objects.equals(ruleSet, that.ruleSet);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(
+        subject, id, version, guid, schemaType, schema, references, metadata, ruleSet);
+  }
 }

client/src/main/java/io/confluent/kafka/schemaregistry/client/rest/RestService.java

@@ -25,6 +25,7 @@
 import com.google.common.io.CharStreams;
 import io.confluent.kafka.schemaregistry.client.SchemaRegistryClientConfig;
 import io.confluent.kafka.schemaregistry.client.rest.entities.Config;
+import io.confluent.kafka.schemaregistry.client.rest.entities.ContextId;
 import io.confluent.kafka.schemaregistry.client.rest.entities.ErrorMessage;
 import io.confluent.kafka.schemaregistry.client.rest.entities.Schema;
 import io.confluent.kafka.schemaregistry.client.rest.entities.SchemaRegistryServerVersion;
@@ -129,6 +130,9 @@ public class RestService implements Closeable, Configurable {
   private static final TypeReference<List<SubjectVersion>> GET_VERSIONS_RESPONSE_TYPE =
       new TypeReference<List<SubjectVersion>>() {
       };
+  private static final TypeReference<List<ContextId>> GET_IDS_RESPONSE_TYPE =
+      new TypeReference<List<ContextId>>() {
+      };
   private static final TypeReference<CompatibilityCheckResponse>
       COMPATIBILITY_CHECK_RESPONSE_TYPE_REFERENCE =
       new TypeReference<CompatibilityCheckResponse>() {
@@ -1410,6 +1414,39 @@ public List<SubjectVersion> getAllVersionsById(Map<String, String> requestProper
     return response;
   }
 
+  public SchemaString getByGuid(String guid, String format)
+      throws IOException, RestClientException {
+    return getByGuid(DEFAULT_REQUEST_PROPERTIES, guid, format);
+  }
+
+  public SchemaString getByGuid(Map<String, String> requestProperties, String guid, String format)
+      throws IOException, RestClientException {
+    UriBuilder builder = UriBuilder.fromPath("/schemas/guids/{guid}");
+    if (format != null) {
+      builder.queryParam("format", format);
+    }
+    String path = builder.build(guid).toString();
+
+    SchemaString response = httpRequest(path, "GET", null, requestProperties,
+        GET_SCHEMA_BY_ID_RESPONSE_TYPE);
+    return response;
+  }
+
+  public List<ContextId> getAllContextIds(String guid) throws IOException, RestClientException {
+    return getAllContextIds(DEFAULT_REQUEST_PROPERTIES, guid);
+  }
+
+  public List<ContextId> getAllContextIds(Map<String, String> requestProperties, String guid)
+      throws IOException, RestClientException {
+    UriBuilder builder = UriBuilder.fromPath("/schemas/guids/{guid}/ids");
+    String path = builder.build(guid).toString();
+
+    List<ContextId> response = httpRequest(path, "GET", null, requestProperties,
+        GET_IDS_RESPONSE_TYPE);
+
+    return response;
+  }
+
   public Integer deleteSchemaVersion(
       Map<String, String> requestProperties,
       String subject,

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

@@ -0,0 +1,95 @@
+/*
+ * Copyright 2025 Confluent Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.confluent.kafka.schemaregistry.client.rest.entities;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.Objects;
+
+@JsonInclude(JsonInclude.Include.NON_EMPTY)
+@JsonIgnoreProperties(ignoreUnknown = true)
+@io.swagger.v3.oas.annotations.media.Schema(description = "Context ID pair")
+public class ContextId implements Comparable<ContextId> {
+
+  private String context;
+  private Integer id;
+
+  @JsonCreator
+  public ContextId(@JsonProperty("context") String context,
+                   @JsonProperty("id") Integer id) {
+    this.context = context;
+    this.id = id;
+  }
+
+  @JsonProperty("context")
+  public String getContext() {
+    return context;
+  }
+
+  @JsonProperty("context")
+  public void setContext(String context) {
+    this.context = context;
+  }
+
+  @JsonProperty("id")
+  public Integer getId() {
+    return this.id;
+  }
+
+  @JsonProperty("id")
+  public void setId(Integer id) {
+    this.id = id;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    ContextId schema1 = (ContextId) o;
+    return Objects.equals(context, schema1.context)
+        && Objects.equals(id, schema1.id);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(context, id);
+  }
+
+  @Override
+  public String toString() {
+    StringBuilder sb = new StringBuilder();
+    sb.append("{context=" + this.context + ",");
+    sb.append("id=" + this.id + "}");
+    return sb.toString();
+  }
+
+  @Override
+  public int compareTo(ContextId that) {
+    int result = this.context.compareTo(that.context);
+    if (result != 0) {
+      return result;
+    }
+    result = this.id - that.id;
+    return result;
+  }
+}

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

@@ -34,21 +34,23 @@ public class ExtendedSchema extends Schema {
 
   @JsonCreator
   public ExtendedSchema(@JsonProperty("subject") String subject,
-                @JsonProperty("version") Integer version,
-                @JsonProperty("id") Integer id,
-                @JsonProperty("schemaType") String schemaType,
-                @JsonProperty("references") List<SchemaReference> references,
-                @JsonProperty("metadata") Metadata metadata,
-                @JsonProperty("ruleset") RuleSet ruleSet,
-                @JsonProperty("schema") String schema,
-                @JsonProperty("aliases") List<String> aliases) {
-    super(subject, version, id, schemaType, references, metadata, ruleSet, schema);
+      @JsonProperty("version") Integer version,
+      @JsonProperty("id") Integer id,
+      @JsonProperty("guid") String guid,
+      @JsonProperty("schemaType") String schemaType,
+      @JsonProperty("references") List<SchemaReference> references,
+      @JsonProperty("metadata") Metadata metadata,
+      @JsonProperty("ruleset") RuleSet ruleSet,
+      @JsonProperty("schema") String schema,
+      @JsonProperty("aliases") List<String> aliases) {
+    super(subject, version, id, guid, schemaType, references, metadata, ruleSet, schema);
     this.aliases = aliases;
   }
 
   public ExtendedSchema(Schema schema, List<String> aliases) {
-    super(schema.getSubject(), schema.getVersion(), schema.getId(), schema.getSchemaType(),
-        schema.getReferences(), schema.getMetadata(), schema.getRuleSet(), schema.getSchema());
+    super(schema.getSubject(), schema.getVersion(), schema.getId(), schema.getGuid(),
+        schema.getSchemaType(), schema.getReferences(), schema.getMetadata(), schema.getRuleSet(),
+        schema.getSchema());
     this.aliases = aliases;
   }
 
@@ -93,6 +95,7 @@ public String toString() {
     sb.append("{subject=" + getSubject() + ",");
     sb.append("version=" + getVersion() + ",");
     sb.append("id=" + getId() + ",");
+    sb.append("guid=" + getGuid() + ",");
     sb.append("schemaType=" + getSchemaType() + ",");
     sb.append("references=" + getReferences() + ",");
     sb.append("metadata=" + getMetadata() + ",");