Nicholas Telford created KAFKA-21372:
----------------------------------------
Summary: 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.4.0, 4.3.0, 4.2.0
Reporter: Nicholas Telford
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)
