[I] [Bug][Producer] A send error or an out-of-order receipt for one producer closes the shared connection, reconnecting every producer and consumer on it [pulsar-client-go]

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

diegofernandes opened a new issue, #1553:
URL: https://github.com/apache/pulsar-client-go/issues/1553

   #### Expected behavior
   
   A send error, or an unexpected send receipt, that concerns one producer makes that producer reconnect. The other producers and consumers sharing the connection stay connected.
   
   That is what the client already does on `CLOSE_PRODUCER` (`handleCloseProducer`). It is also what the Java client does on a send error since apache/pulsar#23038:
   
   ```java
   default:
       // don't close this ctx, otherwise it will close all consumers and producers which use this ctx
       producer.connectionClosed(this, Optional.empty(), Optional.empty());
   ```
   
   #### Actual behavior
   
   Two code paths close the whole connection because of one producer. With `MaxConnectionsPerBroker` at its default of 1, every producer and consumer the client has on that broker shares that connection, so all of them reconnect at once:
   
   - producers re-send their pending messages, some of which the broker had already persisted, so those are written twice when deduplication is off;
   - acknowledgements in flight with `AckWithResponse` fail with "connection closed".
   
   We hit it by deleting a topic with `force=true` while another client was still sending to it. In a load test, each of 3 such deletes closed about 920 producers and consumers at once. Together they wrote 2,619 messages twice and failed 266 acknowledgements, all on topics unrelated to the deleted ones.
   
   Observed on v0.21.0. Both paths are unchanged on master (3bab464).
   
   **1. `connection.handleSendError`** ([pulsar/internal/connection.go#L918-L921](https://github.com/apache/pulsar-client-go/blob/3bab46476cb8f8178b07e410fc26cde588dfa2d0/pulsar/internal/connection.go#L918-L921))
   
   ```go
   default:
   	// By default, for transient error, let the reconnection logic
   	// to take place and re-establish the produce again
   	c.Close()
   ```
   
   The broker answers a send that meets a forced delete with `PersistenceError` ("ManagedLedger … is fenced", or "fenced" once a refused write has fenced the topic). Every error other than `NotAllowedError` and `TopicTerminatedError` takes this branch. The comment's intent is to re-establish the producer. Closing the connection does that, but for every listener on the connection.
   
   Terminating the topic before deleting it does not avoid it. The first write the terminated ledger refuses fences the topic on the broker, and the other sends in flight then get `PersistenceError: fenced`.
   
   **2. `partitionProducer.ReceivedSendReceipt`** ([pulsar/producer_partition.go#L1426-L1431](https://github.com/apache/pulsar-client-go/blob/3bab46476cb8f8178b07e410fc26cde588dfa2d0/pulsar/producer_partition.go#L1426-L1431))
   
   ```go
   if pi.sequenceID < response.GetSequenceId() {
   	// Force connection closing so that messages can be re-transmitted in a new connection
   	p.log.Warnf("Received ack for %v on sequenceId %v - expected: %v, local < remote, closing connection",
   		response.GetMessageId(), response.GetSequenceId(), pi.sequenceID)
   	p._getConn().Close()
   	return
   ```
   
   After the broker closes a producer (`CLOSE_PRODUCER`, as a forced delete does), the producer reconnects on the same connection. The receipts of its earlier sends can then arrive ahead of its oldest pending message, and this closes the connection. In our runs it showed up once the first path no longer closed the connection (with the fix below, or when the topic was terminated before the delete). Otherwise the send error had already closed it. The Java client still closes the channel here (`ProducerImpl.ackReceived`). #650 raised this path in 2021, before it was narrowed to the `local < remote` case.
   
   #### Steps to reproduce
   
   1. One client with `DisableBatching: true`, and broker deduplication off. Two topics, A and B, owned by the same broker.
   2. Send to A from 8 goroutines in a loop. On B, send in a loop and run a `Shared` consumer with `AckWithResponse: true` that acks everything it receives.
   3. From another client, delete A: `DELETE /admin/v2/persistent/<tenant>/<namespace>/A?force=true`.
   4. The client logs `Received send error from server: [PersistenceError]`, then `Connection was closed` for every producer and consumer on the connection. B's consumer receives some messages twice, and acks in flight fail with "connection closed".
   
   Measured on a local 3-broker cluster with that setup, 3 deletes of A per run:
   
   | client | runs | runs where B got duplicates or failed acks | `PersistenceError` send errors | out-of-order receipts |
   |---|---|---|---|---|
   | v0.21.0 | 10 | 5 (2–6 duplicates; once 6 failed acks) | 41 | 0 |
   | v0.21.0 + the fix below | 12 | 0 | 44 | 2 |
   
   #### Proposed fix
   
   In both places, reconnect only the producer, as `handleCloseProducer` already does:
   
   ```go
   // connection.handleSendError, default branch
   producer, ok := c.deletePendingProducers(producerID)
   if !ok {
   	c.log.Warnf("Received send error for unknown producer %d: [%v] : [%s]",
   		producerID, sendError.GetError(), sendError.GetMessage())
   	return
   }
   producer.ConnectionClosed(nil)
   
   // partitionProducer.ReceivedSendReceipt, local < remote
   p._getConn().UnregisterListener(p.producerID)
   p.ConnectionClosed(nil)
   ```
   
   The producer leaves the connection's listeners, reconnects, and re-sends what it had pending. Every other producer and consumer stays connected. A producer whose topic was deleted fails to reconnect with `TopicNotFound`, closes, and fails its pending sends with `ErrTopicNotfound`, as it does today.
   
   We have this change with three unit tests that need no broker. They fail on v0.21.0 and pass with the change, and `./pulsar/internal/...` passes. We're happy to open a PR.
   
   #### System configuration
   **Pulsar version**: broker 5.0.0; pulsar-client-go v0.21.0 (same code on master at 3bab464)
   


-- 
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][Producer] A send error or an out-of-order receipt for one producer closes the shared connection, reconnecting every producer and consumer on it [pulsar-client-go]

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

diegofernandes commented on issue #1553:
URL: https://github.com/apache/pulsar-client-go/issues/1553#issuecomment-6346832702

   **A correction to the proposed fix: reconnecting the producer on the same connection feeds back on itself.**
   
   We have run the fix proposed above under load since we filed this. It keeps the other producers and consumers connected, as intended. On the producer itself, though, it starts a loop that can last as long as that producer is busy. Please don't take it as written.
   
   **The loop.** The proposed fix reconnects the producer and leaves the connection alone, on a send error (`deletePendingProducers` and `ConnectionClosed(nil)`) and on an out-of-order receipt (`UnregisterListener` and `ConnectionClosed(nil)`). Here is what follows:
   
   1. **The receipts in flight are dropped.** While the producer is unregistered, `handleSendReceipt` logs "unexpected send receipt" for each of them.
   2. **The broker never closed the producer.** It answers the new `PRODUCER` request with "Producer with the same id is already created". The client then resends everything pending, including messages the broker is still writing.
   3. **The next receipt comes out of order.** A receipt for a later message arrives ahead of the oldest pending message, whose own receipt was dropped in step 1. That makes the producer re-register again.
   4. **With deduplication on,** each resent message that is still being written is refused with "Cannot determine whether the message is a duplicate at this time". Each refusal is another send error, so the producer re-registers again.
   5. **Without deduplication,** each resend writes the pending messages again.
   
   Under load there are always messages in flight, so the loop does not settle until the load stops.
   
   **What we measured** (5.0.0 standalone, `DisableBatching`). In a load test, one producer of the busiest topic logged about 400,000 "Cannot determine" errors by itself.
   
   | injected | with the proposed fix | with the fix below |
   |---|---|---|
   | One close of a busy producer (about 10k sends/s for 8 s), dedup on | 4,514–4,641 re-registrations, still about 1,400 a second at the end, in 3 of 4 runs; 4.3–4.5 million receipts dropped; 1,020 sends timed out | 1 re-registration, settled; 0 lost (6 of 6 runs) |
   | 10 closes under load (10,000 sends), dedup off | 283–423 re-registrations; about 327,000 messages written twice | 9–10 re-registrations; 639–939 written twice (the receipts a close drops) |
   | 50 send errors under load (10,000 sends), dedup on | 142–283 re-registrations | 0 re-registrations, 0 lost |
   | 50 send errors under load, dedup off | 287,462 messages written twice before the broker ran out of memory | 0 written twice |
   
   **What we run instead.** Nothing reconnects on either path.
   
   - **A send error leaves the producer registered and the message pending.** `handleSendError`'s default branch hands the error to the producer, which only logs it. A 5.0 broker closes the producer itself after every send error that cost a message:
     - a send to a producer the broker is closing ("Producer is closed");
     - a fenced topic, which happens only while the broker closes or deletes the topic, or after a failed write, and those disconnect its producers;
     - a failed write itself (`PersistentTopic.addFailed` fences the topic and disconnects its producers).
   
     The client then reconnects on `CLOSE_PRODUCER` and resends, as today. "Cannot determine…" answers a resend whose original is still being written, and the original's receipt follows. A checksum or metadata error cannot be fixed by a resend, so the message waits for its send timeout.
   - **An out-of-order receipt is kept with its pending message.** The message completes once the messages ahead of it have, so callbacks keep their order. Until then, the message keeps its queue slot and its permit. We tried completing it on the spot, but that released the permit early. The queue then overfilled, and `PutUnsafe` waited inside `writeData` and dropped the queue's lock, so a write could miss the resend. We lost a message per run that way before we changed it. A reconnect does not resend a message whose receipt has come.
   
   **This issue's case still works.**
   
   - A forced delete still closes its producer through the broker's `CLOSE_PRODUCER`, and the reconnect fails with `TopicNotFound`.
   - In our cluster test (8 senders on the deleted topic, 3 deletes per run, and a second topic on the same connection), the second topic had 0 duplicates, 0 failed sends and 0 failed acks, in 5 of 5 runs.
   
   **Tests.**
   
   - Unit, no broker: `TestConnectionSendErrorKeepsItsProducer` and `TestReceiptAheadOfPendingIsKeptForItsMessage`.
   - On a broker:
     - `TestSendErrorsAndClosesUnderLoad` injects send errors or closes under load, with and without deduplication. A second producer on the same connection must not be disturbed.
     - `TestOneCloseOfABusyProducerSettles` checks that one close of a busy producer settles.
   
   We're happy to open a PR with this instead of the fix in the issue.
   


-- 
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