处理无命名空间Avro Schema的策略与实践

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;

import java.io.IOException;

public class AvroSchemaNamespaceInjector {

    public static String ensureNamespace(String avroSchemaJson, String defaultNamespace) throws IOException {
        ObjectMapper mapper = new ObjectMapper();
        JsonNode schemaNode = mapper.readTree(avroSchemaJson);

        if (schemaNode.isObject()) {
            ObjectNode objectNode = (ObjectNode) schemaNode;
            if (!objectNode.has("namespace")) {
                objectNode.put("namespace", defaultNamespace);
                return mapper.writeValueAsString(objectNode);
            }
        }
        return avroSchemaJson; // Schema already has a namespace or is not an object
    }

    public static void main(String[] args) throws IOException {
        String originalSchema = "{\"type\": \"record\", \"name\": \"MyClass\", \"fields\": [{\"name\": \"id\", \"type\": \"int\"}]}";
        String modifiedSchema = ensureNamespace(originalSchema, "com.example.avro");
        System.out.println("Original Schema: " + originalSchema);
        System.out.println("Modified Schema: " + modifiedSchema);

        String schemaWithNamespace = "{\"type\": \"record\", \"name\": \"MyClass\", \"namespace\": \"com.existing.avro\", \"fields\": [{\"name\": \"id\", \"type\": \"int\"}]}";
        String unchangedSchema = ensureNamespace(schemaWithNamespace, "com.example.avro");
        System.out.println("Schema with existing namespace: " + unchangedSchema);
    }
}

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericRecord;
import io.confluent.kafka.serializers.KafkaAvroDeserializer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class GenericAvroConsumer {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "generic-avro-consumer-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class.getName());
        props.put("schema.registry.url", "http://localhost:8081"); // Confluent Schema Registry URL
        // props.put("specific.avro.reader", "false"); // 确保不尝试反序列化为SpecificRecord

        KafkaConsumer<String, GenericRecord> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("your-avro-topic"));

        try {
            while (true) {
                ConsumerRecords<String, GenericRecord> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, GenericRecord> record : records) {
                    GenericRecord avroRecord = record.value();
                    System.out.printf("Offset = %d, Key = %s, Value = %s\n", record.offset(), record.key(), avroRecord);

                    // 访问GenericRecord的字段
                    // 假设Schema有一个名为"id"的int字段和一个名为"name"的String字段
                    Integer id = (Integer) avroRecord.get("id");
                    String name = (String) avroRecord.get("name");
                    System.out.printf("Parsed: ID = %d, Name = %s\n", id, name);
                }
            }
        } finally {
            consumer.close();
        }
    }
}