tchiotludo · tchiotludo · Feb 21, 2024 · Jan 12, 2024 · Jan 12, 2024 · Jan 12, 2024
diff --git a/client/public/index.html b/client/public/index.html
@@ -1,13 +1,13 @@
 <!DOCTYPE html>
 <html lang="en">
   <head>
-    <meta charset="utf-8">
-    <base href="%PUBLIC_URL%/">
-    <link rel="icon" href="%PUBLIC_URL%/favicon.ico">
-    <link rel="manifest" href="%PUBLIC_URL%/manifest.json">
-    <meta name="viewport" content="width=device-width, initial-scale=1">
-    <meta name="theme-color" content="#33b5e5">
-    <meta name="html-head" content="replace">
+    <meta charset="utf-8" />
+    <base href="%PUBLIC_URL%/" />
+    <link rel="icon" href="%PUBLIC_URL%/favicon.ico" />
+    <link rel="manifest" href="%PUBLIC_URL%/manifest.json" />
+    <meta name="viewport" content="width=device-width, initial-scale=1" />
+    <meta name="theme-color" content="#33b5e5" />
+    <meta name="html-head" content="replace" />
 
     <!--
       The AKHQ_PREFIX_PATH placeholder magic value is replaced during serving by server

diff --git a/client/src/containers/SideBar/Sidebar.jsx b/client/src/containers/SideBar/Sidebar.jsx
@@ -21,6 +21,7 @@ class Sidebar extends Component {
     showConnects: false,
     showKsqlDBs: false,
     enableRegistry: false,
+    registryType: '',
     enableConnect: false,
     enableKsqlDB: false,
     roles: JSON.parse(sessionStorage.getItem('roles')),
@@ -95,6 +96,7 @@ class Sidebar extends Component {
     const enableKsqlDB = cluster.ksqldbs !== undefined;
     let newState = {
       enableRegistry: cluster.registry,
+      registryType: cluster.registryType,
       enableConnect: enableConnects,
       allConnects: [],
       selectedConnect: '',
@@ -284,6 +286,7 @@ class Sidebar extends Component {
       selectedTab,
       height,
       enableRegistry,
+      registryType,
       enableConnect,
       enableKsqlDB
     } = this.state;
@@ -355,6 +358,7 @@ class Sidebar extends Component {
             roles.ACL.includes('READ') &&
             this.renderMenuItem('fa fa-fw fa-key', constants.ACLS, 'ACLS')}
           {enableRegistry &&
+            registryType !== 'GLUE' &&
             roles &&
             roles.SCHEMA &&
             roles.SCHEMA.includes('READ') &&

diff --git a/client/src/containers/Topic/Topic/Topic.jsx b/client/src/containers/Topic/Topic/Topic.jsx
@@ -18,6 +18,7 @@ class Topic extends Root {
   state = {
     clusterId: this.props.clusterId,
     topicId: this.props.topicId,
+    registryType: '',
     topic: {},
     selectedTab: '',
     showDeleteModal: false,
@@ -50,11 +51,12 @@ class Topic extends Root {
 
     const roles = this.state.roles || {};
     const tabSelected = getSelectedTab(this.props, this.tabs);
-
+    const registryType = this.props.clusters.find(el => el.id === clusterId).registryType;
     this.setState(
       {
         clusterId,
         topicId,
+        registryType,
         selectedTab:
           roles.TOPIC_DATA && roles.TOPIC_DATA.includes('READ') ? tabSelected : 'configs',
         topicInternal: this.props.location.internal
@@ -210,6 +212,7 @@ class Topic extends Root {
             location={location}
             isAllTopicDataSelected={this.state.isAllTopicDataSelected}
             onSelectAllCheckboxChange={this._handleSelectAllCheckboxChange}
+            registryType={this.state.registryType}
           />
         );
       case 'partitions':

diff --git a/client/src/containers/Topic/Topic/TopicData/TopicData.jsx b/client/src/containers/Topic/Topic/TopicData/TopicData.jsx
@@ -68,6 +68,7 @@ class TopicData extends Root {
     deleteMessage: '',
     compactMessageToDelete: '',
     selectedCluster: this.props.clusterId,
+    registryType: this.props.registryType,
     selectedTopic: this.props.topicId,
     cleanupPolicy: '',
     datetime: '',
@@ -115,11 +116,11 @@ class TopicData extends Root {
     const { clusterId, topicId } = this.props.match.params;
     const query = new URLSearchParams(this.props.location.search);
     const uiOptions = await getClusterUIOptions(clusterId);
-
     this.setState(
       prevState => ({
         selectedCluster: clusterId,
         selectedTopic: topicId,
+        registryType: this.props.registryType,
         sortBy: query.get('sort')
           ? query.get('sort')
           : uiOptions && uiOptions.topicData && uiOptions.topicData.sort
@@ -517,7 +518,11 @@ class TopicData extends Root {
         partition: JSON.stringify(message.partition) || '',
         offset: JSON.stringify(message.offset) || '',
         headers: message.headers || [],
-        schema: { key: message.keySchemaId, value: message.valueSchemaId },
+        schema: {
+          key: message.keySchemaId,
+          value: message.valueSchemaId,
+          registryType: this.state.registryType
+        },
         exceptions: message.exceptions || []
       };
 
@@ -1206,9 +1211,13 @@ class TopicData extends Root {
                       {obj[col.accessor].key !== undefined && (
                         <span
                           className="badge badge-primary clickable"
-                          onClick={() => {
-                            this._redirectToSchema(obj.schema.key);
-                          }}
+                          onClick={
+                            obj[col.accessor].registryType !== 'GLUE'
+                              ? () => {
+                                  this._redirectToSchema(obj.schema.key);
+                                }
+                              : undefined
+                          }
                         >
                           Key: {obj[col.accessor].key}
                         </span>
@@ -1217,9 +1226,13 @@ class TopicData extends Root {
                       {obj[col.accessor].value !== undefined && (
                         <span
                           className="badge badge-primary clickable schema-value"
-                          onClick={() => {
-                            this._redirectToSchema(obj.schema.value);
-                          }}
+                          onClick={
+                            obj[col.accessor].registryType !== 'GLUE'
+                              ? () => {
+                                  this._redirectToSchema(obj.schema.value);
+                                }
+                              : undefined
+                          }
                         >
                           Value: {obj[col.accessor].value}
                         </span>

diff --git a/client/src/utils/Routes.js b/client/src/utils/Routes.js
@@ -180,7 +180,11 @@ class Routes extends Root {
               )}
 
               {roles && roles.TOPIC && roles.TOPIC.includes('READ') && (
-                <Route exact path="/ui/:clusterId/topic/:topicId/:tab?" component={Topic} />
+                <Route
+                  exact
+                  path="/ui/:clusterId/topic/:topicId/:tab?"
+                  render={props => <Topic clusters={clusters} {...props} />}
+                />
               )}
 
               {roles && roles.TOPIC && roles.TOPIC_DATA.includes('READ') && (

diff --git a/docs/.vuepress/config.js b/docs/.vuepress/config.js
@@ -43,7 +43,8 @@ module.exports = {
             text: 'Schema Registry',
             children: [
               '/docs/configuration/schema-registry/tibco.md',
-              '/docs/configuration/schema-registry/schema-references.md',
+                '/docs/configuration/schema-registry/glue.md',
+                '/docs/configuration/schema-registry/schema-references.md',
             ]
           },
           {

diff --git a/docs/docs/configuration/schema-registry/glue.md b/docs/docs/configuration/schema-registry/glue.md
@@ -0,0 +1,29 @@
+# Glue schema registry
+Currently ,glue schema registry support is limited to only de-serialisation of avro/protobuf/json serialized messages.
+It can be configured as below.
+```yaml
+  akhq:
+    environment:
+      AKHQ_CONFIGURATION: |
+        akhq:
+          connections:
+            docker-kafka-server:
+              properties:
+                bootstrap.servers: "kafka:9092"
+              schema-registry:
+                url: "http://schema-registry:8085"
+                type: "glue"
+                glueSchemaRegistryName: Name of schema Registry
+                awsRegion: aws region
+              connect:
+                - name: "connect"
+                  url: "http://connect:8083"
+    ports:
+      - 8080:8080
+    links:
+      - kafka
+      - repo
+```
+Please note that authentication is done using aws default credentials provider.
+
+Url key is required to not break the flow.
diff --git a/src/main/java/org/akhq/controllers/AkhqController.java b/src/main/java/org/akhq/controllers/AkhqController.java
@@ -73,6 +73,7 @@ public List<ClusterDefinition> list() {
             .map(connection -> new ClusterDefinition(
                 connection.getName(),
                 connection.getSchemaRegistry() != null,
+                connection.getSchemaRegistry() != null ? connection.getSchemaRegistry().getType().name() : null,
                 (connection.getConnect() != null ? connection.getConnect() : new ArrayList<Connect>())
                     .stream()
                     .map(Connect::getName)
@@ -279,6 +280,7 @@ public static class AuthPermissions {
     public static class ClusterDefinition {
         private String id;
         private boolean registry;
+        private String registryType;
         private List<String> connects;
         private List<String> ksqldbs;
     }

diff --git a/src/main/java/org/akhq/models/Record.java b/src/main/java/org/akhq/models/Record.java
@@ -1,5 +1,7 @@
 package org.akhq.models;
 
+import com.amazonaws.services.schemaregistry.deserializers.GlueSchemaRegistryDeserializerDataParser;
+import com.amazonaws.services.schemaregistry.deserializers.GlueSchemaRegistryKafkaDeserializer;
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.google.protobuf.Message;
@@ -42,9 +44,9 @@ public class Record {
     private ZonedDateTime timestamp;
     @JsonIgnore
     private TimestampType timestampType;
-    private Integer keySchemaId;
+    private String keySchemaId;
     private String keySubject;
-    private Integer valueSchemaId;
+    private String valueSchemaId;
     private String valueSubject;
     private List<KeyValue<String, String>> headers = new ArrayList<>();
     @JsonIgnore
@@ -89,22 +91,21 @@ public class Record {
     @JsonIgnore
     private Deserializer awsGlueKafkaDeserializer;
 
-
     public Record(RecordMetadata record, SchemaRegistryType schemaRegistryType, byte[] bytesKey, byte[] bytesValue, List<KeyValue<String, String>> headers, Topic topic, Deserializer awsGlueKafkaDeserializer) {
         this.MAGIC_BYTE = schemaRegistryType.getMagicByte();
         this.topic = topic;
         this.partition = record.partition();
         this.offset = record.offset();
         this.timestamp = ZonedDateTime.ofInstant(Instant.ofEpochMilli(record.timestamp()), ZoneId.systemDefault());
         this.bytesKey = bytesKey;
+        this.awsGlueKafkaDeserializer = awsGlueKafkaDeserializer;
         this.keySchemaId = getAvroSchemaId(this.bytesKey);
         this.keySubject = getAvroSchemaSubject(this.keySchemaId);
         this.bytesValue = bytesValue;
         this.valueSchemaId = getAvroSchemaId(this.bytesValue);
         this.valueSubject = getAvroSchemaSubject(this.valueSchemaId);
         this.headers = headers;
         this.truncated = false;
-        this.awsGlueKafkaDeserializer = awsGlueKafkaDeserializer;
     }
 
     public Record(SchemaRegistryClient client, ConsumerRecord<byte[], byte[]> record, SchemaRegistryType schemaRegistryType, Deserializer kafkaAvroDeserializer,
@@ -122,6 +123,7 @@ public Record(SchemaRegistryClient client, ConsumerRecord<byte[], byte[]> record
         this.timestamp = ZonedDateTime.ofInstant(Instant.ofEpochMilli(record.timestamp()), ZoneId.systemDefault());
         this.timestampType = record.timestampType();
         this.bytesKey = record.key();
+        this.awsGlueKafkaDeserializer = awsGlueKafkaDeserializer;
         this.keySchemaId = getAvroSchemaId(this.bytesKey);
         this.keySubject = getAvroSchemaSubject(this.keySchemaId);
         this.bytesValue = bytesValue;
@@ -139,7 +141,6 @@ public Record(SchemaRegistryClient client, ConsumerRecord<byte[], byte[]> record
         this.avroToJsonSerializer = avroToJsonSerializer;
         this.kafkaJsonDeserializer = kafkaJsonDeserializer;
         this.truncated = false;
-        this.awsGlueKafkaDeserializer = awsGlueKafkaDeserializer;
     }
 
     public String getKey() {
@@ -179,20 +180,18 @@ public void setTruncated(Boolean truncated) {
         this.truncated = truncated;
     }
 
-    private String convertToString(byte[] payload, Integer schemaId, boolean isKey) {
+    private String convertToString(byte[] payload, String schemaId, boolean isKey) {
         if (payload == null) {
             return null;
-        }
-        else if (this.awsGlueKafkaDeserializer != null) {
-            return this.awsGlueKafkaDeserializer.deserialize(this.topic.getName(), payload).toString();
-
         } else if (schemaId != null) {
             try {
 
                 Object toType = null;
-
+                if (this.awsGlueKafkaDeserializer != null) {
+                    return this.awsGlueKafkaDeserializer.deserialize(this.topic.getName(), payload).toString();
+                }
                 if (client != null) {
-                    ParsedSchema schema = client.getSchemaById(schemaId);
+                    ParsedSchema schema = client.getSchemaById(Integer.parseInt(schemaId));
                     if ( schema.schemaType().equals(ProtobufSchema.TYPE) ) {
                        toType = kafkaProtoDeserializer.deserialize(topic.getName(), payload);
                        if (!(toType instanceof Message)) {
@@ -304,30 +303,46 @@ public Collection<String> getHeadersValues() {
             .collect(Collectors.toList());
     }
 
-    private Integer getAvroSchemaId(byte[] payload) {
+    private String getAvroSchemaId(byte[] payload) {
         if (topic.isInternalTopic()) {
             return null;
         }
         try {
+
+            if (awsGlueKafkaDeserializer!= null) {
+                ByteBuffer byteBuffer = ByteBuffer.wrap(payload);
+                GlueSchemaRegistryDeserializerDataParser dataParser = GlueSchemaRegistryDeserializerDataParser.getInstance();
+
+                UUID schemaVersionId = dataParser.getSchemaVersionId(byteBuffer);
+                return schemaVersionId.toString();
+            }
+
             ByteBuffer buffer = ByteBuffer.wrap(payload);
             byte magicBytes = buffer.get();
             int schemaId = buffer.getInt();
 
             if (magicBytes == MAGIC_BYTE && schemaId >= 0) {
-                return schemaId;
+                return String.valueOf(schemaId);
             }
         } catch (Exception ignore) {
 
         }
         return null;
     }
 
-    private String getAvroSchemaSubject(Integer schemaId) {
+    private String getAvroSchemaSubject(String schemaId) {
         if (schemaId == null || client == null) {
             return null;
         }
         try {
-            ParsedSchema schemaById = client.getSchemaById(schemaId);
+            if(awsGlueKafkaDeserializer!= null) {
+                String[] schemaArnSplitted = ( (GlueSchemaRegistryKafkaDeserializer) awsGlueKafkaDeserializer)
+                    .getGlueSchemaRegistryDeserializationFacade()
+                    .getSchemaRegistryClient().getSchemaVersionResponse(schemaId).schemaArn().split("/");
+               return  schemaArnSplitted[schemaArnSplitted.length-1];
+            }
+
+            ParsedSchema schemaById = client.getSchemaById(Integer.parseInt(schemaId));
             if (schemaById == null) {
                 return null;
             }

diff --git a/src/test/java/org/akhq/controllers/AkhqControllerTest.java b/src/test/java/org/akhq/controllers/AkhqControllerTest.java
@@ -29,6 +29,7 @@ void list() {
         assertEquals(1, result.get(0).getKsqldbs().size());
         assertEquals("ksqldb", result.get(0).getKsqldbs().get(0));
         assertTrue(result.get(0).isRegistry());
+        assertEquals("CONFLUENT", result.get(0).getRegistryType());
     }
 
     @Test

diff --git a/src/test/java/org/akhq/repositories/RecordRepositoryTest.java b/src/test/java/org/akhq/repositories/RecordRepositoryTest.java
@@ -278,7 +278,7 @@ void produceAndConsumeRecordUsingJsonSchema() throws ExecutionException, Interru
         Record recordToAssert = consumedRecord.get();
         assertEquals(recordToAssert.getKey(), keyJsonString);
         assertEquals(recordToAssert.getValue(), recordAsJsonString);
-        assertEquals(recordToAssert.getValueSchemaId(), valueJsonSchema.getId());
+        assertEquals(recordToAssert.getValueSchemaId(), String.valueOf(valueJsonSchema.getId()));
 
         // clear schema registry as it is shared between tests
         schemaRegistryRepository.delete(KafkaTestCluster.CLUSTER_ID, keyJsonSchema.getSubject());