[Bug] When sending a message, broker don't verifies the logic of recordSequenceId > highestSequenceId.
- 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
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