Bug: Kafka ProtobufDeserializer reads Confluent message indexes as plain varints, dropping fields or failing for non-first message types #8665

import base64

from google.protobuf.descriptor_pb2 import FieldDescriptorProto

from aws_lambda_powertools.utilities.kafka import ConsumerRecords, SchemaConfig, kafka_consumer


def confluent_framed(message, indexes):
    # Message index array as confluent_kafka's ProtobufSerializer writes it: zigzag varints, [0] as a single 0x00
    def zigzag(v):
        v, out = (v << 1) ^ (v >> 63), bytearray()
        while v & ~0x7F:
            out.append((v & 0x7F) | 0x80)
            v >>= 7
        return bytes(out + bytes([v]))

    prefix = b"\x00" if indexes == [0] else zigzag(len(indexes)) + b"".join(zigzag(i) for i in indexes)
    return base64.b64encode(prefix + message.SerializeToString()).decode()


message = FieldDescriptorProto(name="order_id", number=1, json_name="orderId")


@kafka_consumer(schema_config=SchemaConfig(value_schema_type="PROTOBUF", value_schema=FieldDescriptorProto))
def handler(event: ConsumerRecords, context):
    return next(event.records).value


for indexes in ([0], [1, 0], [2]):
    event = {
        "eventSource": "aws:kafka",
        "records": {
            "orders-0": [
                {
                    "topic": "orders", "partition": 0, "offset": 1, "timestamp": 1, "timestampType": "CREATE_TIME",
                    "key": None, "headers": [],
                    "value": confluent_framed(message, indexes),
                    "valueSchemaMetadata": {"schemaId": "1", "dataFormat": "PROTOBUF"},
                },
            ],
        },
    }
    try:
        print(f"message index {indexes}: {handler(event, None)}")
    except Exception as exc:
        print(f"message index {indexes}: {type(exc).__name__}: {exc}")

Output of the snippet on develop:

message index [0]: {'name': 'order_id', 'number': 1, 'json_name': 'orderId'}
message index [1, 0]: KafkaConsumerDeserializationError: Error trying to deserialize protobuf data - DecodeError: Error parsing message with type 'google.protobuf.FieldDescriptorProto': Wire format was corrupt
message index [2]: KafkaConsumerDeserializationError: Error trying to deserialize protobuf data - DecodeError: Error parsing message with type 'google.protobuf.FieldDescriptorProto': Wire format was corrupt