[feat] [perf] Support for transaction consistence test with PerformanceProducer.
- 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.
### Motivation
Currently, `bin/pulsar-perf` do not support for verify the exactly-once semantics. Like Kafka, Pulsar only support for exactly-once semantics in specific working pattern, that is `consume-transform-produce`, while i try to achieve exactly-once semantics in other working pattern. the detailed info see https://github.com/apache/pulsar/issues/19744.
This tool is to verify the transaction consistence with producer-only working pattern. We can use this tool to generate messages with or without transaction, and do the experiment in any disturbing environment, for example, we can restart brokers hundreds of times, unload topics hundreds of times or use `tc qdisc` to drop the packet. Finally, we consume the messages produced **without interference** and check for the data integrity, that is no message lost or duplication.
### Solution
Implement the logic into `PerformanceProducer`.
## Message Format
For the convenience of checking for the data integrity, **we define the message format as number incrementing from 0 to N**. For example, we start the perf process with configuration `--num-messages 100000000`, then we read the content produced to broker successfully and **verify whether the content constitute a range `[0,100000000)`**, which can be implemented easily with `RangeSet`.
## Client-Side Persistence Scheme
Client may corrupt and restart when produce messages in transaction, it need to persist the messages that have been sent in transaction to avoid message lost or duplication.
I adopt a simper way to persist the messages, that is to persist every messages in a local file `tmpX.data` when producer send it in transaction. **If the transaction is committed successfully, the content of local file `tmpX.data` will be cleared. If the transaction is aborted or committed failed, we will resend messages in this aborted transaction with a new transaction.**
**We assign every transaction with a tmp file, such as `tmp0.data`**, to persist the messages have been sent in this transaction. As there will be many transaction running concurrently, we will create many tmp files for persistence, and add concurrent control to these files.
When client is shutdown, it will take a snapshot of current number of message generator `IncrementedNumberMessageFormatter`. When the client is restarted, it will read the value from snapshot and continue generate number from the state before shutdown.
the detailed implementation see https://github.com/apache/pulsar/pull/19781.
I will show my experiment result following.
## Usage
### Non-transactional produce with topic unload
Produce 100 million messages, unload topics every 80 seconds for 50 times.
```
bin/pulsar-perf produce --num-messages 100000000 --rate 100000 --format-class org.apache.pulsar.testclient.IncrementedNumberMessageFormatter --send-timeout 30 --resend-on-failure persistent://test/test/testTxn10
```
the script for unloading topics:
```
#!/bin/bash
# script for restarting broker.
interval=80 # interval in second to restart broker.
unloadCount=50 # total restart times.
topic="persistent://test/test/testTxn10"
while true
do
if test $unloadCount -ne 0
then
echo "unload topic:"$topic
bin/pulsar-admin topics unload $topic
unloadCount=$(expr $unloadCount - 1)
else
break
fi
sleep $interval
done
```
Result as follows:


```
09:59:54.250 [main] INFO org.example.utils.PulsarCommonUtils - only one range exists. RangeSet:[[0..100000000)]
09:59:54.254 [main] INFO org.example.utils.PulsarCommonUtils - duplicate message count:11317, total message count:100011317, duplicate rate:0.011315719400035498%
```
There are 11317 messages are duplicated.
### Transactional produce with broker restarting hundreds of times.
Produce 100 million messages, restarting brokers every 120 seconds for 50 times.
```
nohup bin/pulsar-perf produce --producer-name "txnProducer" -txn -nmt 800 --num-messages 100000000 --txn-test-enabled --max-txn 20 --rate 100000 --format-class org.apache.pulsar.testclient.IncrementedNumberMessageFormatter --send-timeout 0 persistent://test/test/testTxn10 > produce.txt &
```
```
#!/bin/bash
# script for restarting broker.
interval=120 # interval in second to restart broker.
restartCount=50 # total restart times.
while true
do
if test $restartCount -ne 0
then
bin/pulsar-daemon restart broker
restartCount=$(expr $restartCount - 1)
else
break
fi
sleep $interval
done
```
The result is as follows:
```
19:21:23.507 [main] INFO org.example.utils.PulsarCommonUtils - only one range exists. RangeSet:[[0..100000000)]
19:21:23.510 [main] INFO org.example.utils.PulsarCommonUtils - duplicate message count:800, total message count:100000800, duplicate rate:7.999936000511996E-4%
```
There are 800 messages duplicated, which is resulted by one transaction, as we produce 800 messages in a transaction. So, we can see that **exactly-once semantics is not guaranteed in producer-only working pattern!**
The reason accounting for duplication see https://github.com/apache/pulsar/issues/19629, and I have fixed them in https://github.com/apache/pulsar/issues/19744.
Following I will show the experiment result after i merge the PR https://github.com/apache/pulsar/pull/19662, and you can repeat my experiment with these two PR #19662 and #19781.
### Transactional produce with broker restarting hundreds of times (PR merged).
```
nohup bin/pulsar-perf produce --producer-name "txnProducer" -txn -nmt 800 --num-messages 100000000 --txn-test-enabled --max-txn 30 --rate 200000 --format-class org.apache.pulsar.testclient.IncrementedNumberMessageFormatter --send-timeout 0 -threads 4 persistent://test/test/testTxn10 > produce.txt &
```

```
13:17:03.543 [main] INFO org.example.utils.PulsarCommonUtils - only one range exists. RangeSet:[[0..100000000)]
13:17:03.546 [main] INFO org.example.utils.PulsarCommonUtils - duplicate message count:0, total message count:100000000, duplicate rate:0.0%
```
We can see that there is no message duplication or lost.
### Alternatives
_No response_
### Anything else?
_No response_
### Are you willing to submit a PR?
- [X] I'm willing to submit a PR!
Contributor guide
Research direction
Start with the PerformanceProducer entry point and the implementation described in PR #19781, including IncrementedNumberMessageFormatter, transaction tmpX.data persistence, and RangeSet verification. Reproduce the documented bin/pulsar-perf produce commands and disruption experiments. Done means transactional producer-only runs verify the [0, N) range without lost or duplicate messages.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- cli, distributed-systems, testing-qa
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 25/100