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

Abierto
  • ¿Algo no está bien?
#620 0 comentarios 0 reacciones 0 asignados Ver en GitHub

Nadie ha tomado este issue todavía.

Evaluación

Dificultad
4/5
Tiempo estimado
3-5 días
Aptitud para principiantes
55/100
Tipo de issue
Error
Claridad
Bien especificado
Estado de actividad
Activo
Stack tecnológico
cpp
Área
distributed-systems

Línea de trabajo

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.

Escrito por el modelo de indexación a partir del texto del issue.

Descripción

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!
Lenguaje dominante
C++
Estrellas
72
Forks
93
Sin PR fusionados en 30 d

Preparar el entorno

Primeros pasos

  1. Lee el issue completo y luego la guía de contribución del proyecto.
  2. Comenta en el issue que vas a ocuparte — evita que dos personas hagan lo mismo.
  3. Haz un fork del repositorio y trabaja en una rama.
  4. Abre un pull request que haga referencia al número del issue.

Más de apache/pulsar-client-cpp

Todos los issues de apache/pulsar-client-cpp

Issues similares

Más issues de C++

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.