wsz94 commented on pull request #17643: URL: https://github.com/apache/flink/pull/17643#issuecomment-958691412
> I don't think this fixed the data loss issue. Pulsar [internally wrap](https://github.com/apache/pulsar/blob/f0413b1adad395319e5595a863764f8af943dbe8/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java#L422) the CompletableFuture in `receiveAsync()` with a `CompletableFutureCancellationHandler`. The pending receiver would be automatically removed when meeting `TimeoutException`. > > That means we don't meet data loss when we don't manually cancel the future. > > @AHeise Can you confirm this? I think that you are wrong. As the source code, The pending receiver cannot remove the future with `CompletableFutureCancellationHandler`. What you used is `CompletableFuture.get(timeout)` , not the `CompletableFuture.orTimeout(timeout)`. If you don't manually cancel the future, the `future` will lost in connector, but still in the `pending receiver` queue in internal pulsar client. So the data lost. Please take a look. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
