如何处理无命名空间的Avro Schema：Java与Kafka集成指南

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;

public class AvroSchemaNamespaceInjector {

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

        if (schemaNode.isObject() && schemaNode.has("name") && !schemaNode.has("namespace")) {
            ((ObjectNode) schemaNode).put("namespace", defaultNamespace);
            System.out.println("Namespace '" + defaultNamespace + "' injected into schema: " + schemaNode.get("name").asText());
        } else if (!schemaNode.isObject() || !schemaNode.has("name")) {
            System.err.println("Warning: Schema content is not a valid Avro record schema or already has a namespace.");
        }
        return mapper.writerWithDefaultPrettyPrinter().writeValueAsString(schemaNode);
    }

    public static void main(String[] args) throws IOException {
        String originalAvscPath = "path/to/your/schema_without_namespace.avsc";
        String modifiedAvscPath = "path/to/your/schema_with_namespace.avsc";
        String content = new String(Files.readAllBytes(Paths.get(originalAvscPath)));

        String modifiedContent = injectNamespace(content, "com.example.avro");
        Files.write(Paths.get(modifiedAvscPath), modifiedContent.getBytes());

        System.out.println("Modified schema saved to: " + modifiedAvscPath);
        // Now use modifiedAvscPath for Avro code generation
    }
}

bootstrap.servers=localhost:9092
schema.registry.url=http://localhost:8081
key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
value.deserializer=io.confluent.kafka.serializers.KafkaAvroDeserializer
specific.avro.reader=false # 关键配置，设置为false表示使用GenericRecord
group.id=my_consumer_group
auto.offset.reset=earliest

import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class GenericRecordKafkaConsumer {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("schema.registry.url", "http://localhost:8081");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "io.confluent.kafka.serializers.KafkaAvroDeserializer");
        props.put("specific.avro.reader", "false"); // 启用GenericRecord
        props.put("group.id", "my_generic_consumer_group");
        props.put("auto.offset.reset", "earliest");

        try (KafkaConsumer<String, GenericRecord> consumer = new KafkaConsumer<>(props)) {
            consumer.subscribe(Collections.singletonList("my_avro_topic"));

            while (true) {
                ConsumerRecords<String, GenericRecord> records = consumer.poll(Duration.ofMillis(100));
                records.forEach(record -> {
                    GenericRecord genericRecord = record.value();
                    System.out.println("Received message:");
                    // 通过字段名访问数据
                    if (genericRecord != null) {
                        genericRecord.getSchema().getFields().forEach(field -> {
                            System.out.println("  " + field.name() + ": " + genericRecord.get(field.name()));
                        });
                    }
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}