apache / apache/pulsar

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

Open
#26,494 0 comments 0 reactions 1 assignee Claimed by @lhotari View on GitHub
Dominant language
Java
Stars
15.3k
Forks
3.8k
Avg merge
1d 14h
Merged PRs (30d)
160

Description

### Search before asking

- [X] I searched in the [issues](https://github.com/apache/pulsar/issues) and found nothing similar.

### Read release policy

- [X] 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`):

```java
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:

```java
// 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:

```java
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!

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.