使用Kafka S3 Sink Connector时，能否通过Schema Registry解决导出Schema不一致问题？

Absolutely! You can absolutely leverage the Schema Registry to retrieve your original Avro schema and fix the discrepancies (like broken nested structures) you're seeing with the default schema generator. The DefaultSchemaGenerator regenerates schemas from the Connect data model, which can flatten or alter complex types—but using a Schema Registry-aware generator will preserve your original schema exactly.

Here's how to configure it properly:

1. Replace the default schema generator class #

Swap out io.confluent.connect.storage.hive.schema.DefaultSchemaGenerator with the Schema Registry-compatible alternative:

schema.generator.class=io.confluent.connect.storage.hive.schema.SchemaRegistrySchemaGenerator

2. Add Schema Registry connection properties #

You’ll need to include configuration to point the connector to your Schema Registry, plus match the subject naming strategy used when producing messages:

• schema.registry.url: The URL of your Schema Registry instance (e.g., http://schema-registry:8081)
• schema.name.strategy: The strategy used to register your original schema (common options include io.confluent.kafka.serializers.subject.TopicNameStrategy or RecordNameStrategy—make sure this matches what your producers use)
• Ensure your value converter is already configured to use the Avro converter with Schema Registry (if not, add these):

  value.converter=io.confluent.connect.avro.AvroConverter
  value.converter.schema.registry.url=http://schema-registry:8081
  

3. Example full configuration snippet #

Here’s what a working config might look like for your use case:

name=s3-sink-avro-original-schema
connector.class=io.confluent.connect.s3.S3SinkConnector
tasks.max=2
topics=your-target-topic
s3.bucket.name=your-s3-bucket
s3.region=us-east-1
format.class=io.confluent.connect.s3.format.avro.AvroFormat
storage.class=io.confluent.connect.s3.storage.S3Storage
partitioner.class=io.confluent.connect.storage.partitioner.DefaultPartitioner

# Schema Registry-specific configs
schema.generator.class=io.confluent.connect.storage.hive.schema.SchemaRegistrySchemaGenerator
schema.registry.url=http://schema-registry:8081
schema.name.strategy=io.confluent.kafka.serializers.subject.TopicNameStrategy

# Avro converter configs
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://schema-registry:8081

4. Why this fixes your issue #

The SchemaRegistrySchemaGenerator doesn’t reinvent the wheel—it directly fetches the exact Avro schema from your Schema Registry (matching the subject name based on your strategy) instead of generating a new one from Connect’s internal data representation. This ensures nested structures, field types, and all other schema details are preserved exactly as they were in the original Kafka messages.

Troubleshooting tips #

• Double-check that your connector can reach the Schema Registry URL (no network blocks or incorrect ports)
• Verify the subject name in Schema Registry matches what your strategy produces: for TopicNameStrategy, the subject will be your-topic-name-value (for value schemas)
• Ensure the original schema is still present in Schema Registry and hasn’t been deleted or overwritten with an incompatible version

内容的提问来源于stack exchange，提问作者Xiang Zhang