Fix subject redirection from message on large registries

client/src/containers/Topic/Topic/TopicData/TopicData.jsx

@@ -603,9 +603,9 @@ class TopicData extends Root {
   }
 
   _redirectToSchema(id) {
-    const { selectedCluster } = this.state;
+    const { selectedCluster, selectedTopic } = this.state;
 
-    this.getApi(uriSchemaId(selectedCluster, id)).then(response => {
+    this.getApi(uriSchemaId(selectedCluster, id, selectedTopic)).then(response => {
       if (response.data) {
         this.props.history.push({
           pathname: `/ui/${selectedCluster}/schema/details/${response.data.subject}`,

client/src/utils/endpoints.js

@@ -209,8 +209,8 @@ export const uriSchemaRegistry = (clusterId, search, pageNumber) => {
   return `${apiUrl}/${clusterId}/schema?&search=${search}&page=${pageNumber}`;
 };
 
-export const uriSchemaId = (clusterId, id) => {
-  return `${apiUrl}/${clusterId}/schema/id/${id}`;
+export const uriSchemaId = (clusterId, id, topic) => {
+  return `${apiUrl}/${clusterId}/schema/id/${id}?topic=${topic}`;
 };
 
 export const uriSchemaVersions = (clusterId, subject) => {

src/main/java/org/akhq/controllers/SchemaController.java

@@ -158,16 +158,19 @@ private Schema registerSchema(String cluster, @Body Schema schema) throws IOExce
 
     @Get("api/{cluster}/schema/id/{id}")
     @Operation(tags = {"schema registry"}, summary = "Find a schema by id")
-    public Schema redirectId(
+    public Schema getSubjectBySchemaIdAndTopic(
         HttpRequest<?> request,
         String cluster,
-        Integer id
-    ) throws IOException, RestClientException, ExecutionException, InterruptedException {
+        Integer id,
+        @QueryValue String topic
+    ) throws IOException, RestClientException {
         // TODO Do the check on the subject name too
         checkIfClusterAllowed(cluster);
 
-        return this.schemaRepository
-            .getById(cluster, id)
+        return this.schemaRepository.getSubjectsBySchemaId(cluster, id)
+            .stream()
+            .filter(s -> s.getSubject().contains(topic))
+            .findFirst()
             .orElse(null);
     }
 

src/main/java/org/akhq/models/Schema.java

@@ -41,6 +41,12 @@ public class Schema {
 
     private String exception;
 
+    public Schema(int schemaId, String subject, int version) {
+        this.id = schemaId;
+        this.subject = subject;
+        this.version = version;
+    }
+
     public Schema(Schema schema, Schema.Config config) {
         this.id = schema.id;
         this.subject = schema.subject;

src/main/java/org/akhq/repositories/SchemaRegistryRepository.java

@@ -110,6 +110,20 @@ public boolean exist(String clusterId, String subject) throws IOException, RestC
         return found;
     }
 
+    public List<Schema> getSubjectsBySchemaId(String clusterId, int id) throws  IOException, RestClientException {
+        Optional<RestService> maybeRegistryRestClient = Optional.ofNullable(kafkaModule
+            .getRegistryRestClient(clusterId));
+        if(maybeRegistryRestClient.isEmpty()){
+            return List.of();
+        }
+
+        return maybeRegistryRestClient.get()
+            .getAllVersionsById(id)
+            .stream()
+            .map(v -> new Schema(id, v.getSubject(), v.getVersion()))
+            .collect(Collectors.toList());
+    }
+
     public Optional<Schema> getById(String clusterId, Integer id) throws IOException, RestClientException, ExecutionException, InterruptedException {
         for (String subject: this.all(clusterId, Optional.empty(), List.of())) {
             for (Schema version: this.getAllVersions(clusterId, subject)) {