[
https://issues.apache.org/jira/browse/KAFKA-21372?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Mohammad Reduwan reassigned KAFKA-21372:
----------------------------------------
Assignee: Mohammad Reduwan
> PartitionGroup#readyToProcess formats log messages on every call, and most of
> these messages are discarded
> ----------------------------------------------------------------------------------------------------------
>
> Key: KAFKA-21372
> URL: https://issues.apache.org/jira/browse/KAFKA-21372
> Project: Kafka
> Issue Type: Bug
> Components: streams
> Affects Versions: 4.2.0, 4.3.0, 4.4.0
> Reporter: Nicholas Telford
> Assignee: Mohammad Reduwan
> Priority: Minor
>
> h3. Problem
> KAFKA-19682 (#20693, first released in 4.2.0) added a log message to
> {{{}PartitionGroup#readyToProcess{}}}. The method returns this message in
> {{{}ReadyToProcessResult{}}}. It builds the message with {{String.format}}
> and a {{StringBuilder}} for each partition that has no buffered records:
> {code:java}
> final Long fetchedLag = fetchedLags.getOrDefault(partition, -1L);
> appendLog(logMessageBuilder, String.format("Partition %s has fetched lag of
> %d", partition, fetchedLag));
> ...
> appendLog(logMessageBuilder,
> String.format("Partition %s has current lag %d, but no data is buffered
> locally. Waiting to buffer some records.",
> partition, fetchedLag));
> ...
> appendLog(logMessageBuilder, String.format(
> "Partition %s has current lag 0 and current time is %d. " +
> "Waiting for new data to be produced for configured idle time %d
> (deadline is %d).",
> partition, wallClockTime, maxTaskIdleMs, deadline));
> {code}
> No log-level check protects this code. It runs each time
> {{StreamTask#isProcessable}} calls {{{}readyToProcess{}}}, and the
> {{TaskExecutor}} loop does this {*}for each record{*}. The code runs when a
> task has more than one input partition and at least one of those partitions
> has no buffered records. This is a normal condition for a task that reads a
> low-volume topic together with a high-volume topic.
> The caller then discards almost every message:
> * {{StreamTask#isProcessable}} discards the message when the task changes
> from READY to NOT_READY ({{{}wasReady{}}}).
> * {{StreamTask#maybeLogNotReady}} logs the message at INFO only when
> {{NOT_READY_LOG_INTERVAL_MS}} (120 s) passed since the last log. It discards
> all other messages.
> * When a partition reaches its idle deadline and the task becomes ready
> ({{{}enforced{}}}), the method returns {{{}ReadyToProcessResult(true,
> Optional.empty()){}}}. It discards the "has fetched lag" text that it already
> formatted for that partition.
> Before 4.2.0, this method logged the same states with
> {{{}logger.trace("...{}...", ...){}}}. That code formatted no text when TRACE
> was disabled.
> h3. Impact
> We upgraded a Kafka Streams application from 3.8 to 4.4.0. The application
> runs EOS v2 and has several sub-topologies with more than one input partition
> for each task.
> We recorded a 5-minute JFR profile on one instance during the catch-up after
> a restart. In this profile:
> * {{PartitionGroup#readyToProcess}} is 19.7% of the StreamThread CPU
> samples. {{String.format}} called from it is 15.1%, and this includes
> {{DecimalFormatSymbols.<init>}} and {{{}Formatter.parse{}}}.
> * {{ArrayList}} allocations from {{Formatter.parse}} are 13.7% of the
> sampled allocation weight.
> The 3.8 profile of the same application has no equivalent cost.
> h3. Proposed fix
> Do not build the message unless the caller will log it. Possible approaches:
> # Return the not-ready reason as structured data in
> {{{}ReadyToProcessResult{}}}: the partition, the fetched lag, the deadline
> and the reason. Let {{StreamTask#maybeLogNotReady}} format the text only when
> it logs.
> # Alternatively, make {{readyToProcess}} build the message only when the
> caller asks for it. For example, pass a flag that {{StreamTask}} sets when
> {{NOT_READY_LOG_INTERVAL_MS}} passed.
> Both approaches keep the log output of KAFKA-19682 and remove the cost from
> the processing loop.
> h3. Workaround
> {{max.task.idle.ms=-1}} makes {{readyToProcess}} return before the loop. But
> this setting also disables the wait for unfetched data, so a task can process
> records out of order. It is not a safe workaround for applications that need
> the default behaviour.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
