[
https://issues.apache.org/jira/browse/KAFKA-16876?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17851514#comment-17851514
]
Ganesh Sadanala edited comment on KAFKA-16876 at 6/3/24 4:21 AM:
-----------------------------------------------------------------
[~rohanpd] Thank your for confirming it! I see the flow is from
`TaskManager#handleRevocation` -> `TaskManager#prepareCommitAndAddOffsetsToMap`
-> `StreamTask#prepareCommit` -> `StreamTask#flush` ->
`ProcessorStateManager#flushCache`
and registered state stores are iterated inside it, I see that in your case
TaskMigratedException is thrown and is caught inside here:
{code:java}
catch (final RuntimeException exception) {
if (firstException == null) {
// do NOT wrap the error if it is actually caused by Streams itself
if (exception instanceof StreamsException) {
firstException = exception;
} {code}
and finally it is thrown to `StreamTask#flush` method, where I see it is not
caught/handled. Hence, the entire flow leads to Runtime errors and all the
active tasks are not revoked. Please correct me if I am wrong.
So you would want it to be handled inside the `StreamTask#flush` method
appropriately?
Also could you guide me how you produced those exceptions, I want to produce
them in my local to get a better picture.
Anything else you want to share will be beneficial.
Thank you!
was (Author: JIRAUSER305566):
[~rohanpd] Thank your for confirming it! I see the flow is from
`TaskManager#handleRevocation` -> `TaskManager#prepareCommitAndAddOffsetsToMap`
-> `StreamTask#prepareCommit` -> `StreamTask#flush` ->
`ProcessorStateManager#flushCache`
and registered state stores are iterated inside it, I see that in your case
TaskManagerException is thrown and is caught inside here:
{code:java}
catch (final RuntimeException exception) {
if (firstException == null) {
// do NOT wrap the error if it is actually caused by Streams itself
if (exception instanceof StreamsException) {
firstException = exception;
} {code}
and finally it is thrown to `StreamTask#flush` method, where I see it is not
caught/handled. Hence, the entire flow leads to Runtime errors and all the
active tasks are not revoked. Please correct me if I am wrong.
So you would want it to be handled inside the `StreamTask#flush` method
appropriately?
Also could you guide me how you produced those exceptions, I want to produce
them in my local to get a better picture.
Anything else you want to share will be beneficial.
Thank you!
> TaskManager.handleRevocation doesn't handle errors thrown from
> task.prepareCommit
> ---------------------------------------------------------------------------------
>
> Key: KAFKA-16876
> URL: https://issues.apache.org/jira/browse/KAFKA-16876
> Project: Kafka
> Issue Type: Bug
> Components: streams
> Affects Versions: 3.6.0
> Reporter: Rohan Desai
> Assignee: Ganesh Sadanala
> Priority: Minor
>
> `TaskManager.handleRevocation` does not handle exceptions thrown by
> `task.prepareCommit`. In the particular instance I observed, `pepareCommit`
> flushed caches which led to downstream `producer.send` calls that threw a
> `TaskMigratedException`. This means that the tasks that need to be revoked
> are not suspended by `handleRevocation`. `ConsumerCoordinator` stores the
> thrown exception and then moves on to the other task assignment callbacks.
> One of these - `StreamsPartitionAssigner.onCommit` tries to close the tasks
> and raises an `IllegalStateException`. Fortunately, it dirty-closes the tasks
> if close fails so we don't leak any tasks. I think there's maybe two bugs
> here:
> # `TaskManager.handleRevocation` should handle errors from `prepareCommit`.
> It should try not to leave any revoked tasks in an unsuspended state.
> # The `ConsumerCoordinator` just throws the first exception that it sees.
> But it seems bad to throw the `TaskMigratedException` and drop the
> `IllegalStateException` (though in this case I think its relatively benign).
> I think on `IllegalStateException` we really want the streams thread to exit.
> One idea here is to have `ConsumerCoordinator` throw an exception type that
> includes the other exceptions that it has seen in another field. But this
> breaks the contract for clients that catch specific exceptions. I'm not sure
> of a clean solution, but I think its at least worth recording that it would
> be preferable to have the caller of `poll` handle all the thrown exceptions
> rather than just the first one.
>
> Here is the IllegalStateException stack trace I observed:
> {code:java}
> [ 508.535] [service_application2] [inf] [ERROR] 2024-05-30 06:35:04.556
> [e2e-c0a9810b-8b09-46bd-a6d0-f2678ce0a1f3-StreamThread-1] TaskManager -
> stream-thread [e2e-c0a9810b-8b09-46bd-a6d0-f2678ce0a1f3-St
> reamThread-1] Failed to close task 0_3 cleanly. Attempting to close remaining
> tasks before re-throwing:
> [ 508.535] [service_application2] [inf]
> java.lang.IllegalStateException: Illegal state RUNNING while closing active
> task 0_3
> [ 508.535] [service_application2] [inf] at
> org.apache.kafka.streams.processor.internals.StreamTask.close(StreamTask.java:673)
> ~[kafka-streams-3.6.0.jar:?]
> [ 508.535] [service_application2] [inf] at
> org.apache.kafka.streams.processor.internals.StreamTask.closeClean(StreamTask.java:546)
> ~[kafka-streams-3.6.0.jar:?]
> [ 508.535] [service_application2] [inf] at
> org.apache.kafka.streams.processor.internals.TaskManager.closeTaskClean(TaskManager.java:1295)
> ~[kafka-streams-3.6.0.jar:?]
> [ 508.535] [service_application2] [inf] at
> org.apache.kafka.streams.processor.internals.TaskManager.closeAndRecycleTasks(TaskManager.java:630)
> [kafka-streams-3.6.0.jar:?]
> [ 508.535] [service_application2] [inf] at
> org.apache.kafka.streams.processor.internals.TaskManager.handleAssignment(TaskManager.java:350)
> [kafka-streams-3.6.0.jar:?]
> [ 508.535] [service_application2] [inf] at
> org.apache.kafka.streams.processor.internals.StreamsPartitionAssignor.onAssignment(StreamsPartitionAssignor.java:1381)
> [kafka-streams-3.6.0.jar:?]
> [ 508.535] [service_application2] [inf] at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.invokeOnAssignment(ConsumerCoordinator.java:315)
> [kafka-clients-3.6.0.jar:?]
> [ 508.535] [service_application2] [inf] at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:469)
> [kafka-clients-3.6.0.jar:?]
> [ 508.535] [service_application2] [inf] at
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:478)
> [kafka-clients-3.6.0.jar:?]
> [ 508.535] [service_application2] [inf] at
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:389)
> [kafka-clients-3.6.0.jar:?]
> [ 508.535] [service_application2] [inf] at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:564)
> [kafka-clients-3.6.0.jar:?]
> [ 508.535] [service_application2] [inf] at
> org.apache.kafka.clients.consumer.KafkaConsumer.updateAssignmentMetadataIfNeeded(KafkaConsumer.java:1220)
> [kafka-clients-3.6.0.jar:?]
> [ 508.535] [service_application2] [inf] at
> org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1179)
> [kafka-clients-3.6.0.jar:?]
> [ 508.535] [service_application2] [inf] at
> org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1159)
> [kafka-clients-3.6.0.jar:?]
> [ 508.535] [service_application2] [inf] at
> dev.responsive.kafka.internal.clients.DelegatingConsumer.poll(DelegatingConsumer.java:94)
> [kafka-client-0.24.0-dc9acd1.jar:?]
> [ 508.535] [service_application2] [inf] at
> org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:1014)
> [kafka-streams-3.6.0.jar:?]
> [ 508.535] [service_application2] [inf] at
> org.apache.kafka.streams.processor.internals.StreamThread.pollPhase(StreamThread.java:954)
> [kafka-streams-3.6.0.jar:?]
> [ 508.535] [service_application2] [inf] at
> org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:766)
> [kafka-streams-3.6.0.jar:?]
> [ 508.535] [service_application2] [inf] at
> org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:617)
> [kafka-streams-3.6.0.jar:?]
> [ 508.535] [service_application2] [inf] at
> org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:579)
> [kafka-streams-3.6.0.jar:?] {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
