[jira] [Commented] (KAFKA-10340) Source connectors should report error when trying to produce records to non-existent topics instead of hanging forever
[
https://issues.apache.org/jira/browse/KAFKA-10340?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17282875#comment-17282875
]
Goltseva Taisiia commented on KAFKA-10340:
--
And also I cannot understand, why we even have the property
"topic.creation.enable" and do not create all topics by AdminClient? Isn't it a
better approach to create topics always explicitly using AdminClient than to
rely on auto creation?
And also I cannot see how can it has a bad impact on users. Why someone may
want to set "topic.creation.enable = false"?
According to KIP-158 we have an option to set replication.factor and partitions
to -1 to use the broker's default value. So why we cannot always create topics
by AdminClient and use these broker's default values in case connector does not
have any "topic.creation" properties?
As far as I understand this will completely solve the problem described in this
ticket. And also people will be able to set "auto.create.topics.enable = false"
on Kafka cluster without any impact.
Please, correct me if I am wrong.
> Source connectors should report error when trying to produce records to
> non-existent topics instead of hanging forever
> --
>
> Key: KAFKA-10340
> URL: https://issues.apache.org/jira/browse/KAFKA-10340
> Project: Kafka
> Issue Type: Bug
> Components: KafkaConnect
>Reporter: Arjun Satish
>Assignee: Chris Egerton
>Priority: Major
>
> Currently, a source connector will blindly attempt to write a record to a
> Kafka topic. When the topic does not exist, its creation is controlled by the
> {{auto.create.topics.enable}} config on the brokers. When auto.create is
> disabled, the producer.send() call on the Connect worker will hang
> indefinitely (due to the "infinite retries" configuration for said producer).
> In setups where this config is usually disabled, the source connector simply
> appears to hang and not produce any output.
> It is desirable to either log an info or an error message (or inform the user
> somehow) that the connector is simply stuck waiting for the destination topic
> to be created. When the worker has permissions to inspect the broker
> settings, it can use the {{listTopics}} and {{describeConfigs}} API in
> AdminClient to check if the topic exists, the broker can
> {{auto.create.topics.enable}} topics, and if these cases do not exist, either
> throw an error.
> With the recently merged
> [KIP-158|https://cwiki.apache.org/confluence/display/KAFKA/KIP-158%3A+Kafka+Connect+should+allow+source+connectors+to+set+topic-specific+settings+for+new+topics],
> this becomes even more specific a corner case: when topic creation settings
> are enabled, the worker should handle the corner case where topic creation is
> disabled, {{auto.create.topics.enable}} is set to false and topic does not
> exist.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Commented] (KAFKA-10340) Source connectors should report error when trying to produce records to non-existent topics instead of hanging forever
[
https://issues.apache.org/jira/browse/KAFKA-10340?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17282498#comment-17282498
]
Goltseva Taisiia commented on KAFKA-10340:
--
Hi, guys!
According to [KAFKA-5295](https://issues.apache.org/jira/browse/KAFKA-5295) I
supposed that now topics are always created by AdminClient if
"topic.creation.enable = true".
But now I see that topics are created by AdminClient, only when
"topic.creation.enable = true" and "topic.creation" parameters are used.
But I wonder why?
Why do not create topics always by AdminClient if "topic.creation.enable =
true"?
It will partially solve the current problem (described in this ticket) and only
the case with "topic.creation.enable = false" is the problem.
And also current approach can be a problem for us - we wanted to set
"auto.create.topics.enable = false" on Kafka cluster by default. But we cannot
do that, because our connectors can be created with
"topic.creation" parameters or without them, we cannot guarantee that. And in
case connectors have not this property, we get errors.
Any comments and your opinion is appreciated!
> Source connectors should report error when trying to produce records to
> non-existent topics instead of hanging forever
> --
>
> Key: KAFKA-10340
> URL: https://issues.apache.org/jira/browse/KAFKA-10340
> Project: Kafka
> Issue Type: Bug
> Components: KafkaConnect
>Reporter: Arjun Satish
>Assignee: Chris Egerton
>Priority: Major
>
> Currently, a source connector will blindly attempt to write a record to a
> Kafka topic. When the topic does not exist, its creation is controlled by the
> {{auto.create.topics.enable}} config on the brokers. When auto.create is
> disabled, the producer.send() call on the Connect worker will hang
> indefinitely (due to the "infinite retries" configuration for said producer).
> In setups where this config is usually disabled, the source connector simply
> appears to hang and not produce any output.
> It is desirable to either log an info or an error message (or inform the user
> somehow) that the connector is simply stuck waiting for the destination topic
> to be created. When the worker has permissions to inspect the broker
> settings, it can use the {{listTopics}} and {{describeConfigs}} API in
> AdminClient to check if the topic exists, the broker can
> {{auto.create.topics.enable}} topics, and if these cases do not exist, either
> throw an error.
> With the recently merged
> [KIP-158|https://cwiki.apache.org/confluence/display/KAFKA/KIP-158%3A+Kafka+Connect+should+allow+source+connectors+to+set+topic-specific+settings+for+new+topics],
> this becomes even more specific a corner case: when topic creation settings
> are enabled, the worker should handle the corner case where topic creation is
> disabled, {{auto.create.topics.enable}} is set to false and topic does not
> exist.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Created] (KAFKA-12316) Configuration is not defined: topic.creation.default.partitions
Goltseva Taisiia created KAFKA-12316:
Summary: Configuration is not defined:
topic.creation.default.partitions
Key: KAFKA-12316
URL: https://issues.apache.org/jira/browse/KAFKA-12316
Project: Kafka
Issue Type: Bug
Components: KafkaConnect
Affects Versions: 2.6.1
Reporter: Goltseva Taisiia
Hi, guys!
The KIP was implemented:
[https://cwiki.apache.org/confluence/display/KAFKA/KIP-158%3A+Kafka+Connect+should+allow+source+connectors+to+set+topic-specific+settings+for+new+topics]
But it seems you forget to add changes to the class:
[https://github.com/apache/kafka/blob/trunk/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java]
I suppose we need to add something like 'validateClientOverrides()' for configs
starting with 'topic.creation' prefix. Like this:
[https://github.com/apache/kafka/blob/trunk/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java#L420]
For now when I create, for example, a Postgres Source connector and do not
specify login and password (it's mandatory parameters) I get this:
{code:java}
{"error_code": 400,"message": "Connector configuration is invalid and
contains the following 1 error(s):\nConfiguration is not defined:
topic.creation.default.partitions\nConfiguration is not defined:
topic.creation.test1.retention.ms\nConfiguration is not defined:
topic.creation.test1.include\nConfiguration is not defined:
topic.creation.test1.partitions\nConfiguration is not defined:
topic.creation.default.replication.factor\nA value is required\nYou can also
find the above list of errors at the endpoint
`/connector-plugins/{connectorType}/config/validate`"}{code}
But it should be just:
{code:java}
{ "error_code": 400, "message": "Connector configuration is invalid and
contains the following 1 error(s):\nA value is required\nYou can also find the
above list of errors at the endpoint
`/connector-plugins/{connectorType}/config/validate`" }{code}
So, I think, a little change of AbstractHerder class is required.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Comment Edited] (KAFKA-10876) Duplicate connector/task create requests lead to incorrect FAILED status
[
https://issues.apache.org/jira/browse/KAFKA-10876?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17253924#comment-17253924
]
Goltseva Taisiia edited comment on KAFKA-10876 at 12/23/20, 7:15 AM:
-
Hi!
Seems it's the same issue:
https://issues.apache.org/jira/browse/KAFKA-7878
was (Author: xakassi):
Hi!
It could be seems it's the same issue:
https://issues.apache.org/jira/browse/KAFKA-7878
> Duplicate connector/task create requests lead to incorrect FAILED status
>
>
> Key: KAFKA-10876
> URL: https://issues.apache.org/jira/browse/KAFKA-10876
> Project: Kafka
> Issue Type: Bug
> Components: KafkaConnect
>Reporter: Chris Egerton
>Priority: Major
>
> If a Connect worker tries to start a connector or task that it is already
> running, an error will be logged and the connector/task will be marked as
> {{FAILED}}. This logic is implemented in several places:
> *
> [https://github.com/apache/kafka/blob/300909d9e60eb1d5e80f4d744d3662a105ac0c15/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java#L257-L262]
> *
> [https://github.com/apache/kafka/blob/300909d9e60eb1d5e80f4d744d3662a105ac0c15/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java#L299-L306]
> *
> [https://github.com/apache/kafka/blob/300909d9e60eb1d5e80f4d744d3662a105ac0c15/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java#L511-L512]
> *
> [https://github.com/apache/kafka/blob/300909d9e60eb1d5e80f4d744d3662a105ac0c15/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java#L570-L572]
>
> Although it's certainly abnormal for a worker to run into this case and an
> {{ERROR}}-level log message is warranted when it occurs, the connector/task
> should not be marked as {{FAILED}}, as there is still an instance of that
> connector/task still running on the worker.
>
> Either the worker logic should be updated to avoid marking connectors/tasks
> as {{FAILED}} in this case, or it should manually halt the existing
> connector/task before creating a new instance in its place. The first option
> is easier and more intuitive, but if it's ever possible that the
> already-running connector/task instance has an outdated configuration and the
> to-be-created connector/task has an up-to-date configuration, the second
> option would have correct behavior (while the first would not).
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Commented] (KAFKA-10876) Duplicate connector/task create requests lead to incorrect FAILED status
[
https://issues.apache.org/jira/browse/KAFKA-10876?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17253924#comment-17253924
]
Goltseva Taisiia commented on KAFKA-10876:
--
Hi!
It could be seems it's the same issue:
https://issues.apache.org/jira/browse/KAFKA-7878
> Duplicate connector/task create requests lead to incorrect FAILED status
>
>
> Key: KAFKA-10876
> URL: https://issues.apache.org/jira/browse/KAFKA-10876
> Project: Kafka
> Issue Type: Bug
> Components: KafkaConnect
>Reporter: Chris Egerton
>Priority: Major
>
> If a Connect worker tries to start a connector or task that it is already
> running, an error will be logged and the connector/task will be marked as
> {{FAILED}}. This logic is implemented in several places:
> *
> [https://github.com/apache/kafka/blob/300909d9e60eb1d5e80f4d744d3662a105ac0c15/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java#L257-L262]
> *
> [https://github.com/apache/kafka/blob/300909d9e60eb1d5e80f4d744d3662a105ac0c15/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java#L299-L306]
> *
> [https://github.com/apache/kafka/blob/300909d9e60eb1d5e80f4d744d3662a105ac0c15/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java#L511-L512]
> *
> [https://github.com/apache/kafka/blob/300909d9e60eb1d5e80f4d744d3662a105ac0c15/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java#L570-L572]
>
> Although it's certainly abnormal for a worker to run into this case and an
> {{ERROR}}-level log message is warranted when it occurs, the connector/task
> should not be marked as {{FAILED}}, as there is still an instance of that
> connector/task still running on the worker.
>
> Either the worker logic should be updated to avoid marking connectors/tasks
> as {{FAILED}} in this case, or it should manually halt the existing
> connector/task before creating a new instance in its place. The first option
> is easier and more intuitive, but if it's ever possible that the
> already-running connector/task instance has an outdated configuration and the
> to-be-created connector/task has an up-to-date configuration, the second
> option would have correct behavior (while the first would not).
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Comment Edited] (KAFKA-7421) Deadlock in Kafka Connect
[
https://issues.apache.org/jira/browse/KAFKA-7421?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17214685#comment-17214685
]
Goltseva Taisiia edited comment on KAFKA-7421 at 10/16/20, 6:25 AM:
Hi, [~kleiby] !
Let me describe my issue more detailed. Maybe we will be able to figure out
some similar details with you which causing the problem.
We have a base docker image containing several connectors (each connector in a
separate folder under /plugins folder). And one custom converter in /libs. And
this docker image works perfect, never such deadlocks.
Also we have a child docker image which adds additional connector in a separate
folder under /plugins folder. And several additional custom converters in
single JAR in the same folder! We cannot put them in /libs folder currently,
but I will try to do it. And this docker image get this deadlock very often.
Several custom converters in the same folder with a connector is the only
difference with the approach of our base docker image. So, I suppose maybe this
cause our deadlocks. But I'm not sure, I will try to separate converters and
see.
Can you describe your configuration also& Maybe there are some similarities
with us? Do you have SMTs or converters in one folder, in one jar?
was (Author: xakassi):
Hi, [~kleiby] !
Let me describe my issue more detailed. Maybe we will be able to figure out
some similar details with you which causing the problem.
We have a base docker image containing several connectors (each connector in a
separate folder under /plugins folder). And one custom converter in /libs. And
this docker image works perfect, never such deadlocks.
Also we have a child docker image which adds additional connector in a separate
folder under /plugins folder. And several additional custom converters in the
same folder! We cannot put them in /libs folder currently, but I will try to do
it. And this docker image get this deadlock very often.
Several custom converters in the same folder with a connector is the only
difference with the approach of our base docker image. So, I suppose maybe this
cause our deadlocks. But I'm not sure, I will try to separate converters and
see.
Can you describe your configuration also& Maybe there are some similarities
with us? Do you have SMTs or converters in one folder, in one jar?
> Deadlock in Kafka Connect
> -
>
> Key: KAFKA-7421
> URL: https://issues.apache.org/jira/browse/KAFKA-7421
> Project: Kafka
> Issue Type: Bug
> Components: KafkaConnect
>Affects Versions: 2.0.0
>Reporter: Maciej Bryński
>Assignee: Konstantine Karantasis
>Priority: Major
>
> I'm getting this deadlock on half of Kafka Connect runs when having two
> different types connectors (in this configuration it's debezium and hdfs).
> Thread 1:
> {code}
> "pool-22-thread-2@4748" prio=5 tid=0x4d nid=NA waiting for monitor entry
> java.lang.Thread.State: BLOCKED
>waiting for pool-22-thread-1@4747 to release lock on <0x1423> (a
> org.apache.kafka.connect.runtime.isolation.PluginClassLoader)
> at
> org.apache.kafka.connect.runtime.isolation.PluginClassLoader.loadClass(PluginClassLoader.java:91)
> at
> org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader.loadClass(DelegatingClassLoader.java:367)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
> at java.lang.Class.forName0(Class.java:-1)
> at java.lang.Class.forName(Class.java:348)
> at
> org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:715)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.enrich(ConnectorConfig.java:295)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:200)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:194)
> at
> org.apache.kafka.connect.runtime.Worker.startConnector(Worker.java:233)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.startConnector(DistributedHerder.java:916)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.access$1300(DistributedHerder.java:111)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder$15.call(DistributedHerder.java:932)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder$15.call(DistributedHerder.java:928)
> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
> at java.lang.Thread.run(Thread.java:748)
> {code}
> Thread 2:
> {code}
> "pool-22-thread-1@4747" prio=5 ti
[jira] [Comment Edited] (KAFKA-7421) Deadlock in Kafka Connect
[
https://issues.apache.org/jira/browse/KAFKA-7421?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17214685#comment-17214685
]
Goltseva Taisiia edited comment on KAFKA-7421 at 10/15/20, 1:30 PM:
Hi, [~kleiby] !
Let me describe my issue more detailed. Maybe we will be able to figure out
some similar details with you which causing the problem.
We have a base docker image containing several connectors (each connector in a
separate folder under /plugins folder). And one custom converter in /libs. And
this docker image works perfect, never such deadlocks.
Also we have a child docker image which adds additional connector in a separate
folder under /plugins folder. And several additional custom converters in the
same folder! We cannot put them in /libs folder currently, but I will try to do
it. And this docker image get this deadlock very often.
Several custom converters in the same folder with a connector is the only
difference with the approach of our base docker image. So, I suppose maybe this
cause our deadlocks. But I'm not sure, I will try to separate converters and
see.
Can you describe your configuration also& Maybe there are some similarities
with us? Do you have SMTs or converters in one folder, in one jar?
was (Author: xakassi):
Hi, [~kleiby] !
Let me describe my issue more detailed. Maybe we will be able to figure out
some similar details with you which causing the problem.
We have a base docker image containing several connector (each connector in a
separate folder under /plugins folder). And one custom converter in /libs. And
this docker image works perfect, never such deadlocks.
Also we have a child docker image which adds additional connector in a separate
folder under /plugins folder. And several additional custom converters in the
same folder! We cannot put them in /libs folder currently, but I will try to do
it. And this docker image get this deadlock very often.
Several custom converters in the same folder with a connector is the only
difference with the approach of our base docker image. So, I suppose maybe this
cause our deadlocks. But I'm not sure, I will try to separate converters and
see.
Can you describe your configuration also& Maybe there are some similarities
with us? Do you have SMTs or converters in one folder, in one jar?
> Deadlock in Kafka Connect
> -
>
> Key: KAFKA-7421
> URL: https://issues.apache.org/jira/browse/KAFKA-7421
> Project: Kafka
> Issue Type: Bug
> Components: KafkaConnect
>Affects Versions: 2.0.0
>Reporter: Maciej Bryński
>Assignee: Konstantine Karantasis
>Priority: Major
>
> I'm getting this deadlock on half of Kafka Connect runs when having two
> different types connectors (in this configuration it's debezium and hdfs).
> Thread 1:
> {code}
> "pool-22-thread-2@4748" prio=5 tid=0x4d nid=NA waiting for monitor entry
> java.lang.Thread.State: BLOCKED
>waiting for pool-22-thread-1@4747 to release lock on <0x1423> (a
> org.apache.kafka.connect.runtime.isolation.PluginClassLoader)
> at
> org.apache.kafka.connect.runtime.isolation.PluginClassLoader.loadClass(PluginClassLoader.java:91)
> at
> org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader.loadClass(DelegatingClassLoader.java:367)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
> at java.lang.Class.forName0(Class.java:-1)
> at java.lang.Class.forName(Class.java:348)
> at
> org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:715)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.enrich(ConnectorConfig.java:295)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:200)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:194)
> at
> org.apache.kafka.connect.runtime.Worker.startConnector(Worker.java:233)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.startConnector(DistributedHerder.java:916)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.access$1300(DistributedHerder.java:111)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder$15.call(DistributedHerder.java:932)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder$15.call(DistributedHerder.java:928)
> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
> at java.lang.Thread.run(Thread.java:748)
> {code}
> Thread 2:
> {code}
> "pool-22-thread-1@4747" prio=5 tid=0x4c nid=NA w
[jira] [Commented] (KAFKA-7421) Deadlock in Kafka Connect
[
https://issues.apache.org/jira/browse/KAFKA-7421?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17214685#comment-17214685
]
Goltseva Taisiia commented on KAFKA-7421:
-
Hi, [~kleiby] !
Let me describe my issue more detailed. Maybe we will be able to figure out
some similar details with you which causing the problem.
We have a base docker image containing several connector (each connector in a
separate folder under /plugins folder). And one custom converter in /libs. And
this docker image works perfect, never such deadlocks.
Also we have a child docker image which adds additional connector in a separate
folder under /plugins folder. And several additional custom converters in the
same folder! We cannot put them in /libs folder currently, but I will try to do
it. And this docker image get this deadlock very often.
Several custom converters in the same folder with a connector is the only
difference with the approach of our base docker image. So, I suppose maybe this
cause our deadlocks. But I'm not sure, I will try to separate converters and
see.
Can you describe your configuration also& Maybe there are some similarities
with us? Do you have SMTs or converters in one folder, in one jar?
> Deadlock in Kafka Connect
> -
>
> Key: KAFKA-7421
> URL: https://issues.apache.org/jira/browse/KAFKA-7421
> Project: Kafka
> Issue Type: Bug
> Components: KafkaConnect
>Affects Versions: 2.0.0
>Reporter: Maciej Bryński
>Assignee: Konstantine Karantasis
>Priority: Major
>
> I'm getting this deadlock on half of Kafka Connect runs when having two
> different types connectors (in this configuration it's debezium and hdfs).
> Thread 1:
> {code}
> "pool-22-thread-2@4748" prio=5 tid=0x4d nid=NA waiting for monitor entry
> java.lang.Thread.State: BLOCKED
>waiting for pool-22-thread-1@4747 to release lock on <0x1423> (a
> org.apache.kafka.connect.runtime.isolation.PluginClassLoader)
> at
> org.apache.kafka.connect.runtime.isolation.PluginClassLoader.loadClass(PluginClassLoader.java:91)
> at
> org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader.loadClass(DelegatingClassLoader.java:367)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
> at java.lang.Class.forName0(Class.java:-1)
> at java.lang.Class.forName(Class.java:348)
> at
> org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:715)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.enrich(ConnectorConfig.java:295)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:200)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:194)
> at
> org.apache.kafka.connect.runtime.Worker.startConnector(Worker.java:233)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.startConnector(DistributedHerder.java:916)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.access$1300(DistributedHerder.java:111)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder$15.call(DistributedHerder.java:932)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder$15.call(DistributedHerder.java:928)
> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
> at java.lang.Thread.run(Thread.java:748)
> {code}
> Thread 2:
> {code}
> "pool-22-thread-1@4747" prio=5 tid=0x4c nid=NA waiting for monitor entry
> java.lang.Thread.State: BLOCKED
>blocks pool-22-thread-2@4748
>waiting for pool-22-thread-2@4748 to release lock on <0x1421> (a
> org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:406)
> at
> org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader.loadClass(DelegatingClassLoader.java:358)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:411)
> - locked <0x1424> (a java.lang.Object)
> at
> org.apache.kafka.connect.runtime.isolation.PluginClassLoader.loadClass(PluginClassLoader.java:104)
> - locked <0x1423> (a
> org.apache.kafka.connect.runtime.isolation.PluginClassLoader)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
> at
> io.debezium.transforms.ByLogicalTableRouter.(ByLogicalTableRouter.java:57)
> at java.lang.Class.forName0(Class.java:-1)
> at java.lang.Class.forName(Class.java:348)
> at
> org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:715)
> at
> org.a
[jira] [Commented] (KAFKA-10253) Kafka Connect gets into an infinite rebalance loop
[
https://issues.apache.org/jira/browse/KAFKA-10253?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17192798#comment-17192798
]
Goltseva Taisiia commented on KAFKA-10253:
--
Hi, guys!
We had the same problem. In our case the problem was with *group.id* parameter.
It was not unique across Kafka Connect clusters. We found a cluster with the
same *group.id*, changed it and everything returned to normal.
> Kafka Connect gets into an infinite rebalance loop
> --
>
> Key: KAFKA-10253
> URL: https://issues.apache.org/jira/browse/KAFKA-10253
> Project: Kafka
> Issue Type: Bug
> Components: KafkaConnect
>Affects Versions: 2.5.0
>Reporter: Konstantin Lalafaryan
>Priority: Blocker
>
> Hello everyone!
>
> We are running kafka-connect cluster (3 workers) and very often it gets into
> an infinite rebalance loop.
>
> {code:java}
> 2020-07-09 08:51:25,731 INFO [Worker clientId=connect-1, groupId=
> kafka-connect] Rebalance started
> (org.apache.kafka.connect.runtime.distributed.WorkerCoordinator)
> [DistributedHerder-connect-1-1]
> 2020-07-09 08:51:25,731 INFO [Worker clientId=connect-1, groupId=
> kafka-connect] (Re-)joining group
> (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
> [DistributedHerder-connect-1-1]
> 2020-07-09 08:51:25,733 INFO [Worker clientId=connect-1, groupId=
> kafka-connect] Was selected to perform assignments, but do not have latest
> config found in sync request. Returning an empty configuration to trigger
> re-sync.
> (org.apache.kafka.connect.runtime.distributed.IncrementalCooperativeAssignor)
> [DistributedHerder-connect-1-1]
> 2020-07-09 08:51:25,735 INFO [Worker clientId=connect-1, groupId=
> kafka-connect] Successfully joined group with generation 305655831
> (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
> [DistributedHerder-connect-1-1]
> 2020-07-09 08:51:25,735 INFO [Worker clientId=connect-1, groupId=
> kafka-connect] Joined group at generation 305655831 with protocol version 2
> and got assignment: Assignment{error=1,
> leader='connect-1-0008abc5-a152-42fe-a697-a4a4641f72bb',
> leaderUrl='http://10.20.30.221:8083/', offset=12, connectorIds=[],
> taskIds=[], revokedConnectorIds=[], revokedTaskIds=[], delay=0} with
> rebalance delay: 0
> (org.apache.kafka.connect.runtime.distributed.DistributedHerder)
> [DistributedHerder-connect-1-1]
> 2020-07-09 08:51:25,735 INFO [Worker clientId=connect-1, groupId=
> kafka-connect] Rebalance started
> (org.apache.kafka.connect.runtime.distributed.WorkerCoordinator)
> [DistributedHerder-connect-1-1]
> 2020-07-09 08:51:25,735 INFO [Worker clientId=connect-1, groupId=
> kafka-connect] (Re-)joining group
> (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
> [DistributedHerder-connect-1-1]
> 2020-07-09 08:51:25,736 INFO [Worker clientId=connect-1, groupId=
> kafka-connect] Was selected to perform assignments, but do not have latest
> config found in sync request. Returning an empty configuration to trigger
> re-sync.
> (org.apache.kafka.connect.runtime.distributed.IncrementalCooperativeAssignor)
> [DistributedHerder-connect-1-1]
> 2020-07-09 08:51:25,739 INFO [Worker clientId=connect-1, groupId=
> kafka-connect] Successfully joined group with generation 305655832
> (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
> [DistributedHerder-connect-1-1]
> 2020-07-09 08:51:25,739 INFO [Worker clientId=connect-1, groupId=
> kafka-connect] Joined group at generation 305655832 with protocol version 2
> and got assignment: Assignment{error=1,
> leader='connect-1-0008abc5-a152-42fe-a697-a4a4641f72bb',
> leaderUrl='http://10.20.30.221:8083/', offset=12, connectorIds=[],
> taskIds=[], revokedConnectorIds=[], revokedTaskIds=[], delay=0} with
> rebalance delay: 0
> (org.apache.kafka.connect.runtime.distributed.DistributedHerder)
> [DistributedHerder-connect-1-1]
> 2020-07-09 08:51:25,739 INFO [Worker clientId=connect-1, groupId=
> kafka-connect] Rebalance started
> (org.apache.kafka.connect.runtime.distributed.WorkerCoordinator)
> [DistributedHerder-connect-1-1]
> 2020-07-09 08:51:25,739 INFO [Worker clientId=connect-1, groupId=
> kafka-connect] (Re-)joining group
> (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
> [DistributedHerder-connect-1-1]
> 2020-07-09 08:51:25,740 INFO [Worker clientId=connect-1, groupId=
> kafka-connect] Was selected to perform assignments, but do not have latest
> config found in sync request. Returning an empty configuration to trigger
> re-sync.
> (org.apache.kafka.connect.runtime.distributed.IncrementalCooperativeAssignor)
> [DistributedHerder-connect-1-1]
> 2020-07-09 08:51:25,742 INFO [Worker clientId=connect-1, groupId=
> kafka-connect] Successfully joined group with
[jira] [Comment Edited] (KAFKA-7421) Deadlock in Kafka Connect
[
https://issues.apache.org/jira/browse/KAFKA-7421?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17185891#comment-17185891
]
Goltseva Taisiia edited comment on KAFKA-7421 at 8/28/20, 8:16 AM:
---
And I agree with [~ivanyu], I do not understand why *{{PluginClassLoader}}'s
loadClass* method is marked as synchronized, because it has fine-grain
synchronization immediately inside. And as I can see, this method do not use
any fields of *{{PluginClassLoader}}* class.
[~kkonstantine] what do you think about it? Can we propose a PR to remove
synchronized from *{{PluginClassLoader}}'s loadClass* method?
was (Author: xakassi):
And I agree with [~ivanyu], I do not understand why *{{PluginClassLoader}}'s
loadClass* method is marked as synchronized, because it has fine-grain
synchronization immediately inside.
[~kkonstantine] what do you think about it?
> Deadlock in Kafka Connect
> -
>
> Key: KAFKA-7421
> URL: https://issues.apache.org/jira/browse/KAFKA-7421
> Project: Kafka
> Issue Type: Bug
> Components: KafkaConnect
>Affects Versions: 2.0.0
>Reporter: Maciej Bryński
>Assignee: Konstantine Karantasis
>Priority: Major
>
> I'm getting this deadlock on half of Kafka Connect runs when having two
> different types connectors (in this configuration it's debezium and hdfs).
> Thread 1:
> {code}
> "pool-22-thread-2@4748" prio=5 tid=0x4d nid=NA waiting for monitor entry
> java.lang.Thread.State: BLOCKED
>waiting for pool-22-thread-1@4747 to release lock on <0x1423> (a
> org.apache.kafka.connect.runtime.isolation.PluginClassLoader)
> at
> org.apache.kafka.connect.runtime.isolation.PluginClassLoader.loadClass(PluginClassLoader.java:91)
> at
> org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader.loadClass(DelegatingClassLoader.java:367)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
> at java.lang.Class.forName0(Class.java:-1)
> at java.lang.Class.forName(Class.java:348)
> at
> org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:715)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.enrich(ConnectorConfig.java:295)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:200)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:194)
> at
> org.apache.kafka.connect.runtime.Worker.startConnector(Worker.java:233)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.startConnector(DistributedHerder.java:916)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.access$1300(DistributedHerder.java:111)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder$15.call(DistributedHerder.java:932)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder$15.call(DistributedHerder.java:928)
> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
> at java.lang.Thread.run(Thread.java:748)
> {code}
> Thread 2:
> {code}
> "pool-22-thread-1@4747" prio=5 tid=0x4c nid=NA waiting for monitor entry
> java.lang.Thread.State: BLOCKED
>blocks pool-22-thread-2@4748
>waiting for pool-22-thread-2@4748 to release lock on <0x1421> (a
> org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:406)
> at
> org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader.loadClass(DelegatingClassLoader.java:358)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:411)
> - locked <0x1424> (a java.lang.Object)
> at
> org.apache.kafka.connect.runtime.isolation.PluginClassLoader.loadClass(PluginClassLoader.java:104)
> - locked <0x1423> (a
> org.apache.kafka.connect.runtime.isolation.PluginClassLoader)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
> at
> io.debezium.transforms.ByLogicalTableRouter.(ByLogicalTableRouter.java:57)
> at java.lang.Class.forName0(Class.java:-1)
> at java.lang.Class.forName(Class.java:348)
> at
> org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:715)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.enrich(ConnectorConfig.java:295)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:200)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:194)
> at
> org.apache.kafka.connect.runtime.Worker.startConnector(
[jira] [Comment Edited] (KAFKA-7421) Deadlock in Kafka Connect
[
https://issues.apache.org/jira/browse/KAFKA-7421?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17185890#comment-17185890
]
Goltseva Taisiia edited comment on KAFKA-7421 at 8/27/20, 2:43 PM:
---
Hi, guys!
We faced the similar deadlock (not equal but similar):
{code:java}
Found one Java-level deadlock:
=
"pool-9-thread-6":
waiting to lock monitor 0x7f112e5565a8 (object 0x84400d80, a
org.apache.kafka.connect.runtime.isolation.PluginClassLoader),
which is held by "pool-9-thread-5"
"pool-9-thread-5":
waiting to lock monitor 0x7f112e528c48 (object 0x8441a4c8, a
org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader),
which is held by "pool-9-thread-2"
"pool-9-thread-2":
waiting to lock monitor 0x7f112e5565a8 (object 0x84400d80, a
org.apache.kafka.connect.runtime.isolation.PluginClassLoader),
which is held by "pool-9-thread-5"{code}
Thread-1:
{code:java}
"pool-9-thread-6": at
org.apache.kafka.connect.runtime.isolation.PluginClassLoader.loadClass(PluginClassLoader.java:91)
- waiting to lock <0x84400d80> (a
org.apache.kafka.connect.runtime.isolation.PluginClassLoader) at
java.lang.ClassLoader.loadClass(ClassLoader.java:352) at
org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:716) at
org.apache.kafka.common.config.ConfigDef.parseValue(ConfigDef.java:474) at
org.apache.kafka.common.config.ConfigDef.parse(ConfigDef.java:467) at
org.apache.kafka.common.config.AbstractConfig.(AbstractConfig.java:108)
at
org.apache.kafka.common.config.AbstractConfig.(AbstractConfig.java:129)
at
org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:210)
at
org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:206)
at org.apache.kafka.connect.runtime.Worker.startTask(Worker.java:430) at
com.company.streaming.platform.kafka.DistributedHerder.startTask(DistributedHerder.java:1216)
at
com.company.streaming.platform.kafka.DistributedHerder.access$1700(DistributedHerder.java:97)
at
com.company.streaming.platform.kafka.DistributedHerder$13.call(DistributedHerder.java:1231)
at
com.company.streaming.platform.kafka.DistributedHerder$13.call(DistributedHerder.java:1227)
at java.util.concurrent.FutureTask.run(FutureTask.java:266) at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748){code}
Thread-2:
{code:java}
"pool-9-thread-5": at java.lang.ClassLoader.loadClass(ClassLoader.java:399) -
waiting to lock <0x8441a4c8> (a
org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) at
org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader.loadClass(DelegatingClassLoader.java:392)
at java.lang.ClassLoader.loadClass(ClassLoader.java:406) - locked
<0x80a80780> (a java.lang.Object) at
org.apache.kafka.connect.runtime.isolation.PluginClassLoader.loadClass(PluginClassLoader.java:104)
- locked <0x80a80780> (a java.lang.Object) - locked
<0x84400d80> (a
org.apache.kafka.connect.runtime.isolation.PluginClassLoader) at
java.lang.ClassLoader.loadClass(ClassLoader.java:352) at
org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:716) at
org.apache.kafka.common.config.ConfigDef.parseValue(ConfigDef.java:474) at
org.apache.kafka.common.config.ConfigDef.parse(ConfigDef.java:467) at
org.apache.kafka.common.config.AbstractConfig.(AbstractConfig.java:108)
at
org.apache.kafka.common.config.AbstractConfig.(AbstractConfig.java:129)
at
org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:210)
at
org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:206)
at org.apache.kafka.connect.runtime.Worker.startTask(Worker.java:430) at
com.company.streaming.platform.kafka.DistributedHerder.startTask(DistributedHerder.java:1216)
at
com.company.streaming.platform.kafka.DistributedHerder.access$1700(DistributedHerder.java:97)
at
com.company.streaming.platform.kafka.DistributedHerder$13.call(DistributedHerder.java:1231)
at
com.company.streaming.platform.kafka.DistributedHerder$13.call(DistributedHerder.java:1227)
at java.util.concurrent.FutureTask.run(FutureTask.java:266) at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748){code}
Thread-3:
{code:java}
"pool-9-thread-2": at
org.apache.kafka.connect.runtime.isolation.PluginClassLoader.loadClass(PluginClassLoader.java:91)
- waiting to lock <0x84400d80> (a
org.apache.kafka.connect.runtime.isolation.PluginClassLoader) at
org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader.loadClass(DelegatingClassLoader.jav
[jira] [Comment Edited] (KAFKA-7421) Deadlock in Kafka Connect
[
https://issues.apache.org/jira/browse/KAFKA-7421?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17185890#comment-17185890
]
Goltseva Taisiia edited comment on KAFKA-7421 at 8/27/20, 2:42 PM:
---
Hi, guys!
We faced the similar deadlock (not equal but similar):
{code:java}
Found one Java-level deadlock:
=
"pool-9-thread-6":
waiting to lock monitor 0x7f112e5565a8 (object 0x84400d80, a
org.apache.kafka.connect.runtime.isolation.PluginClassLoader),
which is held by "pool-9-thread-5"
"pool-9-thread-5":
waiting to lock monitor 0x7f112e528c48 (object 0x8441a4c8, a
org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader),
which is held by "pool-9-thread-2"
"pool-9-thread-2":
waiting to lock monitor 0x7f112e5565a8 (object 0x84400d80, a
org.apache.kafka.connect.runtime.isolation.PluginClassLoader),
which is held by "pool-9-thread-5"{code}
Thread-1:
{code:java}
"pool-9-thread-6":"pool-9-thread-6": at
org.apache.kafka.connect.runtime.isolation.PluginClassLoader.loadClass(PluginClassLoader.java:91)
- waiting to lock <0x84400d80> (a
org.apache.kafka.connect.runtime.isolation.PluginClassLoader) at
java.lang.ClassLoader.loadClass(ClassLoader.java:352) at
org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:716) at
org.apache.kafka.common.config.ConfigDef.parseValue(ConfigDef.java:474) at
org.apache.kafka.common.config.ConfigDef.parse(ConfigDef.java:467) at
org.apache.kafka.common.config.AbstractConfig.(AbstractConfig.java:108)
at
org.apache.kafka.common.config.AbstractConfig.(AbstractConfig.java:129)
at
org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:210)
at
org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:206)
at org.apache.kafka.connect.runtime.Worker.startTask(Worker.java:430) at
com.company.streaming.platform.kafka.DistributedHerder.startTask(DistributedHerder.java:1216)
at
com.company.streaming.platform.kafka.DistributedHerder.access$1700(DistributedHerder.java:97)
at
com.company.streaming.platform.kafka.DistributedHerder$13.call(DistributedHerder.java:1231)
at
com.company.streaming.platform.kafka.DistributedHerder$13.call(DistributedHerder.java:1227)
at java.util.concurrent.FutureTask.run(FutureTask.java:266) at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748){code}
Thread-2:
{code:java}
"pool-9-thread-5":"pool-9-thread-5": at
java.lang.ClassLoader.loadClass(ClassLoader.java:399) - waiting to lock
<0x8441a4c8> (a
org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) at
org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader.loadClass(DelegatingClassLoader.java:392)
at java.lang.ClassLoader.loadClass(ClassLoader.java:406) - locked
<0x80a80780> (a java.lang.Object) at
org.apache.kafka.connect.runtime.isolation.PluginClassLoader.loadClass(PluginClassLoader.java:104)
- locked <0x80a80780> (a java.lang.Object) - locked
<0x84400d80> (a
org.apache.kafka.connect.runtime.isolation.PluginClassLoader) at
java.lang.ClassLoader.loadClass(ClassLoader.java:352) at
org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:716) at
org.apache.kafka.common.config.ConfigDef.parseValue(ConfigDef.java:474) at
org.apache.kafka.common.config.ConfigDef.parse(ConfigDef.java:467) at
org.apache.kafka.common.config.AbstractConfig.(AbstractConfig.java:108)
at
org.apache.kafka.common.config.AbstractConfig.(AbstractConfig.java:129)
at
org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:210)
at
org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:206)
at org.apache.kafka.connect.runtime.Worker.startTask(Worker.java:430) at
com.company.streaming.platform.kafka.DistributedHerder.startTask(DistributedHerder.java:1216)
at
com.company.streaming.platform.kafka.DistributedHerder.access$1700(DistributedHerder.java:97)
at
com.company.streaming.platform.kafka.DistributedHerder$13.call(DistributedHerder.java:1231)
at
com.company.streaming.platform.kafka.DistributedHerder$13.call(DistributedHerder.java:1227)
at java.util.concurrent.FutureTask.run(FutureTask.java:266) at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748){code}
Thread-3:
{code:java}
"pool-9-thread-2":"pool-9-thread-2": at
org.apache.kafka.connect.runtime.isolation.PluginClassLoader.loadClass(PluginClassLoader.java:91)
- waiting to lock <0x84400d80> (a
org.apache.kafka.connect.runtime.isolation.PluginClassLoader) at
org.apache.kafka.connect.runtime.isolation
[jira] [Commented] (KAFKA-7421) Deadlock in Kafka Connect
[
https://issues.apache.org/jira/browse/KAFKA-7421?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17185891#comment-17185891
]
Goltseva Taisiia commented on KAFKA-7421:
-
And I agree with [~ivanyu], I do not understand why {{PluginClassLoader}}'s
{{loadClass }}method is marked as synchronized, because it has fine-grain
synchronization immediately inside.
[~kkonstantine] what do you think about it?
> Deadlock in Kafka Connect
> -
>
> Key: KAFKA-7421
> URL: https://issues.apache.org/jira/browse/KAFKA-7421
> Project: Kafka
> Issue Type: Bug
> Components: KafkaConnect
>Affects Versions: 2.0.0
>Reporter: Maciej Bryński
>Assignee: Konstantine Karantasis
>Priority: Major
>
> I'm getting this deadlock on half of Kafka Connect runs when having two
> different types connectors (in this configuration it's debezium and hdfs).
> Thread 1:
> {code}
> "pool-22-thread-2@4748" prio=5 tid=0x4d nid=NA waiting for monitor entry
> java.lang.Thread.State: BLOCKED
>waiting for pool-22-thread-1@4747 to release lock on <0x1423> (a
> org.apache.kafka.connect.runtime.isolation.PluginClassLoader)
> at
> org.apache.kafka.connect.runtime.isolation.PluginClassLoader.loadClass(PluginClassLoader.java:91)
> at
> org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader.loadClass(DelegatingClassLoader.java:367)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
> at java.lang.Class.forName0(Class.java:-1)
> at java.lang.Class.forName(Class.java:348)
> at
> org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:715)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.enrich(ConnectorConfig.java:295)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:200)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:194)
> at
> org.apache.kafka.connect.runtime.Worker.startConnector(Worker.java:233)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.startConnector(DistributedHerder.java:916)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.access$1300(DistributedHerder.java:111)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder$15.call(DistributedHerder.java:932)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder$15.call(DistributedHerder.java:928)
> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
> at java.lang.Thread.run(Thread.java:748)
> {code}
> Thread 2:
> {code}
> "pool-22-thread-1@4747" prio=5 tid=0x4c nid=NA waiting for monitor entry
> java.lang.Thread.State: BLOCKED
>blocks pool-22-thread-2@4748
>waiting for pool-22-thread-2@4748 to release lock on <0x1421> (a
> org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:406)
> at
> org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader.loadClass(DelegatingClassLoader.java:358)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:411)
> - locked <0x1424> (a java.lang.Object)
> at
> org.apache.kafka.connect.runtime.isolation.PluginClassLoader.loadClass(PluginClassLoader.java:104)
> - locked <0x1423> (a
> org.apache.kafka.connect.runtime.isolation.PluginClassLoader)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
> at
> io.debezium.transforms.ByLogicalTableRouter.(ByLogicalTableRouter.java:57)
> at java.lang.Class.forName0(Class.java:-1)
> at java.lang.Class.forName(Class.java:348)
> at
> org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:715)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.enrich(ConnectorConfig.java:295)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:200)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:194)
> at
> org.apache.kafka.connect.runtime.Worker.startConnector(Worker.java:233)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.startConnector(DistributedHerder.java:916)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.access$1300(DistributedHerder.java:111)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder$15.call(DistributedHerder.java:932)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder$15.call(DistributedHerder.java:928
[jira] [Comment Edited] (KAFKA-7421) Deadlock in Kafka Connect
[
https://issues.apache.org/jira/browse/KAFKA-7421?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17185891#comment-17185891
]
Goltseva Taisiia edited comment on KAFKA-7421 at 8/27/20, 2:38 PM:
---
And I agree with [~ivanyu], I do not understand why *{{PluginClassLoader}}'s
loadClass* method is marked as synchronized, because it has fine-grain
synchronization immediately inside.
[~kkonstantine] what do you think about it?
was (Author: xakassi):
And I agree with [~ivanyu], I do not understand why {{PluginClassLoader}}'s
{{loadClass }}method is marked as synchronized, because it has fine-grain
synchronization immediately inside.
[~kkonstantine] what do you think about it?
> Deadlock in Kafka Connect
> -
>
> Key: KAFKA-7421
> URL: https://issues.apache.org/jira/browse/KAFKA-7421
> Project: Kafka
> Issue Type: Bug
> Components: KafkaConnect
>Affects Versions: 2.0.0
>Reporter: Maciej Bryński
>Assignee: Konstantine Karantasis
>Priority: Major
>
> I'm getting this deadlock on half of Kafka Connect runs when having two
> different types connectors (in this configuration it's debezium and hdfs).
> Thread 1:
> {code}
> "pool-22-thread-2@4748" prio=5 tid=0x4d nid=NA waiting for monitor entry
> java.lang.Thread.State: BLOCKED
>waiting for pool-22-thread-1@4747 to release lock on <0x1423> (a
> org.apache.kafka.connect.runtime.isolation.PluginClassLoader)
> at
> org.apache.kafka.connect.runtime.isolation.PluginClassLoader.loadClass(PluginClassLoader.java:91)
> at
> org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader.loadClass(DelegatingClassLoader.java:367)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
> at java.lang.Class.forName0(Class.java:-1)
> at java.lang.Class.forName(Class.java:348)
> at
> org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:715)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.enrich(ConnectorConfig.java:295)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:200)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:194)
> at
> org.apache.kafka.connect.runtime.Worker.startConnector(Worker.java:233)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.startConnector(DistributedHerder.java:916)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.access$1300(DistributedHerder.java:111)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder$15.call(DistributedHerder.java:932)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder$15.call(DistributedHerder.java:928)
> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
> at java.lang.Thread.run(Thread.java:748)
> {code}
> Thread 2:
> {code}
> "pool-22-thread-1@4747" prio=5 tid=0x4c nid=NA waiting for monitor entry
> java.lang.Thread.State: BLOCKED
>blocks pool-22-thread-2@4748
>waiting for pool-22-thread-2@4748 to release lock on <0x1421> (a
> org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:406)
> at
> org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader.loadClass(DelegatingClassLoader.java:358)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:411)
> - locked <0x1424> (a java.lang.Object)
> at
> org.apache.kafka.connect.runtime.isolation.PluginClassLoader.loadClass(PluginClassLoader.java:104)
> - locked <0x1423> (a
> org.apache.kafka.connect.runtime.isolation.PluginClassLoader)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
> at
> io.debezium.transforms.ByLogicalTableRouter.(ByLogicalTableRouter.java:57)
> at java.lang.Class.forName0(Class.java:-1)
> at java.lang.Class.forName(Class.java:348)
> at
> org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:715)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.enrich(ConnectorConfig.java:295)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:200)
> at
> org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:194)
> at
> org.apache.kafka.connect.runtime.Worker.startConnector(Worker.java:233)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.startConnector(DistributedHerder.java:916)
> at
> org.apache.kafka.connec
[jira] [Commented] (KAFKA-7421) Deadlock in Kafka Connect
[
https://issues.apache.org/jira/browse/KAFKA-7421?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17185890#comment-17185890
]
Goltseva Taisiia commented on KAFKA-7421:
-
Hi, guys!
We faced the similar deadlock (not equal but similar):
{code:java}
Found one Java-level deadlock:Found one Java-level
deadlock:="pool-9-thread-6": waiting to lock
monitor 0x7f112e5565a8 (object 0x84400d80, a
org.apache.kafka.connect.runtime.isolation.PluginClassLoader), which is held
by "pool-9-thread-5""pool-9-thread-5": waiting to lock monitor
0x7f112e528c48 (object 0x8441a4c8, a
org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader), which is
held by "pool-9-thread-2""pool-9-thread-2": waiting to lock monitor
0x7f112e5565a8 (object 0x84400d80, a
org.apache.kafka.connect.runtime.isolation.PluginClassLoader), which is held
by "pool-9-thread-5"
Java stack information for the threads listed
above:==="pool-9-thread-6": at
org.apache.kafka.connect.runtime.isolation.PluginClassLoader.loadClass(PluginClassLoader.java:91)
- waiting to lock <0x84400d80> (a
org.apache.kafka.connect.runtime.isolation.PluginClassLoader) at
java.lang.ClassLoader.loadClass(ClassLoader.java:352) at
org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:716) at
org.apache.kafka.common.config.ConfigDef.parseValue(ConfigDef.java:474) at
org.apache.kafka.common.config.ConfigDef.parse(ConfigDef.java:467) at
org.apache.kafka.common.config.AbstractConfig.(AbstractConfig.java:108)
at
org.apache.kafka.common.config.AbstractConfig.(AbstractConfig.java:129)
at
org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:210)
at
org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:206)
at org.apache.kafka.connect.runtime.Worker.startTask(Worker.java:430) at
com.company.streaming.platform.kafka.DistributedHerder.startTask(DistributedHerder.java:1216)
at
com.company.streaming.platform.kafka.DistributedHerder.access$1700(DistributedHerder.java:97)
at
com.company.streaming.platform.kafka.DistributedHerder$13.call(DistributedHerder.java:1231)
at
com.company.streaming.platform.kafka.DistributedHerder$13.call(DistributedHerder.java:1227)
at java.util.concurrent.FutureTask.run(FutureTask.java:266) at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)"pool-9-thread-5": at
java.lang.ClassLoader.loadClass(ClassLoader.java:399) - waiting to lock
<0x8441a4c8> (a
org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) at
org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader.loadClass(DelegatingClassLoader.java:392)
at java.lang.ClassLoader.loadClass(ClassLoader.java:406) - locked
<0x80a80780> (a java.lang.Object) at
org.apache.kafka.connect.runtime.isolation.PluginClassLoader.loadClass(PluginClassLoader.java:104)
- locked <0x80a80780> (a java.lang.Object) - locked
<0x84400d80> (a
org.apache.kafka.connect.runtime.isolation.PluginClassLoader) at
java.lang.ClassLoader.loadClass(ClassLoader.java:352) at
org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:716) at
org.apache.kafka.common.config.ConfigDef.parseValue(ConfigDef.java:474) at
org.apache.kafka.common.config.ConfigDef.parse(ConfigDef.java:467) at
org.apache.kafka.common.config.AbstractConfig.(AbstractConfig.java:108)
at
org.apache.kafka.common.config.AbstractConfig.(AbstractConfig.java:129)
at
org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:210)
at
org.apache.kafka.connect.runtime.ConnectorConfig.(ConnectorConfig.java:206)
at org.apache.kafka.connect.runtime.Worker.startTask(Worker.java:430) at
com.company.streaming.platform.kafka.DistributedHerder.startTask(DistributedHerder.java:1216)
at
com.company.streaming.platform.kafka.DistributedHerder.access$1700(DistributedHerder.java:97)
at
com.company.streaming.platform.kafka.DistributedHerder$13.call(DistributedHerder.java:1231)
at
com.company.streaming.platform.kafka.DistributedHerder$13.call(DistributedHerder.java:1227)
at java.util.concurrent.FutureTask.run(FutureTask.java:266) at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)"pool-9-thread-2": at
org.apache.kafka.connect.runtime.isolation.PluginClassLoader.loadClass(PluginClassLoader.java:91)
- waiting to lock <0x84400d80> (a
org.apache.kafka.connect.runtime.isolation.PluginClassLoader) at
org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader.loadClass(DelegatingClassLoader.java:3
[jira] [Comment Edited] (KAFKA-10426) Deadlock in KafkaConfigBackingStore
[
https://issues.apache.org/jira/browse/KAFKA-10426?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17183758#comment-17183758
]
Goltseva Taisiia edited comment on KAFKA-10426 at 8/25/20, 6:47 AM:
Hi, [~ChrisEgerton], [~kkonstantine] !
Could you, please, take a look at my PR?
was (Author: xakassi):
Hi, [~ChrisEgerton] !
Could you, please, take a look at my PR?
> Deadlock in KafkaConfigBackingStore
> ---
>
> Key: KAFKA-10426
> URL: https://issues.apache.org/jira/browse/KAFKA-10426
> Project: Kafka
> Issue Type: Bug
> Components: KafkaConnect
>Affects Versions: 2.4.1, 2.6.0
>Reporter: Goltseva Taisiia
>Assignee: Goltseva Taisiia
>Priority: Critical
> Labels: pull-request-available
>
> Hi, guys!
> We faced the following deadlock:
>
> {code:java}
> KafkaBasedLog Work Thread - _streaming_service_config
> priority:5 - threadId:0x7f18ec22c000 - nativeId:0x950 - nativeId
> (decimal):2384 - state:BLOCKED
> stackTrace:
> java.lang.Thread.State: BLOCKED (on object monitor)
> at
> com.company.streaming.platform.kafka.DistributedHerder$ConfigUpdateListener.onSessionKeyUpdate(DistributedHerder.java:1586)
> - waiting to lock <0xe6136808> (a
> com.company.streaming.platform.kafka.DistributedHerder)
> at
> org.apache.kafka.connect.storage.KafkaConfigBackingStore$ConsumeCallback.onCompletion(KafkaConfigBackingStore.java:707)
> - locked <0xd8c3be40> (a java.lang.Object)
> at
> org.apache.kafka.connect.storage.KafkaConfigBackingStore$ConsumeCallback.onCompletion(KafkaConfigBackingStore.java:481)
> at org.apache.kafka.connect.util.KafkaBasedLog.poll(KafkaBasedLog.java:264)
> at
> org.apache.kafka.connect.util.KafkaBasedLog.access$500(KafkaBasedLog.java:71)
> at
> org.apache.kafka.connect.util.KafkaBasedLog$WorkThread.run(KafkaBasedLog.java:337)
> CustomDistributedHerder-connect-1
> priority:5 - threadId:0x7f1a01e30800 - nativeId:0x93a - nativeId
> (decimal):2362 - state:BLOCKED
> stackTrace:
> java.lang.Thread.State: BLOCKED (on object monitor)
> at
> org.apache.kafka.connect.storage.KafkaConfigBackingStore.snapshot(KafkaConfigBackingStore.java:285)
> - waiting to lock <0xd8c3be40> (a java.lang.Object)
> at
> com.company.streaming.platform.kafka.DistributedHerder.updateConfigsWithIncrementalCooperative(DistributedHerder.java:514)
> - locked <0xe6136808> (a
> com.company.streaming.platform.kafka.DistributedHerder)
> at
> com.company.streaming.platform.kafka.DistributedHerder.tick(DistributedHerder.java:402)
> at
> com.company.streaming.platform.kafka.DistributedHerder.run(DistributedHerder.java:286)
> at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
> at java.lang.Thread.run(Thread.java:748){code}
> DistributedHerder went to updateConfigsWithIncrementalCooperative()
> synchronized method and called configBackingStore.snapshot() which take a
> lock on internal object in KafkaConfigBackingStore class.
>
> Meanwhile KafkaConfigBackingStore in ConsumeCallback inside synchronized
> block on internal object got SESSION_KEY record and called
> updateListener.onSessionKeyUpdate() which take a lock on DistributedHerder.
>
> As I can see the problem is here:
> [https://github.com/apache/kafka/blob/trunk/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaConfigBackingStore.java#L737]
>
> As I understand this call should be performed outside synchronized block:
> {code:java}
> if (started)
>
> updateListener.onSessionKeyUpdate(KafkaConfigBackingStore.this.sessionKey);{code}
>
> I'm going to make a PR.
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Commented] (KAFKA-10426) Deadlock in KafkaConfigBackingStore
[
https://issues.apache.org/jira/browse/KAFKA-10426?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17183758#comment-17183758
]
Goltseva Taisiia commented on KAFKA-10426:
--
Hi, [~ChrisEgerton] !
Could you, please, take a look at my PR?
> Deadlock in KafkaConfigBackingStore
> ---
>
> Key: KAFKA-10426
> URL: https://issues.apache.org/jira/browse/KAFKA-10426
> Project: Kafka
> Issue Type: Bug
> Components: KafkaConnect
>Affects Versions: 2.4.1, 2.6.0
>Reporter: Goltseva Taisiia
>Assignee: Goltseva Taisiia
>Priority: Critical
> Labels: pull-request-available
>
> Hi, guys!
> We faced the following deadlock:
>
> {code:java}
> KafkaBasedLog Work Thread - _streaming_service_config
> priority:5 - threadId:0x7f18ec22c000 - nativeId:0x950 - nativeId
> (decimal):2384 - state:BLOCKED
> stackTrace:
> java.lang.Thread.State: BLOCKED (on object monitor)
> at
> com.company.streaming.platform.kafka.DistributedHerder$ConfigUpdateListener.onSessionKeyUpdate(DistributedHerder.java:1586)
> - waiting to lock <0xe6136808> (a
> com.company.streaming.platform.kafka.DistributedHerder)
> at
> org.apache.kafka.connect.storage.KafkaConfigBackingStore$ConsumeCallback.onCompletion(KafkaConfigBackingStore.java:707)
> - locked <0xd8c3be40> (a java.lang.Object)
> at
> org.apache.kafka.connect.storage.KafkaConfigBackingStore$ConsumeCallback.onCompletion(KafkaConfigBackingStore.java:481)
> at org.apache.kafka.connect.util.KafkaBasedLog.poll(KafkaBasedLog.java:264)
> at
> org.apache.kafka.connect.util.KafkaBasedLog.access$500(KafkaBasedLog.java:71)
> at
> org.apache.kafka.connect.util.KafkaBasedLog$WorkThread.run(KafkaBasedLog.java:337)
> CustomDistributedHerder-connect-1
> priority:5 - threadId:0x7f1a01e30800 - nativeId:0x93a - nativeId
> (decimal):2362 - state:BLOCKED
> stackTrace:
> java.lang.Thread.State: BLOCKED (on object monitor)
> at
> org.apache.kafka.connect.storage.KafkaConfigBackingStore.snapshot(KafkaConfigBackingStore.java:285)
> - waiting to lock <0xd8c3be40> (a java.lang.Object)
> at
> com.company.streaming.platform.kafka.DistributedHerder.updateConfigsWithIncrementalCooperative(DistributedHerder.java:514)
> - locked <0xe6136808> (a
> com.company.streaming.platform.kafka.DistributedHerder)
> at
> com.company.streaming.platform.kafka.DistributedHerder.tick(DistributedHerder.java:402)
> at
> com.company.streaming.platform.kafka.DistributedHerder.run(DistributedHerder.java:286)
> at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
> at java.lang.Thread.run(Thread.java:748){code}
> DistributedHerder went to updateConfigsWithIncrementalCooperative()
> synchronized method and called configBackingStore.snapshot() which take a
> lock on internal object in KafkaConfigBackingStore class.
>
> Meanwhile KafkaConfigBackingStore in ConsumeCallback inside synchronized
> block on internal object got SESSION_KEY record and called
> updateListener.onSessionKeyUpdate() which take a lock on DistributedHerder.
>
> As I can see the problem is here:
> [https://github.com/apache/kafka/blob/trunk/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaConfigBackingStore.java#L737]
>
> As I understand this call should be performed outside synchronized block:
> {code:java}
> if (started)
>
> updateListener.onSessionKeyUpdate(KafkaConfigBackingStore.this.sessionKey);{code}
>
> I'm going to make a PR.
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Updated] (KAFKA-10426) Deadlock in KafkaConfigBackingStore
[
https://issues.apache.org/jira/browse/KAFKA-10426?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Goltseva Taisiia updated KAFKA-10426:
-
Reviewer: Konstantine Karantasis (was: Chris Egerton)
> Deadlock in KafkaConfigBackingStore
> ---
>
> Key: KAFKA-10426
> URL: https://issues.apache.org/jira/browse/KAFKA-10426
> Project: Kafka
> Issue Type: Bug
> Components: KafkaConnect
>Affects Versions: 2.4.1, 2.6.0
>Reporter: Goltseva Taisiia
>Assignee: Goltseva Taisiia
>Priority: Critical
> Labels: pull-request-available
>
> Hi, guys!
> We faced the following deadlock:
>
> {code:java}
> KafkaBasedLog Work Thread - _streaming_service_config
> priority:5 - threadId:0x7f18ec22c000 - nativeId:0x950 - nativeId
> (decimal):2384 - state:BLOCKED
> stackTrace:
> java.lang.Thread.State: BLOCKED (on object monitor)
> at
> com.company.streaming.platform.kafka.DistributedHerder$ConfigUpdateListener.onSessionKeyUpdate(DistributedHerder.java:1586)
> - waiting to lock <0xe6136808> (a
> com.company.streaming.platform.kafka.DistributedHerder)
> at
> org.apache.kafka.connect.storage.KafkaConfigBackingStore$ConsumeCallback.onCompletion(KafkaConfigBackingStore.java:707)
> - locked <0xd8c3be40> (a java.lang.Object)
> at
> org.apache.kafka.connect.storage.KafkaConfigBackingStore$ConsumeCallback.onCompletion(KafkaConfigBackingStore.java:481)
> at org.apache.kafka.connect.util.KafkaBasedLog.poll(KafkaBasedLog.java:264)
> at
> org.apache.kafka.connect.util.KafkaBasedLog.access$500(KafkaBasedLog.java:71)
> at
> org.apache.kafka.connect.util.KafkaBasedLog$WorkThread.run(KafkaBasedLog.java:337)
> CustomDistributedHerder-connect-1
> priority:5 - threadId:0x7f1a01e30800 - nativeId:0x93a - nativeId
> (decimal):2362 - state:BLOCKED
> stackTrace:
> java.lang.Thread.State: BLOCKED (on object monitor)
> at
> org.apache.kafka.connect.storage.KafkaConfigBackingStore.snapshot(KafkaConfigBackingStore.java:285)
> - waiting to lock <0xd8c3be40> (a java.lang.Object)
> at
> com.company.streaming.platform.kafka.DistributedHerder.updateConfigsWithIncrementalCooperative(DistributedHerder.java:514)
> - locked <0xe6136808> (a
> com.company.streaming.platform.kafka.DistributedHerder)
> at
> com.company.streaming.platform.kafka.DistributedHerder.tick(DistributedHerder.java:402)
> at
> com.company.streaming.platform.kafka.DistributedHerder.run(DistributedHerder.java:286)
> at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
> at java.lang.Thread.run(Thread.java:748){code}
> DistributedHerder went to updateConfigsWithIncrementalCooperative()
> synchronized method and called configBackingStore.snapshot() which take a
> lock on internal object in KafkaConfigBackingStore class.
>
> Meanwhile KafkaConfigBackingStore in ConsumeCallback inside synchronized
> block on internal object got SESSION_KEY record and called
> updateListener.onSessionKeyUpdate() which take a lock on DistributedHerder.
>
> As I can see the problem is here:
> [https://github.com/apache/kafka/blob/trunk/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaConfigBackingStore.java#L737]
>
> As I understand this call should be performed outside synchronized block:
> {code:java}
> if (started)
>
> updateListener.onSessionKeyUpdate(KafkaConfigBackingStore.this.sessionKey);{code}
>
> I'm going to make a PR.
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Updated] (KAFKA-10426) Deadlock in KafkaConfigBackingStore
[
https://issues.apache.org/jira/browse/KAFKA-10426?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Goltseva Taisiia updated KAFKA-10426:
-
Reviewer: Chris Egerton (was: Konstantine Karantasis)
> Deadlock in KafkaConfigBackingStore
> ---
>
> Key: KAFKA-10426
> URL: https://issues.apache.org/jira/browse/KAFKA-10426
> Project: Kafka
> Issue Type: Bug
> Components: KafkaConnect
>Affects Versions: 2.4.1, 2.6.0
>Reporter: Goltseva Taisiia
>Assignee: Goltseva Taisiia
>Priority: Critical
> Labels: pull-request-available
>
> Hi, guys!
> We faced the following deadlock:
>
> {code:java}
> KafkaBasedLog Work Thread - _streaming_service_config
> priority:5 - threadId:0x7f18ec22c000 - nativeId:0x950 - nativeId
> (decimal):2384 - state:BLOCKED
> stackTrace:
> java.lang.Thread.State: BLOCKED (on object monitor)
> at
> com.company.streaming.platform.kafka.DistributedHerder$ConfigUpdateListener.onSessionKeyUpdate(DistributedHerder.java:1586)
> - waiting to lock <0xe6136808> (a
> com.company.streaming.platform.kafka.DistributedHerder)
> at
> org.apache.kafka.connect.storage.KafkaConfigBackingStore$ConsumeCallback.onCompletion(KafkaConfigBackingStore.java:707)
> - locked <0xd8c3be40> (a java.lang.Object)
> at
> org.apache.kafka.connect.storage.KafkaConfigBackingStore$ConsumeCallback.onCompletion(KafkaConfigBackingStore.java:481)
> at org.apache.kafka.connect.util.KafkaBasedLog.poll(KafkaBasedLog.java:264)
> at
> org.apache.kafka.connect.util.KafkaBasedLog.access$500(KafkaBasedLog.java:71)
> at
> org.apache.kafka.connect.util.KafkaBasedLog$WorkThread.run(KafkaBasedLog.java:337)
> CustomDistributedHerder-connect-1
> priority:5 - threadId:0x7f1a01e30800 - nativeId:0x93a - nativeId
> (decimal):2362 - state:BLOCKED
> stackTrace:
> java.lang.Thread.State: BLOCKED (on object monitor)
> at
> org.apache.kafka.connect.storage.KafkaConfigBackingStore.snapshot(KafkaConfigBackingStore.java:285)
> - waiting to lock <0xd8c3be40> (a java.lang.Object)
> at
> com.company.streaming.platform.kafka.DistributedHerder.updateConfigsWithIncrementalCooperative(DistributedHerder.java:514)
> - locked <0xe6136808> (a
> com.company.streaming.platform.kafka.DistributedHerder)
> at
> com.company.streaming.platform.kafka.DistributedHerder.tick(DistributedHerder.java:402)
> at
> com.company.streaming.platform.kafka.DistributedHerder.run(DistributedHerder.java:286)
> at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
> at java.lang.Thread.run(Thread.java:748){code}
> DistributedHerder went to updateConfigsWithIncrementalCooperative()
> synchronized method and called configBackingStore.snapshot() which take a
> lock on internal object in KafkaConfigBackingStore class.
>
> Meanwhile KafkaConfigBackingStore in ConsumeCallback inside synchronized
> block on internal object got SESSION_KEY record and called
> updateListener.onSessionKeyUpdate() which take a lock on DistributedHerder.
>
> As I can see the problem is here:
> [https://github.com/apache/kafka/blob/trunk/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaConfigBackingStore.java#L737]
>
> As I understand this call should be performed outside synchronized block:
> {code:java}
> if (started)
>
> updateListener.onSessionKeyUpdate(KafkaConfigBackingStore.this.sessionKey);{code}
>
> I'm going to make a PR.
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Assigned] (KAFKA-10426) Deadlock in KafkaConfigBackingStore
[
https://issues.apache.org/jira/browse/KAFKA-10426?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Goltseva Taisiia reassigned KAFKA-10426:
Assignee: Goltseva Taisiia
> Deadlock in KafkaConfigBackingStore
> ---
>
> Key: KAFKA-10426
> URL: https://issues.apache.org/jira/browse/KAFKA-10426
> Project: Kafka
> Issue Type: Bug
> Components: KafkaConnect
>Affects Versions: 2.4.1, 2.6.0
>Reporter: Goltseva Taisiia
>Assignee: Goltseva Taisiia
>Priority: Critical
>
> Hi, guys!
> We faced the following deadlock:
>
> {code:java}
> KafkaBasedLog Work Thread - _streaming_service_config
> priority:5 - threadId:0x7f18ec22c000 - nativeId:0x950 - nativeId
> (decimal):2384 - state:BLOCKED
> stackTrace:
> java.lang.Thread.State: BLOCKED (on object monitor)
> at
> com.company.streaming.platform.kafka.DistributedHerder$ConfigUpdateListener.onSessionKeyUpdate(DistributedHerder.java:1586)
> - waiting to lock <0xe6136808> (a
> com.company.streaming.platform.kafka.DistributedHerder)
> at
> org.apache.kafka.connect.storage.KafkaConfigBackingStore$ConsumeCallback.onCompletion(KafkaConfigBackingStore.java:707)
> - locked <0xd8c3be40> (a java.lang.Object)
> at
> org.apache.kafka.connect.storage.KafkaConfigBackingStore$ConsumeCallback.onCompletion(KafkaConfigBackingStore.java:481)
> at org.apache.kafka.connect.util.KafkaBasedLog.poll(KafkaBasedLog.java:264)
> at
> org.apache.kafka.connect.util.KafkaBasedLog.access$500(KafkaBasedLog.java:71)
> at
> org.apache.kafka.connect.util.KafkaBasedLog$WorkThread.run(KafkaBasedLog.java:337)
> CustomDistributedHerder-connect-1
> priority:5 - threadId:0x7f1a01e30800 - nativeId:0x93a - nativeId
> (decimal):2362 - state:BLOCKED
> stackTrace:
> java.lang.Thread.State: BLOCKED (on object monitor)
> at
> org.apache.kafka.connect.storage.KafkaConfigBackingStore.snapshot(KafkaConfigBackingStore.java:285)
> - waiting to lock <0xd8c3be40> (a java.lang.Object)
> at
> com.company.streaming.platform.kafka.DistributedHerder.updateConfigsWithIncrementalCooperative(DistributedHerder.java:514)
> - locked <0xe6136808> (a
> com.company.streaming.platform.kafka.DistributedHerder)
> at
> com.company.streaming.platform.kafka.DistributedHerder.tick(DistributedHerder.java:402)
> at
> com.company.streaming.platform.kafka.DistributedHerder.run(DistributedHerder.java:286)
> at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
> at java.lang.Thread.run(Thread.java:748){code}
> DistributedHerder went to updateConfigsWithIncrementalCooperative()
> synchronized method and called configBackingStore.snapshot() which take a
> lock on internal object in KafkaConfigBackingStore class.
>
> Meanwhile KafkaConfigBackingStore in ConsumeCallback inside synchronized
> block on internal object got SESSION_KEY record and called
> updateListener.onSessionKeyUpdate() which take a lock on DistributedHerder.
>
> As I can see the problem is here:
> [https://github.com/apache/kafka/blob/trunk/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaConfigBackingStore.java#L737]
>
> As I understand this call should be performed outside synchronized block:
> {code:java}
> if (started)
>
> updateListener.onSessionKeyUpdate(KafkaConfigBackingStore.this.sessionKey);{code}
>
> I'm going to make a PR.
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Created] (KAFKA-10426) Deadlock in KafkaConfigBackingStore
Goltseva Taisiia created KAFKA-10426:
Summary: Deadlock in KafkaConfigBackingStore
Key: KAFKA-10426
URL: https://issues.apache.org/jira/browse/KAFKA-10426
Project: Kafka
Issue Type: Bug
Components: KafkaConnect
Affects Versions: 2.6.0, 2.4.1
Reporter: Goltseva Taisiia
Hi, guys!
We faced the following deadlock:
{code:java}
KafkaBasedLog Work Thread - _streaming_service_config
priority:5 - threadId:0x7f18ec22c000 - nativeId:0x950 - nativeId
(decimal):2384 - state:BLOCKED
stackTrace:
java.lang.Thread.State: BLOCKED (on object monitor)
at
com.company.streaming.platform.kafka.DistributedHerder$ConfigUpdateListener.onSessionKeyUpdate(DistributedHerder.java:1586)
- waiting to lock <0xe6136808> (a
com.company.streaming.platform.kafka.DistributedHerder)
at
org.apache.kafka.connect.storage.KafkaConfigBackingStore$ConsumeCallback.onCompletion(KafkaConfigBackingStore.java:707)
- locked <0xd8c3be40> (a java.lang.Object)
at
org.apache.kafka.connect.storage.KafkaConfigBackingStore$ConsumeCallback.onCompletion(KafkaConfigBackingStore.java:481)
at org.apache.kafka.connect.util.KafkaBasedLog.poll(KafkaBasedLog.java:264)
at org.apache.kafka.connect.util.KafkaBasedLog.access$500(KafkaBasedLog.java:71)
at
org.apache.kafka.connect.util.KafkaBasedLog$WorkThread.run(KafkaBasedLog.java:337)
CustomDistributedHerder-connect-1
priority:5 - threadId:0x7f1a01e30800 - nativeId:0x93a - nativeId
(decimal):2362 - state:BLOCKED
stackTrace:
java.lang.Thread.State: BLOCKED (on object monitor)
at
org.apache.kafka.connect.storage.KafkaConfigBackingStore.snapshot(KafkaConfigBackingStore.java:285)
- waiting to lock <0xd8c3be40> (a java.lang.Object)
at
com.company.streaming.platform.kafka.DistributedHerder.updateConfigsWithIncrementalCooperative(DistributedHerder.java:514)
- locked <0xe6136808> (a
com.company.streaming.platform.kafka.DistributedHerder)
at
com.company.streaming.platform.kafka.DistributedHerder.tick(DistributedHerder.java:402)
at
com.company.streaming.platform.kafka.DistributedHerder.run(DistributedHerder.java:286)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748){code}
DistributedHerder went to updateConfigsWithIncrementalCooperative()
synchronized method and called configBackingStore.snapshot() which take a lock
on internal object in KafkaConfigBackingStore class.
Meanwhile KafkaConfigBackingStore in ConsumeCallback inside synchronized block
on internal object got SESSION_KEY record and called
updateListener.onSessionKeyUpdate() which take a lock on DistributedHerder.
As I can see the problem is here:
[https://github.com/apache/kafka/blob/trunk/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaConfigBackingStore.java#L737]
As I understand this call should be performed outside synchronized block:
{code:java}
if (started)
updateListener.onSessionKeyUpdate(KafkaConfigBackingStore.this.sessionKey);{code}
I'm going to make a PR.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Commented] (KAFKA-560) Garbage Collect obsolete topics
[ https://issues.apache.org/jira/browse/KAFKA-560?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17120958#comment-17120958 ] Goltseva Taisiia commented on KAFKA-560: I'm looking for an option on Kafka for auto deletion of old unused topics. There is no such option, right? I wonder why? I think it's no less important than deleting old messages via retention policy. Because there are collecting lot's of unused topics consuming recources in dev environments. It would be very useful from my point of view to delete such old topics automatically. > Garbage Collect obsolete topics > --- > > Key: KAFKA-560 > URL: https://issues.apache.org/jira/browse/KAFKA-560 > Project: Kafka > Issue Type: New Feature >Reporter: Jay Kreps >Assignee: Harsha >Priority: Major > Labels: project > > Old junk topics tend to accumulate over time. Code may migrate to use new > topics leaving the old ones orphaned. Likewise there are some use cases for > temporary transient topics. It would be good to have a tool that could delete > any topic that had not been written to in a configurable period of time and > had no active consumer groups. Something like >./bin/delete-unused-topics.sh --last-write [date] --zookeeper [zk_connect] > This requires API support to get the last update time. I think it may be > possible to do this through the OffsetRequest now? -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (KAFKA-7878) Connect Task already exists in this worker when failed to create consumer
[
https://issues.apache.org/jira/browse/KAFKA-7878?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16925725#comment-16925725
]
Goltseva Taisiia commented on KAFKA-7878:
-
We have the similar problem. Sometimes after Kafka had problems for some time,
error occurs during tasks restarting - *Task already exists in this worker.*
It's also similar to:
[https://stackoverflow.com/questions/43802156/inconsistent-connector-state-connectexception-task-already-exists-in-this-work]
Only process restarting fixes the issue.
> Connect Task already exists in this worker when failed to create consumer
> -
>
> Key: KAFKA-7878
> URL: https://issues.apache.org/jira/browse/KAFKA-7878
> Project: Kafka
> Issue Type: Bug
>Affects Versions: 1.0.1, 2.0.1
>Reporter: Loïc Monney
>Priority: Major
>
> *Assumption*
> 1. DNS is not available during a few minutes
> 2. Consumer group rebalances
> 3. Client is not able to resolve DNS entries anymore and fails
> 4. Task seems already registered, so at next rebalance the task will fail due
> to *Task already exists in this worker* and the only way to recover is to
> restart the connect process
> *Real log entries*
> * Distributed cluster running one connector on top of Kubernetes
> * Connect 2.0.1
> * kafka-connect-hdfs 5.0.1
> {noformat}
> [2019-01-28 13:31:25,914] WARN Removing server kafka.xxx.net:9093 from
> bootstrap.servers as DNS resolution failed for kafka.xxx.net
> (org.apache.kafka.clients.ClientUtils:56)
> [2019-01-28 13:31:25,915] ERROR WorkerSinkTask\{id=xxx-22} Task failed
> initialization and will not be started.
> (org.apache.kafka.connect.runtime.WorkerSinkTask:142)
> org.apache.kafka.connect.errors.ConnectException: Failed to create consumer
> at
> org.apache.kafka.connect.runtime.WorkerSinkTask.createConsumer(WorkerSinkTask.java:476)
> at
> org.apache.kafka.connect.runtime.WorkerSinkTask.initialize(WorkerSinkTask.java:139)
> at org.apache.kafka.connect.runtime.Worker.startTask(Worker.java:452)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.startTask(DistributedHerder.java:873)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.access$1600(DistributedHerder.java:111)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder$13.call(DistributedHerder.java:888)
> at
> org.apache.kafka.connect.runtime.distributed.DistributedHerder$13.call(DistributedHerder.java:884)
> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
> at java.lang.Thread.run(Thread.java:748)
> Caused by: org.apache.kafka.common.KafkaException: Failed to construct kafka
> consumer
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.(KafkaConsumer.java:799)
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.(KafkaConsumer.java:615)
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.(KafkaConsumer.java:596)
> at
> org.apache.kafka.connect.runtime.WorkerSinkTask.createConsumer(WorkerSinkTask.java:474)
> ... 10 more
> Caused by: org.apache.kafka.common.config.ConfigException: No resolvable
> bootstrap urls given in bootstrap.servers
> at
> org.apache.kafka.clients.ClientUtils.parseAndValidateAddresses(ClientUtils.java:66)
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.(KafkaConsumer.java:709)
> ... 13 more
> [2019-01-28 13:31:25,925] INFO Finished starting connectors and tasks
> (org.apache.kafka.connect.runtime.distributed.DistributedHerder:868)
> [2019-01-28 13:31:25,926] INFO Rebalance started
> (org.apache.kafka.connect.runtime.distributed.DistributedHerder:1239)
> [2019-01-28 13:31:25,927] INFO Stopping task xxx-22
> (org.apache.kafka.connect.runtime.Worker:555)
> [2019-01-28 13:31:26,021] INFO Finished stopping tasks in preparation for
> rebalance
> (org.apache.kafka.connect.runtime.distributed.DistributedHerder:1269)
> [2019-01-28 13:31:26,021] INFO [Worker clientId=connect-1,
> groupId=xxx-cluster] (Re-)joining group
> (org.apache.kafka.clients.consumer.internals.AbstractCoordinator:509)
> [2019-01-28 13:31:30,746] INFO [Worker clientId=connect-1,
> groupId=xxx-cluster] Successfully joined group with generation 29
> (org.apache.kafka.clients.consumer.internals.AbstractCoordinator:473)
> [2019-01-28 13:31:30,746] INFO Joined group and got assignment:
> Assignment\{error=0, leader='connect-1-05961f03-52a7-4c02-acc2-0f1fb021692e',
> leaderUrl='http://192.168.46.59:8083/', offset=32, connectorIds=[],
> taskIds=[xxx-22]}
> (org.apache.kafka.connect.runtime.distributed.DistributedHerder:1217)
> [2019-01-28 13:31:30,747] INFO Starting connectors and tasks using c
