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

Đang mở
  • Có gì không ổn?
#620 0 bình luận 0 reaction 0 người được giao Xem trên GitHub

Chưa có ai nhận issue này.

Đánh giá

Độ khó
4/5
Thời gian dự kiến
3-5 ngày
Mức phù hợp với người mới
55/100
Loại issue
Lỗi
Độ rõ ràng
Đặc tả rõ ràng
Mức độ hoạt động
Sôi nổi
Công nghệ
cpp
Lĩnh vực
distributed-systems

Hướng nghiên cứu

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.

Do mô hình lập chỉ mục viết ra từ nội dung của issue.

Mô tả

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!
Ngôn ngữ chính
C++
Star
72
Fork
93
Không có pull request nào được merge trong 30 ngày

Chuẩn bị môi trường

Bắt đầu từ đâu

  1. Đọc hết issue, rồi đọc hướng dẫn đóng góp của dự án.
  2. Bình luận trên issue rằng bạn sẽ nhận — tránh hai người làm cùng một việc.
  3. Fork repository và làm thay đổi trên một nhánh.
  4. Mở pull request có tham chiếu số hiệu của issue.

Issue khác của apache/pulsar-client-cpp

Tất cả issue của apache/pulsar-client-cpp

Issue tương tự

Thêm issue về C++

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.