将胶水模式注册表与 MSK 连接器结合使用

Question

我一直在尝试创建一个MSK连接器，并使用Glue schema注册表。配置如下。

connector.class=io.confluent.connect.s3.S3SinkConnector
s3.region=eu-west-1
topics.dir=topics/dir
flush.size=200
tasks.max=2
s3.part.size=5242880
timezone=GMT
# value.converter.schema.registry.url=http://someIP:8081
key.converter.schemaName=my-topic-schema
locale=US
format.class=io.confluent.connect.s3.format.parquet.ParquetFormat
value.converter.schemaName=my-topic-schema
value.converter=io.confluent.connect.avro.StringConverter
s3.bucket.name=my-bucket
key.converter=io.confluent.connect.avro.StringConverter
# key.converter.schema.registry.url==http://someIP:8081
partition.duration.ms=3600000
schema.compatibility=BACKWARD
topics=osb
value.converter.registry.name=Glue-Schema-Registry
key.converter.registry.name=Glue-Schema-Registry
key.converter.schemas.enable=true
partitioner.class=io.confluent.connect.storage.partitioner.TimeBasedPartitioner
value.converter.schemas.enable=true
storage.class=io.confluent.connect.s3.storage.S3Storage
rotate.schedule.interval.ms=0
path.format='year'=YYYY/'month'=MM/'day'=dd/'hour'=HH
timestamp.extractor=RecordField
timestamp.field=timestamp

首先，我使用运行在EC2上的合流模式注册中心，我在“key/value . converter . schema . registry . URL”字段中添加了EC2的ip，它工作得很好。现在我尝试使用Glue Schema Registry。但是我不知道如何用Glue Schema Registry连接连接器。

Answer 1

这些类不存在

key.converter=io.confluent.connect.avro.StringConverter
value.converter=io.confluent.connect.avro.StringConverter

StringConverterclassname以org.apache.kafka开头

同样，看起来您已经添加了一堆随机转换器属性，除了 url 之外，这些属性对于字符串或融合 Avro 转换器无效

要使用Glue，您需要使用＜code＞AWSKafkaAvroConverter＜/code＞，它是这个repo的一部分，

https://github . com/aw slabs/AWS-glue-schema-registry/tree/master/avro-kafkaconnect-converter

并且记录在这里

https://docs.aws.amazon.com/glue/latest/dg/schema-registry-integrations.html#schema-apache kafka connect的注册表集成