apache / apache/pulsar

[Bug] Deserialized BatchMessageIdImpl cannot be used for acknowledgment

Open
#19,030 2 comments 0 reactions 1 assignee Claimed by @BewareMyPower View on GitHub
Stale type/bug
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.

### Version

OS: Ubuntu 20.04
Pulsar: master (41edd2e)

### Minimal reproduce step

Add a unit test that does these things:
1. Create a producer to send N messages in the same batch.
2. Create a consumer to receive these messages and store the `MessageId` objects, which share the same ledger id, entry id, batch size, only the batch indexes are different.
3. Convert these `MessageId` objects by a serialization (`MessageId#toByteArray`) and a deserialization (`MessageId#fromByteArray`).
4. Acknowledge these `MessageId` objects.
5. Restart the consumer, it still receives the 1st message.

```java
@Test
public void testSerialization() throws Exception {
var topic = "test-serialization-origin";
@Cleanup var producer = pulsarClient.newProducer(Schema.INT32)
.topic(topic)
.batchingMaxMessages(100)
.batchingMaxPublishDelay(1, TimeUnit.DAYS)
.create();
@Cleanup var consumer = pulsarClient.newConsumer(Schema.INT32)
.topic(topic)
.subscriptionName("sub")
.isAckReceiptEnabled(true)
.subscribe();

final var numMessages = 10;
for (int i = 0; i < numMessages; i++) {
producer.sendAsync(i);
}
producer.flush();
final var msgIds = new ArrayList();
for (int i = 0; i < numMessages; i++) {
msgIds.add(consumer.receive().getMessageId());
}
for (int i = 1; i < numMessages; i++) {
final var lhs = (BatchMessageIdImpl) msgIds.get(0);
final var rhs = (BatchMessageIdImpl) msgIds.get(i);
assertEquals(lhs.getLedgerId(), rhs.getLedgerId());
assertEquals(lhs.getEntryId(), rhs.getEntryId());
assertEquals(lhs.getBatchSize(), rhs.getBatchSize());
assertEquals(lhs.getBatchSize(), numMessages);
}

var deserializedMsgIds = new ArrayList();
for (var msgId : msgIds) {
var deserialized = MessageId.fromByteArray(msgId.toByteArray());
assertTrue(deserialized instanceof BatchMessageIdImpl);
deserializedMsgIds.add(deserialized);
}
for (var msgId : deserializedMsgIds) {
consumer.acknowledge(msgId);
}
consumer.close();

consumer = pulsarClient.newConsumer(Schema.INT32)
.topic(topic)
.subscriptionName("sub")
.isAckReceiptEnabled(true)
.subscribe();
final var msg = consumer.receive(3, TimeUnit.SECONDS);
assertNotNull(msg);
assertEquals(msg.getValue(), 0);
}
```

### What did you expect to see?

The restarted consumer should receive nothing.

### What did you see instead?

The restarted consumer received the 1st message.

### Anything else?

The root cause is from https://github.com/apache/pulsar/pull/1424, which make all `MessageId` instances in the same batch share the same `BatchMessageAcker` object. However, when `MessageId` instances are created from a deserialization. It's impossible to make them share the same `BatchMessageAcker`.

### Are you willing to submit a PR?

- [X] 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.