Kafka是否原生支持模式注册表，还是这是一个Confluent平台的特性？

Kafka本身不原生支持模式注册表，但可以使用Confluent平台提供的Schema Registry来实现模式注册和管理。下面是一个使用Schema Registry的代码示例：

// 导入所需的依赖
import io.confluent.kafka.serializers.KafkaAvroSerializer;
import io.confluent.kafka.serializers.KafkaAvroDeserializer;
import io.confluent.kafka.serializers.AbstractKafkaAvroSerDeConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Collections;
import java.util.Properties;

public class KafkaSchemaRegistryExample {
    public static void main(String[] args) {
        // 设置Kafka和Schema Registry的连接信息
        String bootstrapServers = "localhost:9092";
        String schemaRegistryUrl = "http://localhost:8081";

        // 创建生产者配置
        Properties producerProps = new Properties();
        producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
        producerProps.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);

        // 创建生产者
        KafkaProducer<String, User> producer = new KafkaProducer<>(producerProps);

        // 创建消费者配置
        Properties consumerProps = new Properties();
        consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group");
        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class);
        consumerProps.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);

        // 创建消费者
        KafkaConsumer<String, User> consumer = new KafkaConsumer<>(consumerProps);

        // 订阅主题
        consumer.subscribe(Collections.singletonList("test-topic"));

        // 发送消息并消费消息
        for (int i = 0; i < 10; i++) {
            // 创建消息
            User user = new User("User " + i, i);
            ProducerRecord<String, User> record = new ProducerRecord<>("test-topic", user.getName(), user);

            // 发送消息
            producer.send(record);

            // 消费消息
            ConsumerRecords<String, User> records = consumer.poll(1000);
            for (ConsumerRecord<String, User> consumerRecord : records) {
                System.out.println("Received: " + consumerRecord.value());
            }
        }

        // 关闭生产者和消费者
        producer.close();
        consumer.close();
    }
}

以上代码示例演示了如何使用Schema Registry进行数据序列化和反序列化。在代码中，通过设置AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG属性来指定Schema Registry的URL。然后，使用KafkaAvroSerializer和KafkaAvroDeserializer来对Avro数据进行序列化和反序列化。

请注意，为了使上述代码示例正常工作，您需要添加相应的依赖，例如confluent-kafka-avro-serializer和confluent-kafka-avro-deserializer。