[I] [Bug] Chunk message is incomplete when resending to a different cluster [pulsar]

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

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

   ### Search before asking
   
   - [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 don't get bug fixes. I will attempt to reproduce the issue on a supported version of Pulsar client and Pulsar broker.
   
   
   ### Version
   
   - Pulsar client: 4.0.2
   - Java: 17.0.x
   - Broker: 4.0.3
   
   
   ### Minimal reproduce step
   
   ```java
   @Test
   public void chunkMessage() throws Exception {
       String r1="pulsar://localhost:6650";
       String r2="pulsar://localhost:6651";
       @Cleanup
       PulsarClient pulsarClient = PulsarClient.builder()
               .serviceUrl(r1)
               .build();
       String topic = "test" + System.nanoTime();
       // Send a message to the r1 cluster.
       @Cleanup
       Producer<byte[]> producer = pulsarClient.newProducer()
               .enableBatching(false)
               .enableChunking(true)
               .chunkMaxMessageSize(10)
               .topic(topic).create();
       // When sending the chunk message, update the service URL to r2.
       // Assuming the chunk message [0,1,2,3,4,5], when the url to r2, and then the r1 holds [0,1,2], r2 holds [3,4,5].
       new Thread(() -> {
           try {
               System.out.println("sleep...");
               Thread.sleep(500);
               System.out.println("update...");
               // Update the service URL to r2.
               pulsarClient.updateServiceUrl(r2);
           } catch (PulsarClientException | InterruptedException e) {
               throw new RuntimeException(e);
           }
       }).start();
       try {
           System.out.println("send...");
           MessageId send = producer.send(new byte[4000000]);
           System.out.println(send);
           System.out.println("send ok...");
       } catch (Exception e) {
           throw new RuntimeException(e);
       }
   
       // Try to receive the message from r2.
       @Cleanup
       PulsarClient pulsarClient2 = PulsarClient.builder()
               .serviceUrl(r2)
               .build();
       @Cleanup
       Consumer<byte[]> consumer = pulsarClient2.newConsumer()
               .topic(topic)
               .subscriptionName("test")
               .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
               .subscribe();
       Message<byte[]> receive = consumer.receive(30, TimeUnit.SECONDS);
       assertNotNull(receive); // Got null.
   }
   ```
   
   ### What did you expect to see?
   
   I can receive a chunk message from the second cluster.
   
   ### What did you see instead?
   
   I cannot receive a chunk message from the second cluster.
   
   ### Anything else?
   
   _No response_
   
   ### 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] Chunk message is incomplete when resending to a different cluster [pulsar]

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

merlimat commented on issue #24153:
URL: https://github.com/apache/pulsar/issues/24153#issuecomment-6378785742

   Confirmed on master: on reconnect, `ProducerImpl.recoverProcessOpSendMsgFrom()` (pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java:2928) resends only the operations still pending, and the chunks of a chunked message that the first cluster already acknowledged are no longer in that queue. When `updateServiceUrl()` moves the producer to another cluster mid-message, the new cluster gets only the trailing chunks, so the message can't be assembled there. Likely fix: when the producer reconnects to a different cluster (service URL changed) while a chunked message is partly acknowledged, resend that message from its first chunk (keeping the chunk ctx/payload until the last chunk is acknowledged), or fail it so the application can resend.


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