Using catalog::apicurio triggers unexpected behaviour

[ankitk-me]

Describe the bug
When using catalog::apicurio with binding::asyncapi, I was able to successfully produce a valid message to Kafka topic. Able to validate that based on response code: 204 & by consuming kafka topic using a third party client.

But in the Zilla log, observed mentioned log line:

1723729278272:zilla-http-kafka-asyncapi:south_kafka_client/kafka_api kafka_client0: Skipping invalid message on topic petstore-pets, partition 0, offset 0

This log is triggered during fetch, if the message fails validation.

While debugging further, below are my findings:

• We always encode schemaId during produce as int32 (4 bytes), but while consuming by default we decode 8 bytes to determine schema Id. We should make this consistent by either encoding schemaId as 8 bytes or by default decode 4 bytes for schema Id.

• I tried resolving this issue by setting "id-encoding" to "legacy", in catalog config but still same issue persisted.

• To validate my findings, I hard coded this.encodeId = this::encodeDefaultId; to use default encoding strategy and the issue was resolved.

To Reproduce
Steps to reproduce the behavior:

1. Start an asyncapi-http-kafka example with apicurio catalog.

catalogs:
  apicurio_catalog:
    type: "apicurio"
    options:
      url: "http://apicurio.zillabase.dev:8080"
      group-id: "default"

2. Produce a valid message to Kafka topic.
3. See error Skipping invalid message on topic in Zilla log.

Expected behavior
On fetch, we should receive valid event.