You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@kafka.apache.org by "Matthias J. Sax (JIRA)" <ji...@apache.org> on 2017/05/03 22:09:04 UTC
[jira] [Commented] (KAFKA-5167) streams task gets stuck after
re-balance due to LockException
[ https://issues.apache.org/jira/browse/KAFKA-5167?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15995777#comment-15995777 ]
Matthias J. Sax commented on KAFKA-5167:
----------------------------------------
[~Narendra Kumar] Thanks for reporting this. We fixed couple of bugs with regard to rebalance and state directory locks in {{0.10.2.1}} that got released last week. Can you try the new release to see if it got already fixed there?
> streams task gets stuck after re-balance due to LockException
> -------------------------------------------------------------
>
> Key: KAFKA-5167
> URL: https://issues.apache.org/jira/browse/KAFKA-5167
> Project: Kafka
> Issue Type: Bug
> Components: streams
> Affects Versions: 0.10.2.0
> Reporter: Narendra Kumar
> Attachments: logs.txt
>
>
> During rebalance processor node's close() method gets called two times once from StreamThread.suspendTasksAndState() and once from StreamThread.closeNonAssignedSuspendedTasks(). I have some instance filed which I am closing in processor's close method. This instance's close method throws some exception if I call close more than once. Because of this exception, the Kafka streams does not attempt to close the statemanager ie. task.closeStateManager(true) is never called. When a task moves from one thread to another within same machine the task blocks trying to get lock on state directory which is still held by unclosed statemanager and keep throwing the following exception:
> 2017-04-30 12:34:17 WARN StreamThread:1214 - Could not create task 0_1. Will retry.
> org.apache.kafka.streams.errors.LockException: task [0_1] Failed to lock the state directory for task 0_1
> at org.apache.kafka.streams.processor.internals.ProcessorStateManager.<init>(ProcessorStateManager.java:100)
> at org.apache.kafka.streams.processor.internals.AbstractTask.<init>(AbstractTask.java:73)
> at org.apache.kafka.streams.processor.internals.StreamTask.<init>(StreamTask.java:108)
> at org.apache.kafka.streams.processor.internals.StreamThread.createStreamTask(StreamThread.java:864)
> at org.apache.kafka.streams.processor.internals.StreamThread$TaskCreator.createTask(StreamThread.java:1237)
> at org.apache.kafka.streams.processor.internals.StreamThread$AbstractTaskCreator.retryWithBackoff(StreamThread.java:1210)
> at org.apache.kafka.streams.processor.internals.StreamThread.addStreamTasks(StreamThread.java:967)
> at org.apache.kafka.streams.processor.internals.StreamThread.access$600(StreamThread.java:69)
> at org.apache.kafka.streams.processor.internals.StreamThread$1.onPartitionsAssigned(StreamThread.java:234)
> at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:259)
> at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:352)
> at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:303)
> at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:290)
> at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1029)
> at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:995)
> at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:592)
> at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:361)
--
This message was sent by Atlassian JIRA
(v6.3.15#6346)