[Bug] Transaction log and pending-ack store entries are parsed as MessageMetadata on every cache insert and always fail

Search before asking

• I searched in the issues and found nothing similar.

Read release policy

• I understand that unsupported versions don't get bug fixes. I will attempt to reproduce the issue on a supported version of Pulsar client and Pulsar broker.

Version

Affects master and branch-4.2 (any branch containing PIP-430's entry cache work, #24623 / #24682 / #24836; #26463 adds a second occurrence per entry on master).

Minimal reproduce step

The entry cache parses every entry it handles as a MessageMetadata, but it is used by managed ledgers whose entries are not Pulsar messages at all. The transaction coordinator log and the pending-ack store write protobuf records through TxnLogBufferedWriter, and their managed ledgers do get their entries cached: MLTransactionLogImpl opens a cursor, and ManagedLedgerImpl calls cursor.setActive() when a cursor is opened, so shouldCacheAddedEntry() is true.

Every batched entry those writers produce starts with a 4-byte prefix (TxnLogBufferedWriter.doFlush):

ByteBuf prefixByteBuf = PulsarByteBufAllocator.DEFAULT.buffer(4);
prefixByteBuf.writeShort(BATCHED_ENTRY_DATA_PREFIX_MAGIC_NUMBER);
prefixByteBuf.writeShort(BATCHED_ENTRY_DATA_PREFIX_VERSION);

and the two magic numbers collide:

// TxnLogBufferedWriter
public static final short BATCHED_ENTRY_DATA_PREFIX_MAGIC_NUMBER = 0x0e01;

// Commands
public static final short magicCrc32c = 0x0e01;

So Commands.hasChecksum() returns true on a transaction-log entry, and skipChecksumIfPresent skips Short.BYTES + Integer.BYTES = 6 bytes where the prefix is only 4 — landing two bytes into the payload. The readUnsignedInt() that follows reads a garbage metadata size, and parseFrom fails.

Reproducer:

ByteBuf prefix = Unpooled.buffer(4);
prefix.writeShort(0x0e01); // BATCHED_ENTRY_DATA_PREFIX_MAGIC_NUMBER
prefix.writeShort(1);      // BATCHED_ENTRY_DATA_PREFIX_VERSION
ByteBuf content = Unpooled.buffer(64);
for (int i = 0; i < 64; i++) {
    content.writeByte(i);
}
ByteBuf entryData = Unpooled.wrappedUnmodifiableBuffer(prefix, content);

System.out.println("hasChecksum = " + Commands.hasChecksum(entryData));
Commands.parseMessageMetadata(entryData.duplicate(), new MessageMetadata());

prints

hasChecksum = true

and then throws.

Error message or exception stacktrace

java.lang.IllegalArgumentException: Invalid unknonwn tag type: 6
	at org.apache.pulsar.common.api.proto.LightProtoCodec.skipUnknownField(...)
	at org.apache.pulsar.common.api.proto.MessageMetadata.parseFrom(MessageMetadata.java:...)
	at org.apache.pulsar.common.protocol.Commands.parseMessageMetadata(Commands.java:509)
	at org.apache.bookkeeper.mledger.impl.EntryImpl.initializeMessageMetadataIfNeeded(EntryImpl.java:314)

EntryImpl.initializeMessageMetadataIfNeeded catches Throwable and logs it, so this never surfaces as a failure — it shows up as a WARN "Failed to parse message metadata for entry" per entry, plus the cost of building and throwing the exception.

Per batched transaction-log entry that is:

• one failed parse when the entry is inserted into the cache (RangeEntryCacheImpl.insert, since #26463 — master only), and
• one more on the first cache read, because the parse left messageMetadata null so RangeCacheEntryWrapper.getValueInternal retries it under the StampedLock write lock before setting messageMetadataInitialized = true.

On branch-4.2 only the second one applies.

Anything else?

Two directions for a fix, both of which look reasonable:

1. Make the metadata parse opt-in per managed ledger, e.g. a ManagedLedgerConfig flag that PersistentTopic's ledgers enable and the transaction log / pending-ack store leave off. This also removes the wasted work rather than just the noise, and covers any other non-message managed ledger.
2. Give TxnLogBufferedWriter a prefix magic number that doesn't collide with Commands.magicCrc32c. This is a persisted format change, so it needs a version bump and a compatibility path for existing ledgers — probably not worth it on its own, but worth noting that the collision is also a latent hazard for anything else that sniffs the magic.

Option 1 seems clearly preferable. Either way, EntryImpl.initializeMessageMetadataIfNeeded swallowing Throwable and logging at WARN per entry deserves a second look — a payload the broker never intended to parse should not produce a warn-level log line at all.

Are you willing to submit a PR?

• I'm willing to submit a PR!