cassandra - 使用 AvroConverter 启动时出现 Kafka Connect 错误

Question

我正试图从一个话题转向 Cassandra。该主题具有由 kafka 流创建的 avro 文件的模式注册表。

这是我的 connect-standalone.properties:

bootstrap.servers=localhost:9092

key.converter=org.apache.kafka.connect.json.JsonConverter 
value.converter=org.apache.kafka.connect.json.JsonConverter

key.converter.schemas.enable=true   
value.converter.schemas.enable=true

offset.storage.file.filename=/tmp/connect.offsets  
offset.flush.interval.ms=10000

plugin.path=/usr/share/java

这是我的 cassandra-sink.properties:

connector.class=io.confluent.connect.cassandra.CassandraSinkConnector 
cassandra.contact.points=ip.to.endpoint 
cassandra.port=9042 
cassandra.keyspace=my_keyspace 
cassandra.write.mode=Update 
tasks.max=1 
topics=avro-sink 
key.converter.schema.registry.url=http://localhost:8081 
value.converter.schema.registry.url=http://localhost:8081 
key.converter=org.apache.kafka.connect.storage.StringConverter 
value.converter=org.apache.kafka.connect.avro.AvroConverter 
name=cassandra-sink-connector 
internal.key.converter=org.apache.kafka.connect.storage.StringConverter
internal.value.converter=org.apache.kafka.connect.avro.AvroConverter
transforms=createKey 
transforms.createKey.fields=id,timestamp
transforms.createKey.type=org.apache.kafka.connect.transforms.ValueToKey

这是错误:

[2019-03-15 16:32:05,238] ERROR Failed to create job for /etc/kafka/cassandra-sink.properties (org.apache.kafka.connect.cli.ConnectStandalone:108)
[2019-03-15 16:32:05,238] ERROR Stopping after connector error (org.apache.kafka.connect.cli.ConnectStandalone:119)
java.util.concurrent.ExecutionException: org.apache.kafka.connect.runtime.rest.errors.BadRequestException: Connector configuration is invalid and contains the following 1 error(s):
Invalid value org.apache.kafka.connect.avro.AvroConverter for configuration value.converter: Class org.apache.kafka.connect.avro.AvroConverter could not be found.
You can also find the above list of errors at the endpoint `/{connectorType}/config/validate`
        at org.apache.kafka.connect.util.ConvertingFutureCallback.result(ConvertingFutureCallback.java:79)
        at org.apache.kafka.connect.util.ConvertingFutureCallback.get(ConvertingFutureCallback.java:66)
        at org.apache.kafka.connect.cli.ConnectStandalone.main(ConnectStandalone.java:116)
Caused by: org.apache.kafka.connect.runtime.rest.errors.BadRequestException: Connector configuration is invalid and contains the following 1 error(s):
Invalid value org.apache.kafka.connect.avro.AvroConverter for configuration value.converter: Class org.apache.kafka.connect.avro.AvroConverter could not be found.
You can also find the above list of errors at the endpoint `/{connectorType}/config/validate`
        at org.apache.kafka.connect.runtime.AbstractHerder.maybeAddConfigErrors(AbstractHerder.java:423)
        at org.apache.kafka.connect.runtime.standalone.StandaloneHerder.putConnectorConfig(StandaloneHerder.java:188)
        at org.apache.kafka.connect.cli.ConnectStandalone.main(ConnectStandalone.java:113)

此经纪人是可以在 https://github.com/confluentinc/cp-docker-images 中找到的经纪人

有什么想法吗？谢谢!

Answer 1

我通过删除所有 converter 属性解决了这个问题，因为在融合的 docker 容器上，默认值已经是 avro 值和键值字符串