feat(topicdata): display __consumer_offsets contents (#856)

close #654
tchiotludo · Oct 16, 2021 · ed4d391 · ed4d391
1 parent 82bb7b5
commit ed4d391
Show file tree

Hide file tree

Showing 2 changed files with 24 additions and 0 deletions.
diff --git a/build.gradle b/build.gradle
@@ -81,6 +81,7 @@ dependencies {
     implementation "io.micronaut.cache:micronaut-cache-core"
 
     // kafka
+    implementation group: "org.apache.kafka", name: "kafka_" + kafkaScalaVersion, version: kafkaVersion
     implementation group: "org.apache.kafka", name: "kafka-clients", version: kafkaVersion
     implementation group: "io.confluent", name: "kafka-schema-registry-client", version: confluentVersion
     implementation group: "io.confluent", name: "kafka-avro-serializer", version: confluentVersion

diff --git a/src/main/java/org/akhq/models/Record.java b/src/main/java/org/akhq/models/Record.java
@@ -8,6 +8,7 @@
 import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient;
 import io.confluent.kafka.schemaregistry.json.JsonSchema;
 import io.confluent.kafka.schemaregistry.protobuf.ProtobufSchema;
+import kafka.coordinator.group.GroupMetadataManager;
 import lombok.*;
 import org.akhq.configs.SchemaRegistryType;
 import org.akhq.utils.AvroToJsonDeserializer;
@@ -189,6 +190,28 @@ private String convertToString(byte[] payload, Integer schemaId, boolean isKey)
 
                 return new String(payload);
             }
+        } else if (topic.isInternalTopic() && topic.getName().equals("__consumer_offsets")) {
+            if (isKey) {
+                try {
+                    return GroupMetadataManager.readMessageKey(ByteBuffer.wrap(payload)).key().toString();
+                } catch (Exception exception) {
+                    this.exceptions.add(Optional.ofNullable(exception.getMessage())
+                            .filter(msg -> !msg.isBlank())
+                            .orElseGet(() -> exception.getClass().getCanonicalName()));
+
+                    return new String(payload);
+                }
+            } else {
+                try {
+                    return GroupMetadataManager.readOffsetMessageValue(ByteBuffer.wrap(payload)).toString();
+                } catch (Exception exception) {
+                    this.exceptions.add(Optional.ofNullable(exception.getMessage())
+                        .filter(msg -> !msg.isBlank())
+                        .orElseGet(() -> exception.getClass().getCanonicalName()));
+
+                    return new String(payload);
+                }
+            }
         } else {
             if (protobufToJsonDeserializer != null) {
                 try {