[
https://issues.apache.org/jira/browse/KAFKA-10134?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17190802#comment-17190802
]
Peter Davis commented on KAFKA-10134:
-------------------------------------
I'd like to see the title of this bug clarified: It is worse than just "High
CPU usage". I have a couple of Kafka Streams apps with a high number of
tasks/threads and this issue is causing infinite rebalance loops where the
entire *cluster stops processing and cannot successfully rebalance*. This
causes hard downtime. I've had to roll back to 2.4.1.
I'm currently working on building the patch, will test.
Late to the party since you've already got the fix in progress, but in case it
helps, I'd like to share what I'm seeing:
The rebalance failures seems to be associated the TimeoutExceptions,
DisconnectionExceptions and other side effects as noted in earlier comments.
When many StreamThreads are all spinning, then each time a rebalance is
attempted, when there are a large number of threads it is likely that _some_
thread will fail, and the rebalance never succeeds. The downward spiral begins
as ConsumerThreads become "fenced" and it triggers a full (not incremental)
rebalance, and eventually all data flow gets blocked. I've tried different
combinations of session.timeout.ms, rebalance.timeout.ms, max.poll.time.ms,
default.api.timeout.ms (as recommended in the text of the timeout exceptions)
to no avail.
Of my applications, the ones that are affected include
* one stateless app with num.stream.threads=24. With more than 1 instance
(2-4x=48-96 threads), it will often never rebalance correctly, or only after
multiple attempts (30+ minutes).
* one stateful app with 36 partitions of large-ish (500MB-1GB each) state
stores which can take a while to restore. This app successfully starts if I
shut down all instances, delete state stores, set initial rebalance delay, and
start all up simultaneously – but if any instance restarts or I attempt to
scale up later, then rebalance will never succeed. Additionally, when state
stores are reassigned, there are "LockExceptions" (DEBUG level logs) in a tight
loop, and the state stores fail to be closed cleanly, which forces the restore
process to begin all over again. The only way I can successfully do a rolling
restart is if I use static membership and increase the session timeout. If
there is only a single instance of the app, then it works with no problems (but
this is not a solution as I need multiple instances for scale).
Other side effects: the tight loop logs several DEBUG logs, which filled up log
storage and caused pod evictions, which caused state stores to become invalid
and restore (workaround: disable this logging).
Additionally, have seen the following exceptions sporadically, not sure if
these are separate bugs:
{{2020-08-31T00:40:47.786Z ERROR Uncaught stream processing error!
KafkaStreamsConfiguration java.lang.IllegalStateException: There are
insufficient bytes available to read assignment from the sync-group response
(actual byte size 0) , this is not expected; it is possible that the leader's
assign function is buggy and did not return any assignment for this member, or
*because static member is configured and the protocol is buggy* hence did not
get the assignment for this member}}
{{ at
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:367)}}
{{ at
org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:440)}}
{{ at
org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:359)}}
{{ at
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:513)}}
{{ at
org.apache.kafka.clients.consumer.KafkaConsumer.updateAssignmentMetadataIfNeeded(KafkaConsumer.java:1268)}}
{{ at
org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1230)}}
{{ at
org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1210)}}
{{ at
org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:766)}}
{{ at
org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:624)}}
{{ at
org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:551)}}
{{ at
org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:510)}}
{{2020-09-03T15:53:17.524Z ERROR Uncaught stream processing error!
KafkaStreamsConfiguration java.lang.IllegalStateException: Active task 3_0
should have been suspended}}
{{ at
org.apache.kafka.streams.processor.internals.TaskManager.handleAssignment(TaskManager.java:281)}}
{{ ... 13 common frames omitted}}
{{Wrapped by: java.lang.RuntimeException: Unexpected failure to close 1 task(s)
[[3_0]]. First unexpected exception (for task 3_0) follows.}}
{{ at
org.apache.kafka.streams.processor.internals.TaskManager.handleAssignment(TaskManager.java:349)}}
{{ at
org.apache.kafka.streams.processor.internals.StreamsPartitionAssignor.onAssignment(StreamsPartitionAssignor.java:1428)}}
{{ at
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.invokeOnAssignment(ConsumerCoordinator.java:279)}}
{{ at
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:421)}}
{{ ... 10 common frames omitted}}
{{Wrapped by: org.apache.kafka.common.KafkaException: User rebalance callback
throws an error}}
{{ at
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:436)}}
{{ at
org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:440)}}
{{ at
org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:359)}}
{{ at
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:513)}}
{{ at
org.apache.kafka.clients.consumer.KafkaConsumer.updateAssignmentMetadataIfNeeded(KafkaConsumer.java:1268)}}
{{ at
org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1230)}}
{{ at
org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1210)}}
{{ at
org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:766)}}
{{ at
org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:628)}}
{{ at
org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:551)}}
{{ at
org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:510)}}
> High CPU issue during rebalance in Kafka consumer after upgrading to 2.5
> ------------------------------------------------------------------------
>
> Key: KAFKA-10134
> URL: https://issues.apache.org/jira/browse/KAFKA-10134
> Project: Kafka
> Issue Type: Bug
> Components: clients
> Affects Versions: 2.5.0
> Reporter: Sean Guo
> Assignee: Guozhang Wang
> Priority: Blocker
> Fix For: 2.5.2, 2.6.1
>
> Attachments: consumer3.log.2020-08-20.log,
> consumer5.log.2020-07-22.log
>
>
> We want to utilize the new rebalance protocol to mitigate the stop-the-world
> effect during the rebalance as our tasks are long running task.
> But after the upgrade when we try to kill an instance to let rebalance happen
> when there is some load(some are long running tasks >30S) there, the CPU will
> go sky-high. It reads ~700% in our metrics so there should be several threads
> are in a tight loop. We have several consumer threads consuming from
> different partitions during the rebalance. This is reproducible in both the
> new CooperativeStickyAssignor and old eager rebalance rebalance protocol. The
> difference is that with old eager rebalance rebalance protocol used the high
> CPU usage will dropped after the rebalance done. But when using cooperative
> one, it seems the consumers threads are stuck on something and couldn't
> finish the rebalance so the high CPU usage won't drop until we stopped our
> load. Also a small load without long running task also won't cause continuous
> high CPU usage as the rebalance can finish in that case.
>
> "executor.kafka-consumer-executor-4" #124 daemon prio=5 os_prio=0
> cpu=76853.07ms elapsed=841.16s tid=0x00007fe11f044000 nid=0x1f4 runnable
> [0x00007fe119aab000]"executor.kafka-consumer-executor-4" #124 daemon prio=5
> os_prio=0 cpu=76853.07ms elapsed=841.16s tid=0x00007fe11f044000 nid=0x1f4
> runnable [0x00007fe119aab000] java.lang.Thread.State: RUNNABLE at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:467)
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.updateAssignmentMetadataIfNeeded(KafkaConsumer.java:1275)
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1241)
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1216)
> at
>
> By debugging into the code we found it looks like the clients are in a loop
> on finding the coordinator.
> I also tried the old rebalance protocol for the new version the issue still
> exists but the CPU will be back to normal when the rebalance is done.
> Also tried the same on the 2.4.1 which seems don't have this issue. So it
> seems related something changed between 2.4.1 and 2.5.0.
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
