we have a Kafka cluster (as a 3rd party hosted service), which has SSL enabled. We are now trying to setup Kafka Connect (Confluent 5.0) with a 3rd party Sink (WePay BigQuery connector). When starting Kafka connect in standalone mode, everything works like a charm. Unfortunately when enabling distributed mode, Kafka Connect suddenly fails with the following:
[2018-09-25 15:01:46,248] INFO Kafka version : 1.0.0 (org.apache.kafka.common.utils.AppInfoParser:109)
[2018-09-25 15:01:46,248] INFO Kafka commitId : aaa7af6d4a11b29d (org.apache.kafka.common.utils.AppInfoParser:110)
[2018-09-25 15:01:46,667] INFO Kafka cluster ID: Q9PaAEeWSbOavVmHTQS5sA (org.apache.kafka.connect.util.ConnectUtils:59)
[2018-09-25 15:01:46,685] INFO Logging initialized @10512ms to org.eclipse.jetty.util.log.Slf4jLog (org.eclipse.jetty.util.log:193)
[2018-09-25 15:01:46,726] INFO Added connector for http://:8083 (org.apache.kafka.connect.runtime.rest.RestServer:119)
[2018-09-25 15:01:46,760] INFO Advertised URI: http://192.168.4.207:8083/ (org.apache.kafka.connect.runtime.rest.RestServer:267)
[2018-09-25 15:01:46,796] INFO Kafka version : 1.0.0 (org.apache.kafka.common.utils.AppInfoParser:109)
[2018-09-25 15:01:46,796] INFO Kafka commitId : aaa7af6d4a11b29d (org.apache.kafka.common.utils.AppInfoParser:110)
ERROR Stopping due to error
(org.apache.kafka.connect.cli.ConnectDistributed:117)
java.lang.NoSuchMethodError:
org.apache.kafka.common.metrics.Sensor.add
(Lorg/apache/kafka/common/metrics/CompoundStat;)Z
at org.apache.kafka.connect.runtime.Worker$WorkerMetricsGroup.<init> .
(Worker.java:731)
at org.apache.kafka.connect.runtime.Worker.<init>(Worker.java:112)
at
org.apache.kafka.connect.cli.ConnectDistributed.main
(ConnectDistributed.java:88)
Tried to Google for the specific error, but couldn't find anything. It looks like a version issue somewhere (hence the NoSuchMethodError), but have no clue where to start.
When used with Confluent 4.1.2 there's a different error:
[2018-09-26 15:14:05,498] ERROR Stopping due to error (org.apache.kafka.connect.cli.ConnectDistributed:112)
org.apache.kafka.common.KafkaException: Failed to construct kafka consumer
at org.apache.kafka.connect.runtime.distributed.WorkerGroupMember.<init>(WorkerGroupMember.java:144)
at org.apache.kafka.connect.runtime.distributed.DistributedHerder.<init>(DistributedHerder.java:182)
at org.apache.kafka.connect.runtime.distributed.DistributedHerder.<init>(DistributedHerder.java:159)
at org.apache.kafka.connect.cli.ConnectDistributed.main(ConnectDistributed.java:95)
Caused by: java.lang.NoSuchMethodError: org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.<init>(Lorg/apache/kafka/common/utils/LogContext;Lorg/apache/kafka/clients/KafkaClient;Lorg/apache/kafka/clients/Metadata;Lorg/apache/kafka/common/utils/Time;JJI)V
at org.apache.kafka.connect.runtime.distributed.WorkerGroupMember.<init>(WorkerGroupMember.java:114)
... 3 more
When we use the same but with Kafka Connect (Confluent 3.0), there's a different error:
[2018-09-26 10:04:24,588] INFO AvroDataConfig values:
schemas.cache.config = 1000
enhanced.avro.schema.support = false
connect.meta.data = true
(io.confluent.connect.avro.AvroDataConfig:169)
Exception in thread "main" java.lang.NoSuchMethodError: org.apache.kafka.common.utils.AppInfoParser.unregisterAppInfo(Ljava/lang/String;Ljava/lang/String;)V
at org.apache.kafka.connect.runtime.distributed.WorkerGroupMember.stop(WorkerGroupMember.java:194)
at org.apache.kafka.connect.runtime.distributed.WorkerGroupMember.<init>(WorkerGroupMember.java:122)
at org.apache.kafka.connect.runtime.distributed.DistributedHerder.<init>(DistributedHerder.java:150)
at org.apache.kafka.connect.runtime.distributed.DistributedHerder.<init>(DistributedHerder.java:132)
at org.apache.kafka.connect.cli.ConnectDistributed.main(ConnectDistributed.java:82)
Here's the distributed.properties:
bootstrap.servers=*****
group.id=testGroup
key.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=****
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=****
internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter
internal.key.converter.schemas.enable=false
internal.value.converter.schemas.enable=false
config.storage.topic=connect-configs
offset.storage.topic=connect-offsets
status.storage.topic=connect-statuses
security.protocol=SSL
ssl.truststore.location=truststore.jks
ssl.truststore.password=****
ssl.keystore.type=PKCS12
ssl.keystore.location=keystore.p12
ssl.keystore.password=****
ssl.key.password=****
plugin.path=/*/confluent-5.0.0/share/java
And for reference the standalone.properties:
bootstrap.servers=***
key.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=***
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=***
internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter
internal.key.converter.schemas.enable=false
internal.value.converter.schemas.enable=false
offset.storage.file.filename=connect.offsets
consumer.security.protocol=SSL
consumer.ssl.truststore.location=truststore.jks
consumer.ssl.truststore.password=***
consumer.ssl.keystore.type=PKCS12
consumer.ssl.keystore.location=keystore.p12
consumer.ssl.keystore.password=***
consumer.ssl.key.password=***
Any help would be much appreciated.
0.10.0, I believe, you want 4.1.x, to get1.1, which is what the wepay connector is built against. Yes, the connect can be newer than the brokers, but the implementation of the connectors themselves must share the common Connect API versions and exclude them when actually built (you might want to post an issue to their Github with your errors) – OneCricketeer