模式注册表允许奇怪的Avro模式更改。

在使用Avro模式注册表时，可以通过以下方式允许奇怪的Avro模式更改：

import io.confluent.kafka.serializers.AbstractKafkaAvroSerDeConfig;
import io.confluent.kafka.serializers.KafkaAvroSerializer;
import org.apache.avro.Schema;
import org.apache.avro.SchemaParseException;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class AvroSchemaRegistryExample {

    private static final String TOPIC_NAME = "your-topic-name";
    private static final String SCHEMA_REGISTRY_URL = "http://your-schema-registry-url";

    public static void main(String[] args) {
        // Configure Kafka producer
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "your-bootstrap-servers");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
        properties.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, SCHEMA_REGISTRY_URL);

        // Create Kafka producer
        KafkaProducer<String, Object> producer = new KafkaProducer<>(properties);

        // Create a strange Avro schema
        String strangeSchemaString = "{\"type\":\"record\",\"name\":\"strange_record\",\"fields\":[{\"name\":\"field1\",\"type\":\"string\"}]}";
        Schema strangeSchema;
        try {
            strangeSchema = new Schema.Parser().parse(strangeSchemaString);
        } catch (SchemaParseException e) {
            e.printStackTrace();
            return;
        }

        // Create a strange Avro record
        GenericRecord strangeRecord = new GenericData.Record(strangeSchema);
        strangeRecord.put("field1", "some value");

        // Create a Kafka record with the strange Avro record
        ProducerRecord<String, Object> kafkaRecord = new ProducerRecord<>(TOPIC_NAME, strangeRecord);

        // Send the Kafka record
        producer.send(kafkaRecord, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if (exception != null) {
                    exception.printStackTrace();
                } else {
                    System.out.println("Message sent successfully to Kafka topic: " + metadata.topic());
                    System.out.println("Partition: " + metadata.partition());
                    System.out.println("Offset: " + metadata.offset());
                }
            }
        });

        // Close the Kafka producer
        producer.close();
    }
}

在上面的代码中，我们使用KafkaAvroSerializer来将Avro记录序列化为字节数组，并使用AbstractKafkaAvroSerDeConfig来指定Avro模式注册表的URL。我们创建了一个奇怪的Avro模式和相应的Avro记录，并将其发送到Kafka主题中。