protected T deserialize(String topic, Headers headers, byte[] payload)
        throws SerializationException, InvalidConfigurationException {
        // Phase 1: Pre-validation
        if (payload == null) {
            return null;
        }
        if (schemaRegistry == null) {
            throw new InvalidConfigurationException("Schema registry not found, make sure the schema.registry.url is set");
        }

        int schemaId = 0;
        byte[] messageBytes;
        MessageIndexes indexes;
        Message message;

        try {
            // Phase 2: Message Header Parsing
            ByteBuffer buffer = processHeader(payload);
            schemaId = extractSchemaId(buffer);
            indexes = extractMessageIndexes(buffer);
            messageBytes = extractMessageBytes(buffer);

            // Phase 3: Schema Processing
            ProtobufSchemaWrapper protobufSchemaWrapper = processSchema(topic, schemaId, indexes);
            Descriptors.Descriptor targetDescriptor = protobufSchemaWrapper.getDescriptor();

            // Phase 4: Message Deserialization
            message = deserializeMessage(targetDescriptor, messageBytes);

            return (T) message;
        } catch (InterruptedIOException e) {
            throw new TimeoutException("Error deserializing Protobuf message for id " + schemaId, e);
        } catch (IOException | RuntimeException e) {
            throw new SerializationException("Error deserializing Protobuf message for id " + schemaId, e);
        }
    }
Deserialize protobuf message from the given byte array.
The implementation follows the open-closed principle by breaking down the
deserialization process into multiple phases that can be extended by subclasses.

@param topic   The Kafka topic
@param headers The Kafka record headers
@param payload The serialized protobuf payload
@return The deserialized object
core/src/main/java/kafka/automq/table/deserializer/proto/AbstractCustomKafkaProtobufDeserializer.java