[I] [Bug][broker] Key_Shared delivers a key's message after a newer one when consumers reconnect while their old consumers are still registered [pulsar]

["lhotari (via GitHub)" <gi...@apache.org>]

lhotari opened a new issue, #26867:
URL: https://github.com/apache/pulsar/issues/26867

   ### Search before reporting
   
   - [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](https://pulsar.apache.org/contribute/release-policy/#supported-versions) don't get bug fixes. I will attempt to reproduce the issue on a supported version of Pulsar client and Pulsar broker.
   
   ### User environment
   
   - Broker and client: master at f2f026c7db79 (5.1.0-SNAPSHOT), in Docker containers of the test image
   - One broker with `-Xmx768m -XX:MaxDirectMemorySize=512m`, two bookies
   - Host: Linux 7.1.5, x86-64, 8 cores / 16 threads, with the framework's performance environment (fixed CPU frequency, no turbo, no thermald)
   - Found with the network conditions of the Pulsar Performance Testing Framework, #26859
   
   ### Issue Description
   
   Key_Shared subscriptions in the default mode (`AUTO_SPLIT` with consistent hashing, without out-of-order delivery) delivered a key's message after a newer message of the same key, when the consumers' connections disappeared without the broker knowing and the consumers reconnected while the broker still had the old consumers.
   
   The scenario: `iot-telemetry-small`, 1,000 msg/s, 30 topics, 20 Key_Shared subscriptions per topic, each consumed by 10 pods with a multi-topic consumer and a message listener: 6,000 consumers. Each message carries a device ID as its key and the device's sequence number. 100 s into the measurement, the consumers' 371 TCP connections are silently disconnected: their packets are dropped in both directions and the sockets on the consumers' side are destroyed, so the consumers reconnect at once, while the broker keeps the old consumers on the half-open connections until its keep-alive closes them 50–60 s later.
   
   On master with a 768 MB broker, it reproduced in 5 runs out of 5, with 5, 11, 1, 3 and 3 messages out of order. For example, in one application, topic `iot-telemetry-27`:
   
   ```text
   device=64257 expected=3 observed=4 messageId=3:4490:-1
   device=64257 recovered=3 messageId=3:4180:-1
   ```
   
   Sequence 4 (`3:4490`) was delivered, then sequence 3 (`3:4180`), the older message.
   
   What the broker's log and a simulation of the hash ring show for the three violations:
   
   - Each device's key hash belonged to one of the old consumers, the one whose connection had vanished, the whole time the new consumers joined. A reconnected consumer gets name index 1, because the old one with index 0 is still registered, so its ring points differ and it didn't take these hashes over.
   - The older message was produced after the packet filters went on, so it was dispatched to the old consumer and never arrived.
   - The hash moved only when the keep-alive removed the old consumer ("Removed consumer with pending acks", 12–43 pending acks), to one of the new consumers, and the older message was redelivered then.
   - The newer message, produced about 10 s later, still reached the application first.
   
   All the violations fall in the window where the broker's keep-alive removes the old consumers. In one run, the applications' connections were silently disconnected at 23:52:18.2 UTC, the broker removed the old consumers between about 23:53:03 and 23:53:18, and each violation fell in that window, 49–54 s after the disconnection; the newer message arrived first and the older one 1–4 s later, both with a redelivery count of 0:
   
   ```text
   device=32314 expected=1 observed=2 messageId=30:4234:-1 topic=.../iot-telemetry-4 redeliveryCount=0 receivedEpochMs=1791762787080  (23:53:07.080)
   device=32314 recovered=1 messageId=30:3933:-1 topic=.../iot-telemetry-4 redeliveryCount=0 receivedEpochMs=1791762790421  (23:53:10.421)
   ```
   
   The mechanism, which a unit test of `PersistentStickyKeyDispatcherMultipleConsumers` reproduces:
   
   1. A replay read is in flight with a newer entry of a sticky key hash H, such as one that a consumer without permits couldn't take earlier.
   2. The consumer that owns H is removed. `PersistentDispatcherMultipleConsumers.removeConsumer` adds its pending entries to the replay queue, among them an older entry of H, and H moves to another consumer.
   3. The read completes. The guard in `PersistentStickyKeyDispatcherMultipleConsumers.trySendMessagesToConsumers` discards a read only when the replay queue's first position became older during the read; when the queue already held an older position of another hash, which a consumer without permits leaves there, the read isn't discarded. Replay reads don't check for the hash's entries in the replay queue, as normal reads do in `canDispatchEntry`.
   4. The newer entry of H goes to H's new owner, and the older one only with the next replay read.
   
   It happened with 768 MB brokers on master in 5 runs out of 5 and with 512 MB brokers that had #26863 in 2 runs out of 3, but not with 768 MB brokers that had #26863 (2 runs) nor with 4 GB brokers (2 runs), so GC pressure likely widens the window. A 512 MB broker without #26863 runs out of heap in this scenario instead (#26862).
   
   **Expected:** a key's messages are delivered in order, also when consumers reconnect while their old consumers are still registered.
   
   ### Error messages
   
   ```text
   No errors; the applications' ordering check reports the violations.
   ```
   
   ### Reproducing the issue
   
   With the network conditions of the Pulsar Performance Testing Framework, on master since #26859:
   
   ```bash
   ./gradlew :tests:performance:launcher:run -Pperformance.keepLauncherLog \
     --args="--scenario tests/performance/scenarios/iot-telemetry-small.yaml --extends network/silent-client-disconnect --set 'cluster.brokers.env.PULSAR_MEM=-Xmx768m -XX:MaxDirectMemorySize=512m'"
   ```
   
   The applications report each violation in `applications/<application>/ordering-violations.txt`, with each message's redelivery count and receipt time, and exit with status 1, which fails the run.
   
   A unit test of `PersistentStickyKeyDispatcherMultipleConsumers` reproduces it deterministically: a replay read of the newer entry is in flight, the consumer holding the older entry as a pending ack is removed, and the read completes; the hash's new owner gets the newer entry first.
   
   ### Are you willing to submit a PR?
   
   - [x] I'm willing to submit a PR!
   
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [I] [Bug][broker] Key_Shared delivers a key's message after a newer one when consumers reconnect while their old consumers are still registered [pulsar]

["merlimat (via GitHub)" <gi...@apache.org>]

merlimat closed issue #26867: [Bug][broker] Key_Shared delivers a key's message after a newer one when consumers reconnect while their old consumers are still registered
URL: https://github.com/apache/pulsar/issues/26867


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org