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

Ouverte
  • Quelque chose ne va pas ?
#620 0 commentaires 0 réactions 0 personnes assignées Voir sur GitHub

Personne n'a encore pris cette issue.

Évaluation

Difficulté
4/5
Temps estimé
3-5 jours
Accessibilité débutants
55/100
Type d'issue
Bug
Clarté
Clairement spécifiée
Activité
Active
Stack technique
cpp
Domaine
distributed-systems

Piste de recherche

Start in lib/ConsumerImpl.cc, at ConsumerImpl::processPossibleToDLQ, and trace how the dead-letter producer is configured and how message data is copied. Compare the C++ behavior with the Java example in the issue and investigate how writer schema information can be retrieved for a message. Done means DLQ copies retain a schema that decodes their original bytes, including the V1 reproduction.

Rédigé par le modèle d'indexation à partir du texte de l'issue.

Description

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.

repro.py (Python, which runs the C++ client) 
"""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!
Langage dominant
C++
Étoiles
72
Forks
93
Aucune PR mergée en 30 j

Préparer son environnement

Par où commencer

  1. Lisez l'issue en entier, puis le guide de contribution du projet.
  2. Signalez en commentaire que vous la prenez — cela évite que deux personnes fassent le même travail.
  3. Forkez le dépôt et travaillez sur une branche.
  4. Ouvrez une pull request qui référence le numéro de l'issue.

Autres issues de apache/pulsar-client-cpp

Toutes les issues de apache/pulsar-client-cpp

Issues similaires

Plus d'issues C++

Recevez les nouvelles issues par e-mail

Un résumé court des issues GitHub adaptées aux débutants.