Guys, I've notice quite a weird behaviour in my pulsar setup. I have two pods in kubernetes which consume messages from a topic. Both consumers have different randomly generated subscription ids and exclusive subscription types. It's been working well for a while but now I encountered an issue that one consumer just randomly stopped consuming messages and I don't understand what's the problem. I tried to get stats of this topic but it shows that lastConsumedTimestamp is the same in two pods. Could you help me please with it? What can be a reason of this behaviour and how can I debug it?

Can you share the output of the topic stats command? It might provide more information. Thanks.

TopicStatsImpl(count=0, msgRateIn=0.033333334581666714, msgThroughputIn=9.200000344540012, msgRateOut=0.06666667084444476, msgThroughputOut=18.400001153066754, bytesInCounter=739315, msgInCounter=2500, bytesOutCounter=1481302, msgOutCounter=5009, averageMsgSize=275.99999999999994, msgChunkPublished=false, storageSize=0, backlogSize=0, publishRateLimitedTimes=0, earliestMsgPublishTimeInBacklogs=0, offloadedStorageSize=0, lastOffloadLedgerId=0, lastOffloadSuccessTimeStamp=0, lastOffloadFailureTimeStamp=0, ongoingTxnCount=0, abortedTxnCount=0, committedTxnCount=0, publishers=[PublisherStatsImpl(count=0, accessMode=Shared, msgRateIn=0.0, msgThroughputIn=0.0, averageMsgSize=0.0, chunkedMessageRate=0.0, producerId=1, supportsPartialProducer=false, producerName=pulsar-15-17, address=/10.42.2.79:45030, connectedSince=2024-06-10T075939.948542866Z, clientVersion=Pulsar-Java-v3.2.2, metadata={}), PublisherStatsImpl(count=0, accessMode=Shared, msgRateIn=0.0, msgThroughputIn=0.0, averageMsgSize=0.0, chunkedMessageRate=0.0, producerId=1, supportsPartialProducer=false, producerName=pulsar-15-19, address=/10.42.26.94:42764, connectedSince=2024-06-10T082611.203106689Z, clientVersion=Pulsar-Java-v3.2.2, metadata={})], publishersMap={}, waitingPublishers=0, subscriptions={d8d622c0-1ede-4fd0-81c9-83bd01fca7c0=SubscriptionStatsImpl(msgRateOut=0.03333333638000028, msgThroughputOut=9.200000840880076, bytesOutCounter=39318, msgOutCounter=121, msgRateRedeliver=0.0, messageAckRate=0.0, chunkedMessageRate=0, msgBacklog=0, backlogSize=0, earliestMsgPublishTimeInBacklog=0, msgBacklogNoDelayed=0, blockedSubscriptionOnUnackedMsgs=false, msgDelayed=0, unackedMessages=0, type=Exclusive, activeConsumerName=null, msgRateExpired=0.0, totalMsgExpired=0, lastExpireTimestamp=0, lastConsumedFlowTimestamp=0, lastConsumedTimestamp=0, lastAckedTimestamp=0, lastMarkDeleteAdvancedTimestamp=0, consumers=[ConsumerStatsImpl(msgRateOut=0.03333333638000028, msgThroughputOut=9.200000840880076, bytesOutCounter=39318, msgOutCounter=121, msgRateRedeliver=0.0, messageAckRate=0.0, chunkedMessageRate=0.0, consumerName=ebccb, availablePermits=879, unackedMessages=0, avgMessagesPerEntry=1, blockedConsumerOnUnackedMsgs=false, readPositionWhenJoining=null, address=/10.42.2.79:45030, connectedSince=2024-06-10T075939.961179119Z, clientVersion=Pulsar-Java-v3.2.2, lastAckedTimestamp=0, lastConsumedTimestamp=1718033077289, lastConsumedFlowTimestamp=1718006379962, keyHashRanges=null, metadata={})], isDurable=false, isReplicated=false, allowOutOfOrderDelivery=false, keySharedMode=null, consumersAfterMarkDeletePosition={}, nonContiguousDeletedMessagesRanges=0, nonContiguousDeletedMessagesRangesSerializedSize=0, delayedMessageIndexSizeInBytes=0, bucketDelayedIndexStats={}, subscriptionProperties={}, filterProcessedMsgCount=0, filterAcceptedMsgCount=0, filterRejectedMsgCount=0, filterRescheduledMsgCount=0), 7014ec53-1a25-4673-92a4-cb7db0d105ab=SubscriptionStatsImpl(msgRateOut=0.03333333446444448, msgThroughputOut=9.200000312186678, bytesOutCounter=37175, msgOutCounter=114, msgRateRedeliver=0.0, messageAckRate=0.0, chunkedMessageRate=0, msgBacklog=0, backlogSize=0, earliestMsgPublishTimeInBacklog=0, msgBacklogNoDelayed=0, blockedSubscriptionOnUnackedMsgs=false, msgDelayed=0, unackedMessages=0, type=Exclusive, activeConsumerName=null, msgRateExpired=0.0, totalMsgExpired=0, lastExpireTimestamp=0, lastConsumedFlowTimestamp=0, lastConsumedTimestamp=0, lastAckedTimestamp=0, lastMarkDeleteAdvancedTimestamp=0, consumers=[ConsumerStatsImpl(msgRateOut=0.03333333446444448, msgThroughputOut=9.200000312186678, bytesOutCounter=37175, msgOutCounter=114, msgRateRedeliver=0.0, messageAckRate=0.0, chunkedMessageRate=0.0, consumerName=aefdf, availablePermits=886, unackedMessages=0, avgMessagesPerEntry=1, blockedConsumerOnUnackedMsgs=false, readPositionWhenJoining=null, address=/10.42.26.94:42764, connectedSince=2024-06-10T082611.214930299Z, clientVersion=Pulsar-Java-v3.2.2, lastAckedTimestamp=0, lastConsumedTimestamp=1718033077289, lastConsumedFlowTimestamp=1718007971215, keyHashRanges=null, metadata={})], isDurable=false, isReplicated=false, allowOutOfOrderDelivery=false, keySharedMode=null, consumersAfterMarkDeletePosition={}, nonContiguousDeletedMessagesRanges=0, nonContiguousDeletedMessagesRangesSerializedSize=0, delayedMessageIndexSizeInBytes=0, bucketDelayedIndexStats={}, subscriptionProperties={}, filterProcessedMsgCount=0, filterAcceptedMsgCount=0, filterRejectedMsgCount=0, filterRescheduledMsgCount=0)}, replication={eu-to-uk=ReplicatorStatsImpl(msgRateIn=0.0, msgThroughputIn=0.0, msgRateOut=0.0, msgThroughputOut=0.0, msgRateExpired=0.0, replicationBacklog=0, connected=true, replicationDelayInSeconds=0, inboundConnection=null, inboundConnectedSince=null, outboundConnection=[id: 0xa9f19ebe, L/10.42.26.5750918 - Rpulsar uk.traveltime.dev/198.244.201.546651], outboundConnectedSince=2024-06-06T192142.450681195Z), eu-to-us-west=ReplicatorStatsImpl(msgRateIn=0.0, msgThroughputIn=0.0, msgRateOut=0.0, msgThroughputOut=0.0, msgRateExpired=0.0, replicationBacklog=0, connected=true, replicationDelayInSeconds=0, inboundConnection=null, inboundConnectedSince=null, outboundConnection=[id: 0xfd0bd014, L/10.42.26.5739582 - Rpulsar us west.traveltime.dev/15.204.100.1636651], outboundConnectedSince=2024-06-06T192144.727861395Z)}, deduplicationStatus=null, topicEpoch=null, nonContiguousDeletedMessagesRanges=0, nonContiguousDeletedMessagesRangesSerializedSize=0, delayedMessageIndexSizeInBytes=0, bucketDelayedIndexStats={}, compaction=CompactionStatsImpl(lastCompactionRemovedEventCount=0, lastCompactionSucceedTimestamp=0, lastCompactionFailedTimestamp=0, lastCompactionDurationTimeInMills=0), ownerBroker=pulsar-broker-1.pulsar-broker.pulsar.svc.cluster.local:8080)

The

connectedSince=2024-06-10T08:26:11.214930299Z,

and

connectedSince=2024-06-10T07:59:39.948542866Z

look different for the consumers. So one of them is disconnected as far as the broker is concerned.

what can I do not to disconnect them? Cause I cannot redeploy pods at the same time

I would check the network connection between the pod and the broker.

I see, thanks. And I have one more question

These are the standard output metrics from the pulsar client, regardless of sync/async, etc.

but why are they equal to zero?

If the connection was broken between the broker and the client, then the acks you sent async would not reach the broker, and thus it would report zero acks received.

I see but if a connection is broken, is there a way how to find out about it on a client side?

If you are sending the messages async, you must attach an exception handler to the CompletableFuture object returned from the sendAsync() method call. Then, when the exception occurs in the future, that code will be invoked, and you will be informed.

yes, I have it and I see now that messages are consumed

When the messages were published to Pulsar, were they batched or not?

they weren't

I have a similar issue -- https://github.com/apache/pulsar/issues/21082#issuecomment-2155810564 There have been multiple such issues reported. We moved from 2.7.1 to 3.0.3 on the server side when the issue started. Even moving the client SDK (Java) to 3.0.4 did not help. Unloading the topic using the pulsar-admin CLI helps resolve the issue. But this happens randomly and infrequently.

Yes, I also saw this. @David K what could be the reason of such behaviour ?

It's hard to identify the root cause at this point. I would need to review the similar issues to see which of them have been resolved in the version of Pulsar that you are running.

pulsar client: "3.2.2" pulsar version: "3.0.5"

Non-persistent topics are a bit more cumbersome to troubleshoot. Is there any reason why you don’t use persistent topics?

speed

In scenarios where your consumers are caught up with the most recent message, the performance for persistent vs non-persistent should be identical due to the fact that they are both reading messages out of the in-memory cache on the broker.

so I can use persistent topics without journal disk, right?

Yes. That is an option

the problem is that we also need to write data in topics as fast as possible

What is your target latency for publishing?

2-5 milliseconds ideally

seems to be other related threads: https://github.com/apache/pulsar/issues/21104 https://github.com/apache/pulsar/issues/16511

@Daniel, in the ticket you reference, “The same example works when downgrading the client to 3.0.4”. Can you try downgrading your client from 3.2.2 to 3.0.4?

thanks for your advices. I'll try to downgrade and use persistent topics and come back to you

FYI, here are the publish latencies from our 2022 benchmark testing

@David K could you please give me a hint which config I should use to disable writing to a journal? I've used this one https://github.com/apache/bookkeeper/pull/2401/files#diff-c21e6118c062b6ee5639f47d3d0f15a8c8d4666b69279ccd8e605f2ba4da4908 but I'm not sure that Journal is disabled