[jira] [Comment Edited] (SPARK-27720) ConcurrentModificationException on closing DirectKafkaInputDStream

["Gabor Somogyi (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/SPARK-27720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16840438#comment-16840438 ] 

Gabor Somogyi edited comment on SPARK-27720 at 5/15/19 2:12 PM:
----------------------------------------------------------------

[~ov7a]
Kafka normally does the following in case of consumers (in an oversimplified way):
{code:java}
def someOperation() = {
  try {
    acquire() // This saves current thread Id
    execute()
  } finally {
    release() // This removes the saved thread Id
  }
}
{code}
If the execution is in the execute block and suddenly another thread comes and tries to do an operation (for example close) then the second thread's acquire throws the mentioned exception.

Some stacktrace and/or minimal steps to reproduce would be good to double check what's going on.



was (Author: gsomogyi):
[~ov7a]
Kafka normally does the following in case of consumers (in an oversimplified way):
{code:java}
def someOperation() = {
  try {
    acquire() // This saves current thread Id
    execute()
  } finally {
    release()
  }
}
{code}
If the execution is in the execute block and suddenly another thread comes and tries to do an operation (for example close) then the second thread's acquire throws the mentioned exception.

Some stacktrace and/or minimal steps to reproduce would be good to double check what's going on.


> ConcurrentModificationException on closing DirectKafkaInputDStream
> ------------------------------------------------------------------
>
>                 Key: SPARK-27720
>                 URL: https://issues.apache.org/jira/browse/SPARK-27720
>             Project: Spark
>          Issue Type: Bug
>          Components: DStreams
>    Affects Versions: 2.4.1
>            Reporter: ov7a
>            Priority: Minor
>
> If a DirectKafkaInputDStream is started in one thread and is being stopped in another thread (e.g. by shutdown hook) a java.util.ConcurrentModificationException (KafkaConsumer is not safe for multi-threaded access) is thrown.
> This happens even if "spark.streaming.kafka.consumer.cache.enabled" is set to "false".
>  



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org