it's also worth pointing out that the current implementation will not work at all for any protobuf which references another protobuf via an import statement, since it parses only the schema stored by the registry and not any schema referenced by that schema.

+1

it's missing the "message-indexes" part.

I have code which encodes it correctly. It isn't integrated with the schema registry client. I built a separate serializer so that I am using schema registry client only to talk to the registry. Code looks like this:

    private encodePayloadHeader(schemaInfo: SchemaInfo, msgIndexes: number[]): ProtoBuffer {
        let writer = new BufferWriter()
                .uint32(0) // don't zig zag encode. uint32 writes variable number of bytes, so this is 1 byte
                .fixed32(NetworkOrder(schemaInfo.schemaId)) // make big-endian, then write 4 bytes int32
                .sint32(msgIndexes.length) // zig-zag encoded array length

        for (const msgIndex of msgIndexes) {
            writer.sint32(msgIndex) // zig-zag encoded index value
        }
        return writer.finish()
}

I generate the full buffer more or less like this:

        const buf = this.encodePayloadHeader(schemaInfo, msgIndexes)
        const writer = schemaInfo.type.encode(thing)
        return Buffer.concat([buf, writer.finish()])

On the decode side, it looks like this:

    private decodeHeader(topic: string, buffer: Buffer): ProtoInfo {
        let bufferReader = Reader.create(buffer)
        const magicByte = bufferReader.uint32()
        const schemaId = HostOrder(bufferReader.fixed32())
        const arrayLen = bufferReader.sint32()
        const msgIndexes = new Array<number>(arrayLen)
        for (let i = 0; i < arrayLen; i++) {
            msgIndexes[i] = bufferReader.sint32()
        }
        return {
            magicByte: magicByte,
            schemaId: schemaId,
            msgIndexes: msgIndexes,
            bytesRead: bufferReader.pos,
        }
    }

    public async deserialize(topic: string, buffer: Buffer): Promise<Message<{}>> {
        if (buffer.length < 6) {
            throw new Error(`buffer with length ${buffer.length} is not long enough to contain a protobuf`)
        }
        const protoInfo = this.decodeHeader(topic, buffer)

        const type = await this.protobufResolver.ResolveProtobuf(topic, protoInfo.schemaId, protoInfo.msgIndexes)
        let bufferReader = Reader.create(buffer)
        bufferReader.skip(protoInfo.bytesRead)
        return type.decode(bufferReader)
    }

protobufResolver uses registry client, topic name, and info parsed from wire protocol to resolve a protobuf Type instance, which is then used to decode the protobuf. That allows me to inject whatever logic I want into the deserializer via protobufResolver for figuring out the type that is encoded in the payload, since correctly computing the message type from message indexes isn't really possible with protobufjs as it is currently implemented. At least not if you also have imported references to other protobufs in your .proto files, since the schema parsed out of the registry won't include the references. By delegating to a resolver, I can resort to quick hacks like hardcoding the type name based on indexes and topic name, for example.

I'll probably open-source the serializer and deserializer I built, but it's not likely to happen for a week or two.

Hi @ideasculptor , got the same issue. Looking forward for your fixes. Many thanks.

Same problem, any ETA for a fix?

Same issue, do we have an ETA for this fix?

I'll probably open-source the serializer and deserializer I built, but it's not likely to happen for a week or two.

@ideasculptor have you had the chance of working on a PR to address this issue? I have recently started using kafkajs/confluent-schema-registry and I can confirm that the library works for JSON and AVRO types, but fails for PROTOBUF.

Any update on this?

@ideasculptor Do you plan to proceed with it?

Oops, I wrote a comment thinking this was the golang version of the client. @NikitaKemarskiyLeia - it seems unlikely that I'm going to submit an actual PR given how long it has been since I touched this. But the code I pasted above has been working for us very well for quite some time and is cross-compatible with other kafka clients in other languages that do have full protobuf schema registry support, so you've got everything you need to implement your own serialization and deserialization.

Admittedly, copying and pasting those few dozen lines of code is sub-optimal, but having to dig back into this codebase and figure out who to integrate it properly for a PR would be a fair bit of work for me and it is pretty far down my priority queue at the moment, so I wouldn't hold your breath.

@ideasculptor Thanks for the response, I'll probably work on this PR by myself then. Your comment was also useful – I have applied your fix to my code (unless the bug is fixed in the library).

Hi @NikitaKemarskiyLeia , did you got a chance to work on the PR?

@Meet-Modi Unfortunately have no time for such contribution now :(
Using the solution provided by @ideasculptor

@tulios @Nevon Looks like a lot of people are facing this issue when communicating the data across different services and it is pending for more than 1.5 years. It would be great if same can be picked in next release considering the impact. Thanks in advance.

@ideasculptor could you please help with definition of this.protobufResolver.ResolveProtobuf to get the type?

@ideasculptor hello, thank you a lot for sharing workaround, or could you please advice how to fetch msg ids, share your codebase

@NikitaKemarskiyLeia might be you can help