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

未關閉
  • 哪裡不對？
#620 0 則留言 0 個 reaction 已指派 0 人 在 GitHub 檢視

還沒有人認領這個 Issue。

評估

難度
4/5
預估耗時
3-5 天
新手友好度
55/100
Issue 類型
缺陷
描述清晰度
描述清楚
活躍度
活躍
技術堆疊
cpp
領域
distributed-systems

研究方向

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.

由索引模型根據 Issue 內容生成。

描述

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!
主要語言
C++
星號
72
分支
93
30 天內沒有已合併 PR

環境準備

從這裡開始

  1. 先讀完整個 Issue，再讀專案的貢獻指南。
  2. 在 Issue 下留言說明你要接手 —— 這能避免兩個人做同樣的事。
  3. Fork 儲存庫，在一個分支上完成修改。
  4. 送出 Pull Request，並在描述裡引用這個 Issue 編號。

apache/pulsar-client-cpp 的其他 Issue

查看 apache/pulsar-client-cpp 的全部 Issue

相似的 Issue

更多 C++ Issue

把新 issue 寄到你的電子郵件信箱

精選適合新手參與的 GitHub issue 摘要。