[
https://issues.apache.org/jira/browse/KAFKA-21171?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
sanghyeok An updated KAFKA-21171:
---------------------------------
Labels: client streams (was: )
> Repeated task offset snapshot rebuilding adds overhead with the Streams
> rebalance protocol
> ------------------------------------------------------------------------------------------
>
> Key: KAFKA-21171
> URL: https://issues.apache.org/jira/browse/KAFKA-21171
> Project: Kafka
> Issue Type: Improvement
> Reporter: sanghyeok An
> Assignee: sanghyeok An
> Priority: Minor
> Labels: client, streams
>
> h2. Summary
> With {{{}group.protocol=streams{}}}, {{StreamThread}} rebuilds the task
> offset sum snapshot on every processing loop iteration, even when the offsets
> and task states have not changed. This performs task enumeration, map
> copying, filtering, and allocation independently of whether the next
> heartbeat will report new offsets.
> A local JMH benchmark shows that the cost of one refresh grows approximately
> linearly with task count. In a separate 10-minute application profile, this
> path accounted for *580 of 1,207 Java execution samples (48.1%)* from the
> Streams processing thread, including descendant calls.
> h2. Current behavior
> {{TaskManager.maybeUpdateTaskOffsetSumSnapshot()}} performs the following
> work:
> # Copies the process-wide local offset map from {{{}StateDirectory{}}}.
> # Builds a map of this thread's tasks through {{{}allTasks(){}}}, including
> tasks held by the state updater and tasks awaiting initialization.
> # Removes running active tasks and overlays live changelog offset sums for
> other eligible tasks.
> # Builds a separate map with heartbeat task IDs and publishes a read-only
> snapshot.
> The snapshot is refreshed on each loop iteration in both processing-loop
> variants. Heartbeat code separately checks assignment changes, the offset
> reporting interval, warm-up progress, and whether the values changed. These
> checks limit offset transmission but do not avoid the preceding snapshot
> construction.
> The copied local offset map is in memory; this path does not read task state
> from disk on every iteration. Under {{{}group.protocol=classic{}}}, this
> snapshot refresh method returns immediately; classic reports offset sums
> through its subscription metadata path.
> h2. Environment
> * Kafka {{{}4.5.0-SNAPSHOT{}}}, checkout
> {{{}e0b738a78819d95dd7779ec3d98effbb9f6a0f9c{}}}.
> * macOS AArch64; JFR reports 14 hardware threads.
> * JMH: JMH 1.37, JDK 25.0.3.
> * Application profile: JDK 21.0.5+11-LTS.
> * Affected released versions have not been established.
> h2. JMH results
> {{TaskOffsetSumSnapshotBenchmark.refreshSnapshot}} measures one refresh with
> unchanged offsets. All tasks belong to one stream thread and each has one
> changelog partition.
> The fixture uses the real task manager, task registry, state manager, state
> updater, and local offset map. It does not start a broker or perform disk
> I/O. RUNNING tasks are registered as active tasks and produce an empty final
> snapshot. RESTORING tasks remain in the state updater input queue; the
> updater thread is not started.
> Settings: one worker, three forks, {{{}-Xms2g -Xmx2g{}}}, five 2-second
> warmup iterations and five 2-second measurement iterations per fork,
> {{{}avgt{}}}, and {{{}-prof gc{}}}. There are 15 measurements per parameter
> combination. Time errors below are the JMH 99.9% confidence interval
> half-widths.
> ||Task count||Task state||Time (us/op, mean ± error)||Allocation (B/op)||
> |0|RUNNING|0.257 ± 0.002|2,120.001|
> |0|RESTORING|0.259 ± 0.001|2,120.001|
> |100|RUNNING|8.481 ± 0.057|31,890.708|
> |100|RESTORING|22.821 ± 0.111|77,048.131|
> |1,000|RUNNING|89.110 ± 1.960|346,829.769|
> |1,000|RESTORING|248.191 ± 1.702|831,546.901|
> |5,000|RUNNING|429.778 ± 5.789|1,732,682.127|
> |5,000|RESTORING|1,304.563 ± 22.457|4,111,075.094|
> The RUNNING cases show that copying and filtering still incur a cost when all
> local entries are excluded from the final snapshot. Allocation is measured
> per refresh, not retained heap size.
> The results were collected in two runs: task count 1,000, and task counts
> 0/100/5,000. For example, the latter used:
> {code:bash}
> java -jar
> jmh-benchmarks/build/libs/kafka-jmh-benchmarks-4.5.0-SNAPSHOT-all.jar \
> TaskOffsetSumSnapshotBenchmark \
> -p taskCount=0,100,5000 -p taskState=RUNNING,RESTORING \
> -f 3 -wi 5 -i 5 -w 2s -r 2s \
> -prof gc -foe true \
> -rf json -rff jmh-benchmarks/build/task-offset-snapshot-scaling.json
> {code}
> Benchmark source: [^TaskOffsetSumSnapshotBenchmark.java]
> Raw results: [^task-offset-snapshot-1000.json],
> [^task-offset-snapshot-scaling.json]
> h2. Application profile and flame graph
> The application uses one sub-topology:
> {noformat}
> snapshot-bench-input
> -> groupByKey()
> -> count(persistent key-value store)
> {noformat}
> The configured test uses {{{}group.protocol=streams{}}},
> {{{}num.stream.threads=1{}}}, and {{{}poll.ms=10{}}}. JFR was recorded with
> {{{}settings=profile{}}}, a 60-second start delay, and a 600-second duration.
> The intended workload was 100 tasks with no continuous input; the exact
> assigned task count and input activity during this recording were not
> independently captured.
> The recording contains 1,225 {{jdk.ExecutionSample}} events, including 1,207
> from {{{}StreamThread-1{}}}. The flame graph uses these Java execution events
> only. Native-method, allocation, and waiting events are excluded; none of the
> recorded execution stacks are marked truncated.
> ||Processing-thread call path, including descendants||Samples||Share of 1,207
> samples||
> |TaskManager.maybeUpdateTaskOffsetSumSnapshot|580|48.1%|
> |TaskManager.process|255|21.1%|
> |StreamThread.maybeCommit|157|13.0%|
> |TaskManager.recordTaskProcessRatio|128|10.6%|
> |TaskManager.punctuate|39|3.2%|
> |pollPhase and other paths|48|4.0%|
> The {{process}} and {{maybeCommit}} paths include readiness checks; their
> presence does not establish that records were processed or commits completed.
> Within the 580 snapshot-path samples, the sampled call sites were:
> ||Snapshot operation||Samples||
> |allTasks() and task iteration preparation|284|
> |Copying the StateDirectory offset map|194|
> |Removing running active tasks from the copied map|98|
> |Other|4|
> Descendant frames include {{{}TaskId.hashCode(){}}}, {{HashMap}}
> insertion/removal, {{ConcurrentHashMap}} traversal, and state-updater task
> enumeration.
> *48.1% is a share of Java execution samples, not whole-process CPU
> utilization or the fraction of a CPU core consumed.*
> Separately, cumulative thread CPU counters in the first and last recorded
> thread dumps increased by 63.956 seconds over 543.632 seconds of elapsed time
> for the Streams processing thread. This corresponds to approximately *11.8%
> of one CPU core* over that interval. The sample share does not establish an
> exact CPU time or potential saving for the snapshot method.
> Flame graphs: [^stream-thread.svg], [^all-java.svg]. Download and open the
> SVG in a browser to zoom and search for
> {{{}maybeUpdateTaskOffsetSumSnapshot{}}}.
> h2. Interpretation and limits
> The microbenchmark demonstrates repeated work with unchanged inputs, and the
> application profile identifies snapshot construction as a prominent Java
> execution path in the measured run.
> An earlier 120-second recording with {{poll.ms=100}} contained only 27 Java
> execution samples and no snapshot-path samples. It is too sparse to establish
> the snapshot's CPU share. The application ID also changed between runs, and
> task/input conditions were not fully recorded, so these profiles are not a
> controlled comparison of poll intervals.
> The JMH and application runs used different JDK versions. No production
> throughput regression, lock contention bottleneck, or measured improvement is
> established by these results.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
