decoding of AMQP-encoded header values

Author: d-rk

CHANGELOG.md

@@ -7,6 +7,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 - [#306](https://github.com/deviceinsight/kafkactl/pull/306) Support building arm image by updating to goreleaser v2 
 - new flag `--print-all` for consume command which combines the other print parameters into one
+- decoding of AMQP-encoded header values
 
 ## 5.15.0 - 2025-11-13
 

internal/consume/MessageDeserializer.go

@@ -4,6 +4,7 @@ import (
 	"encoding/base64"
 	"encoding/hex"
 	"sort"
+	"time"
 	"unicode/utf8"
 
 	"github.com/IBM/sarama"
@@ -62,14 +63,63 @@ func encodeRecordHeaders(headers []*sarama.RecordHeader) map[string]string {
 		}
 
 		key := *encodeBytes(header.Key, "")
-		value := *encodeBytes(header.Value, "")
+		var value string
+		if encodedValue := encodeBytes(decodeAMQPValue(header.Value), ""); encodedValue != nil {
+			value = *encodedValue
+		}
 
 		data[key] = value
 	}
 
 	return data
 }
 
+// decodeAMQPValue attempts to decode AMQP-encoded header values.
+// Azure Event Hubs wraps Kafka header values in AMQP type descriptors when using the Kafka interface.
+// This not moved to the azure plugin, because of the required RPC overhead per header value.
+// For regular Kafka servers, this function returns the input unchanged.
+func decodeAMQPValue(data []byte) []byte {
+	if len(data) < 2 {
+		return data
+	}
+
+	switch data[0] {
+	case 0xa1: // str8: string with 1-byte length
+		length := int(data[1])
+		if len(data) >= 2+length {
+			return data[2 : 2+length]
+		}
+	case 0xb1: // str32: string with 4-byte length
+		if len(data) >= 6 {
+			length := int(data[1])<<24 | int(data[2])<<16 | int(data[3])<<8 | int(data[4])
+			if len(data) >= 5+length {
+				return data[5 : 5+length]
+			}
+		}
+	case 0xa0: // vbin8: binary with 1-byte length
+		length := int(data[1])
+		if len(data) >= 2+length {
+			return data[2 : 2+length]
+		}
+	case 0xb0: // vbin32: binary with 4-byte length
+		if len(data) >= 6 {
+			length := int(data[1])<<24 | int(data[2])<<16 | int(data[3])<<8 | int(data[4])
+			if len(data) >= 5+length {
+				return data[5 : 5+length]
+			}
+		}
+	case 0x83: // timestamp: 64-bit milliseconds since Unix epoch
+		if len(data) >= 9 {
+			ms := int64(data[1])<<56 | int64(data[2])<<48 | int64(data[3])<<40 | int64(data[4])<<32 |
+				int64(data[5])<<24 | int64(data[6])<<16 | int64(data[7])<<8 | int64(data[8])
+			t := time.UnixMilli(ms)
+			return []byte(t.Format(time.RFC3339))
+		}
+	}
+
+	return data
+}
+
 func toSortedArray(headers map[string]string) []string {
 
 	var column []string