[Bug] Dead-lettered messages are tagged with the consumer's schema, not the schema they were written with #620

Search before asking

• I searched in the issues and found nothing similar.

Version

• pulsar-client-cpp main (f71d7e9). The code path is the same in 4.0.1, which the reproduction below runs through pulsar-client-python 3.10.0.
• Broker: Pulsar 4.0.8 standalone.

Minimal reproduce step

A topic has an Avro schema V1, {name: string}. A consumer subscribes with the compatible V2, {name: string, age: int = 0}, and has a dead letter policy. A V1 producer sends one message, and the consumer negatively acknowledges it until it is dead-lettered.

"""A dead-lettered message is tagged with the consumer's schema version, not the version it was written with."""

import io
import json
import urllib.request
import uuid

import fastavro
import pulsar
from pulsar.schema import AvroSchema

ADMIN = "http://localhost:8080/admin/v2/schemas/public/default"
V1 = {"type": "record", "name": "User", "fields": [{"name": "name", "type": "string"}]}
V2 = {**V1, "fields": [*V1["fields"], {"name": "age", "type": "int", "default": 0}]}

topic = f"dlq-schema-{uuid.uuid4().hex[:8]}"
dlq = f"{topic}-DLQ"
client = pulsar.Client("pulsar://localhost:6650")


def schema_version(msg):
    return int.from_bytes(msg.schema_version().encode("latin-1"), "big")


# A V1 producer writes one message; the consumer reads with the compatible V2.
producer = client.create_producer(topic, schema=AvroSchema(None, schema_definition=V1))
consumer = client.subscribe(
    topic,
    "sub",
    consumer_type=pulsar.ConsumerType.Shared,
    schema=AvroSchema(None, schema_definition=V2),
    negative_ack_redelivery_delay_ms=100,
    dead_letter_policy=pulsar.ConsumerDeadLetterPolicy(
        max_redeliver_count=1, dead_letter_topic=dlq, initial_subscription_name="inspect"
    ),
)
producer.send({"name": "alice"})

# Fail the message until it is dead-lettered.
for _ in range(3):
    try:
        msg = consumer.receive(timeout_millis=3000)
    except pulsar.Timeout:
        break
    print("received: redelivery", msg.redelivery_count(), "schema version", schema_version(msg))
    consumer.negative_acknowledge(msg)

# Read the DLQ copy as raw bytes.
inspector = client.subscribe(dlq, "inspect")
copy = inspector.receive(timeout_millis=10000)
version = schema_version(copy)
tagged = json.loads(json.load(urllib.request.urlopen(f"{ADMIN}/{dlq}/schema/{version}"))["data"])
print("DLQ copy: schema version", version, "=", "V2" if tagged == V2 else "V1" if tagged == V1 else tagged)

for name, schema in (("tagged schema", tagged), ("original V1", V1)):
    try:
        print(f"decode with {name}:", fastavro.schemaless_reader(io.BytesIO(copy.data()), fastavro.parse_schema(schema)))
    except Exception as e:
        print(f"decode with {name}: {type(e).__name__}: {e}")

client.close()

Output:

received: redelivery 0 schema version 0
received: redelivery 1 schema version 0
DLQ copy: schema version 0 = V2
decode with tagged schema: EOFError:
decode with original V1: {'name': 'alice'}

What did you expect to see?

The DLQ copy should carry the schema it was written with (V1), so a consumer of the DLQ topic can decode it. The Java client does this. It sends each dead-lettered message with that message's own schema:

producerDLQ.newMessage(Schema.AUTO_PRODUCE_BYTES(message.getReaderSchema().get()))
        .value(message.getData())

What did you see instead?

The DLQ copy holds V1 bytes but is tagged with the DLQ topic's version of V2. Decoding it with its tagged schema fails, so the message can't be read from the DLQ, which is where people go to inspect failed messages. When a schema change reorders or retypes fields, the decode can succeed and return wrong values instead of failing.

The cause is in ConsumerImpl::processPossibleToDLQ (lib/ConsumerImpl.cc). The dead-letter producer is created once with the consumer's schema:

producerConfiguration.setSchema(config_.getSchema());

The message is then copied with setAllocatedContent(message.getData(), message.getLength()), which drops its schema version. Every message gets sent under the consumer's schema, whichever schema it was written with.

Anything else?

Java's approach depends on a multi-schema producer, where each message can register its own schema through GetOrCreateSchema. The C++ producer has one schema per producer, so there are two ways to fix this:

1. Keep one dead-letter producer per writer schema version. Look up the writer schema with Client::getSchemaInfoAsync(topic, message.getLongSchemaVersion()) and create the producer with that SchemaInfo. The broker registers or reuses that schema on the DLQ topic, so the copy is tagged with the DLQ topic's version of the right schema. This needs no protocol work.
2. Add per-message schemas to the producer, as Java does. This is a bigger change, but it would also unlock AUTO_PRODUCE_BYTES in general.

Messages without a schema version (!message.hasSchemaVersion()) can keep today's behavior.

Bindings built on this library (Python, Node.js) hit this too. Node.js just added Message.getLongSchemaVersion() and Client.getSchemaInfo() (apache/pulsar-client-node#509, #512) so consumers can decode messages written with an older schema. Their dead-lettered copies still lose that information.

Are you willing to submit a PR?

• I'm willing to submit a PR!