poorbarcode commented on code in PR #25805:
URL: https://github.com/apache/pulsar/pull/25805#discussion_r3286692042
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java:
##########
@@ -1720,12 +1720,26 @@ private ByteBuf processMessageChunk(ByteBuf
compressedPayload, MessageMetadata m
*/
void notifyPendingReceivedCallback(final Message<T> message, Exception
exception) {
if (pendingReceives.isEmpty()) {
+ if (getState() != State.Closing && getState() != State.Closed) {
+ log.error().attr("message", message)
+ .attr("pendingReceives-size", pendingReceives.size())
+ .log(" If you received this log, it means that you
encountered a bug: a message was"
+ + " dropped internally, the client-side will encounter
a crucial issue: this message will"
+ + " never be consumed until the consumer is restarted
or the topic is unloaded.");
+ }
return;
}
// fetch receivedCallback from queue
final CompletableFuture<Message<T>> receivedFuture =
nextPendingReceive();
if (receivedFuture == null) {
+ if (getState() != State.Closing && getState() != State.Closed) {
+ log.error().attr("message", message)
+ .attr("receivedFuture-polled-out", null)
Review Comment:
Improved
--
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]
