apache / apache/pulsar

[Bug] When sending a message, broker don't verifies the logic of recordSequenceId > highestSequenceId.

Open
#21,886 0 comments 0 reactions 0 assignees View on GitHub
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

The latest master branch code

### Minimal reproduce step

This is a question I had after reading the pulsar source code. I’m not sure if it’s a real problem.

In function org.apache.pulsar.broker.service.ServerCnx#handleSend#line 1838,There is a logic to verify sequenceId, the code is as follows:

```java
// Persist the message
if (send.hasHighestSequenceId() && send.getSequenceId() <= send.getHighestSequenceId()) {
producer.publishMessage(send.getProducerId(), send.getSequenceId(), send.getHighestSequenceId(),
headersAndPayload, send.getNumMessages(), send.isIsChunk(), send.isMarker(), position);
} else {
producer.publishMessage(send.getProducerId(), send.getSequenceId(), headersAndPayload,
send.getNumMessages(), send.isIsChunk(), send.isMarker(), position);
}
```

But he did not check the following logic (if the following situation occurs, an exception should be returned):
```java
send.hasHighestSequenceId() && send.getSequenceId() > send.getHighestSequenceId()
```

### What did you expect to see?

I hope to check the following logic (if the following situation occurs, an exception should be returned):
```java
send.hasHighestSequenceId() && send.getSequenceId() > send.getHighestSequenceId()
```

### What did you see instead?

I find `org.apache.pulsar.broker.service.Producer#publishMessage` function makes a judgment of `lowestSequenceId > highestSequenceId`:
```java
public void publishMessage(long producerId, long lowestSequenceId, long highestSequenceId,
ByteBuf headersAndPayload, long batchSize, boolean isChunked, boolean isMarker, Position position) {
if (lowestSequenceId > highestSequenceId) {
```
so `org.apache.pulsar.broker.service.ServerCnx#handleSend` function can be changed to:
```java
// Persist the message
if (send.hasHighestSequenceId()) {
producer.publishMessage(send.getProducerId(), send.getSequenceId(), send.getHighestSequenceId(),
headersAndPayload, send.getNumMessages(), send.isIsChunk(), send.isMarker(), position);
} else {
producer.publishMessage(send.getProducerId(), send.getSequenceId(), headersAndPayload,
send.getNumMessages(), send.isIsChunk(), send.isMarker(), position);
}
```

### Anything else?

no

### Are you willing to submit a PR?

- [X] I'm willing to submit a PR!

Contributor guide

Open the contributing guide

Research direction

Start by reading org.apache.pulsar.broker.service.ServerCnx#handleSend around the sequenceId and highestSequenceId handling, then inspect org.apache.pulsar.broker.service.Producer#publishMessage and its lowestSequenceId check. Confirm the intended behavior for sequenceId greater than highestSequenceId and identify the relevant broker-service test location. Done means the invalid ordering is handled consistently with the issue’s expected exception behavior.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
backend, distributed-systems
Issue type
Bug
Difficulty
3/5
Estimated time
1-2 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
42/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.