https://pulsar.apache.org/ logo
Join Slack
Powered by
<@U078Y59KC3B> What is the purpose of the broker c...
# ask-ai
t
@RunLLM What is the purpose of the broker configuration field 
preciseDispatcherFlowControl
?
r
The 
preciseDispatcherFlowControl
broker configuration enables more accurate dispatch rate limiting in Apache Pulsar by considering the average number of messages per entry when reading from BookKeeper. This helps prevent over-delivery, especially when batching is enabled, by calculating the number of entries to read as 
number of messages to read / average messages per entry
instead of simply counting entries as messages. Only one of 
preciseDispatcherFlowControl
or 
dispatchThrottlingOnBatchMessageEnabled
should be enabled at a time, as they are mutually exclusive. For example, with a rate limit of 10 messages/sec and an average of 6 messages per entry, the broker reads 2 entries per second if 
preciseDispatcherFlowControl
is enabled, closely matching the quota. Without it, the broker may read more entries and exceed the quota. See Pulsar Throttling Configurations for details. Would you like more detail? Sources: - Pulsar Throttling Configurations - Pulsar Throttling Limitations šŸ‘ 0 button šŸ‘Ž 0 button šŸ’” Hint: Mention @runllm in the thread for followups.
t
@RunLLM Can 
preciseDispatcherFlowControl
lead to a local backlog?
r
Yes, enabling 
preciseDispatcherFlowControl
can lead to a local backlog if the broker underestimates the number of messages per entry, causing it to read fewer entries than needed to satisfy consumer permits. This may result in messages accumulating locally because the dispatch rate is throttled more precisely and may not keep up with consumer demand if the average is underestimated or not yet initialized, especially on new topics or subscriptions. See Pulsar Throttling Configurations and related PR discussion. Would you like more detail? Sources: - Pulsar Throttling Configurations - PR #18581 discussion šŸ‘ 0 button šŸ‘Ž 0 button šŸ’” Hint: Mention @runllm in the thread for followups.
t
@RunLLM Is 
preciseDispatcherFlowControl
a dynamic configuration meaning it can be changed without restarting the broker?
r
Yes, 
preciseDispatcherFlowControl
is a dynamic configuration, so it can be changed at runtime without restarting the broker. You can update it using the dynamic config update APIs or CLI tools like 
pulsar-admin brokers update-dynamic-config
. See Pulsar Throttling Configurations and Admin API Brokers. Would you like more detail? Sources: - Pulsar Throttling Configurations šŸ‘ 0 button šŸ‘Ž 0 button šŸ’” Hint: Mention @runllm in the thread for followups.
l
usually it's better to enable 
preciseDispatcherFlowControl
and I think it should be Pulsar's default. bumping 
dispatcherMaxReadBatchSize
to 
1000
could be useful if Pulsar isn't dispatching fast enough.
btw. dispatcherMaxReadBatchSize related issue: https://github.com/apache/pulsar/issues/24695
t
Ok thank you, that's good to know we have 
preciseDispatcherFlowControl
enabled but I'm seeing a backlog building slowly (production) Messages are produced by a job hourly, sending messages by batch and we have a dispatch rate (16 partition, rate at 
5
/ second). Once it has finished reading the messages over the course of ~45 mins, there is 100+ messages still in the local backlog each time. I'm going to look at 
dispatcherMaxReadBatchSize
, thanks! Would setting it as a dynamic config work (trying to prevent restarting all the brokers)?
Copy code
pulsar-admin brokers update-dynamic-config --config dispatcherMaxReadBatchSize --value 1000
Also we are on 
4.0.6
I'm still investigating it but I'm wondering if the backlog metrics (called 
local backlog
in the dashboard) is off. Looking at the pulsar admin command, I see a backlog for a 
partition-0
, but the consumers are idle as if there is nothing else to read from the topics after they are done with the bacth that was just sent
l
There are 2 bugs that were fixed recently
what subscription type do you have?
t
We have a shared subscription (multiple consumers)
Looking at the partitioned topic stats, we have out subscription with a backlog, but we do have another one 
__compaction
which I understand is related to Pulsar reclaiming space (deleted of acked messages - ledgers)
Copy code
"__compaction" : {
      "msgRateOut" : 0.0,
      "msgThroughputOut" : 0.0,
      "bytesOutCounter" : 199818,
      "msgOutCounter" : 5190,
      "msgRateRedeliver" : 0.0,
      "messageAckRate" : 0.0,
      "chunkedMessageRate" : 0.0,
      "msgBacklog" : 716,
      "backlogSize" : 5874706,
      "earliestMsgPublishTimeInBacklog" : 0,
      "msgBacklogNoDelayed" : 716,
      "blockedSubscriptionOnUnackedMsgs" : false,
      "msgDelayed" : 0,
      "msgInReplay" : 0,
      "unackedMessages" : 0,
      "type" : "Exclusive",
      "msgRateExpired" : 0.0,
      "totalMsgExpired" : 0,
      "lastExpireTimestamp" : 0,
      "lastConsumedFlowTimestamp" : 0,
      "lastConsumedTimestamp" : 0,
      "lastAckedTimestamp" : 0,
      "lastMarkDeleteAdvancedTimestamp" : 0,
      "consumers" : [ ],
      "isDurable" : true,
      "isReplicated" : false,
      "allowOutOfOrderDelivery" : false,
      "consumersAfterMarkDeletePosition" : { },
      "drainingHashesCount" : 0,
      "drainingHashesClearedTotal" : 0,
      "drainingHashesUnackedMessages" : 0,
      "nonContiguousDeletedMessagesRanges" : 0,
      "nonContiguousDeletedMessagesRangesSerializedSize" : 0,
      "delayedMessageIndexSizeInBytes" : 0,
      "subscriptionProperties" : { },
      "filterProcessedMsgCount" : 0,
      "filterAcceptedMsgCount" : 0,
      "filterRejectedMsgCount" : 0,
      "filterRescheduledMsgCount" : 0,
      "dispatchThrottledMsgEventsBySubscriptionLimit" : 0,
      "dispatchThrottledBytesEventsBySubscriptionLimit" : 0,
      "dispatchThrottledMsgEventsByTopicLimit" : 596,
      "dispatchThrottledBytesEventsByTopicLimit" : 0,
      "dispatchThrottledMsgEventsByBrokerLimit" : 0,
      "dispatchThrottledBytesEventsByBrokerLimit" : 0,
      "replicated" : false,
      "durable" : true
    },
This subscription also has a backlog, but does not seems to go back down. Looking at the compaction status of one of the partitions I can this back
Copy code
Error in compaction
null

Reason: Error compacting: org.apache.pulsar.common.util.FutureUtil$LowOverheadTimeoutException: Timeout
So I'm wondering if errors from compaction creates a backlog for the 
__compaction
subscription as well. There are no more consumers, I tried to unsubscribe, hoping to clean this back log but I got this error
Copy code
Stacktrace:

org.apache.pulsar.client.admin.PulsarAdminException$ServerSideErrorException: 
 --- An unexpected error occurred in the server ---

Message: Timeout

Stacktrace:

org.apache.pulsar.common.util.FutureUtil$LowOverheadTimeoutException: Timeout
	at org.apache.pulsar.compaction.PublishingOrderCompactor.phaseOneLoop(...)(Unknown Source)

	at org.apache.pulsar.client.admin.internal.BaseResource.getApiException(BaseResource.java:272)
	at org.apache.pulsar.client.admin.internal.BaseResource$4.failed(BaseResource.java:237)
	at org.glassfish.jersey.client.JerseyInvocation$1.failed(JerseyInvocation.java:898)
	at org.glassfish.jersey.client.JerseyInvocation$1.completed(JerseyInvocation.java:879)
	at org.glassfish.jersey.client.ClientRuntime.processResponse(ClientRuntime.java:232)
	at org.glassfish.jersey.client.ClientRuntime.access$200(ClientRuntime.java:62)
	at org.glassfish.jersey.client.ClientRuntime$2.lambda$response$0(ClientRuntime.java:176)
	at org.glassfish.jersey.internal.Errors$1.call(Errors.java:248)
	at org.glassfish.jersey.internal.Errors$1.call(Errors.java:244)
	at org.glassfish.jersey.internal.Errors.process(Errors.java:292)
	at org.glassfish.jersey.internal.Errors.process(Errors.java:274)
	at org.glassfish.jersey.internal.Errors.process(Errors.java:244)
	at org.glassfish.jersey.process.internal.RequestScope.runInScope(RequestScope.java:288)
	at org.glassfish.jersey.client.ClientRuntime$2.response(ClientRuntime.java:176)
	at org.apache.pulsar.client.admin.internal.http.AsyncHttpConnector.lambda$apply$1(AsyncHttpConnector.java:278)
	at java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(Unknown Source)
	at java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(Unknown Source)
	at java.base/java.util.concurrent.CompletableFuture.postComplete(Unknown Source)
	at java.base/java.util.concurrent.CompletableFuture.complete(Unknown Source)
	at org.apache.pulsar.client.admin.internal.http.AsyncHttpConnector.lambda$retryOperation$4(AsyncHttpConnector.java:337)
	at java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(Unknown Source)
	at java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(Unknown Source)
	at java.base/java.util.concurrent.CompletableFuture.postComplete(Unknown Source)
	at java.base/java.util.concurrent.CompletableFuture.cancel(Unknown Source)
	at org.apache.pulsar.client.admin.internal.http.AsyncHttpConnector.lambda$executeRequest$10(AsyncHttpConnector.java:406)
	at java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(Unknown Source)
	at java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(Unknown Source)
	at java.base/java.util.concurrent.CompletableFuture.postComplete(Unknown Source)
	at java.base/java.util.concurrent.CompletableFuture.complete(Unknown Source)
	at com.spotify.futures.ConcurrencyReducer.lambda$invoke$0(ConcurrencyReducer.java:173)
	at java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(Unknown Source)
	at java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(Unknown Source)
	at java.base/java.util.concurrent.CompletableFuture.postComplete(Unknown Source)
	at java.base/java.util.concurrent.CompletableFuture.complete(Unknown Source)
	at org.asynchttpclient.netty.NettyResponseFuture.loadContent(NettyResponseFuture.java:222)
	at org.asynchttpclient.netty.NettyResponseFuture.done(NettyResponseFuture.java:257)
	at org.asynchttpclient.netty.handler.AsyncHttpClientHandler.finishUpdate(AsyncHttpClientHandler.java:241)
	at org.asynchttpclient.netty.handler.HttpHandler.handleChunk(HttpHandler.java:113)
	at org.asynchttpclient.netty.handler.HttpHandler.handleRead(HttpHandler.java:142)
	at org.asynchttpclient.netty.handler.AsyncHttpClientHandler.channelRead(AsyncHttpClientHandler.java:78)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
	at io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:107)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
	at io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelRead(CombinedChannelDuplexHandler.java:436)
	at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:346)
	at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:318)
	at io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:251)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:442)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
	at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1357)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:440)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:868)
	at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
	at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:796)
	at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:732)
	at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:658)
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:562)
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:998)
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	at java.base/java.lang.Thread.run(Unknown Source)
Caused by: javax.ws.rs.InternalServerErrorException: HTTP 500 {"reason":"\n --- An unexpected error occurred in the server ---\n\nMessage: Timeout\n\nStacktrace:\n\norg.apache.pulsar.common.util.FutureUtil$LowOverheadTimeoutException: Timeout\n\tat org.apache.pulsar.compaction.PublishingOrderCompactor.phaseOneLoop(...)(Unknown Source)\n"}
	at org.glassfish.jersey.client.JerseyInvocation.convertToException(JerseyInvocation.java:977)
	at org.glassfish.jersey.client.JerseyInvocation.access$700(JerseyInvocation.java:82)
	... 64 more
l
Compacted topic?
t
We tried to compact (triggered with pulsar-admin), hoping it would clean (instead of waiting for the compaction to trigger) the data out of acked messages. Reading with pulsar -admin we could see messages that we think were already processed and acked in our platform
l
Compaction is for a different purpose
t
Is compaction with pulsar-admin different from minor and major compaction (deletes ledgers)?
l
Yes, different
t
Ok I do remeber compaction of a topic, based on message keys to get the latest value
l
Yes, it's key based. Fairly heavy process
t
We don't set any keys in our messages (don't have a use for it) so this subscription is just useless basically now
l
Have to step out for a while now, bbl
šŸ‘ 1
Yes, compaction without keys could result in a lot of extra CPU and memory being spent
šŸ‘ 1
Open in Slack
PreviousNext
Powered by