处理无命名空间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.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;

public class AvroSchemaModifier {

    /**
     * 读取AVSC文件内容，并为其动态添加命名空间。
     * 如果Schema中已存在命名空间，则不做修改。
     *
     * @param avscFilePath 原始AVSC文件的路径。
     * @param targetNamespace 要添加的目标命名空间。
     * @return 带有命名空间的Schema JSON字符串。
     * @throws IOException 文件读取或JSON处理异常。
     */
    public static String addNamespaceToSchema(String avscFilePath, String targetNamespace) throws IOException {
        String schemaContent = new String(Files.readAllBytes(Paths.get(avscFilePath)));
        ObjectMapper mapper = new ObjectMapper();
        JsonNode schemaNode = mapper.readTree(schemaContent);

        // 检查是否为对象类型且不包含namespace字段
        if (schemaNode.isObject() && !schemaNode.has("namespace")) {
            ((ObjectNode) schemaNode).put("namespace", targetNamespace);
        }
        return mapper.writerWithDefaultPrettyPrinter().writeValueAsString(schemaNode);
    }

    public static void main(String[] args) {
        try {
            String originalSchemaPath = "path/to/your/schema.avsc"; // 替换为你的AVSC文件路径
            String newNamespace = "com.example.avro"; // 定义一个命名空间
            String modifiedSchemaJson = addNamespaceToSchema(originalSchemaPath, newNamespace);
            System.out.println("Modified Schema with Namespace:\n" + modifiedSchemaJson);

            // 实际应用中，你需要将 modifiedSchemaJson 写入一个临时文件，
            // 然后配置 Avro Maven 插件指向这个临时文件来生成 Java 类。
            // 例如，在pom.xml中配置avro-maven-plugin，指向这个临时文件：
            /*
            <plugin>
                <groupId>org.apache.avro</groupId>
                <artifactId>avro-maven-plugin</artifactId>
                <version>1.11.1</version>
                <executions>
                    <execution>
                        <phase>generate-sources</phase>
                        <goals>
                            <goal>schema</goal>
                        </goals>
                        <configuration>
                            <sourceDirectory>${project.build.directory}/generated-avro-schemas</sourceDirectory>
                            <outputDirectory>${project.build.directory}/generated-sources/avro</outputDirectory>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
            */
            // 然后在构建前，通过Java代码或脚本将修改后的Schema写入
            // ${project.build.directory}/generated-avro-schemas 目录下。

        } catch (IOException e) {
            System.err.println("Error processing Avro schema: " + e.getMessage());
            e.printStackTrace();
        }
    }
}

import io.confluent.kafka.serializers.KafkaAvroDeserializer;
import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

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

public class KafkaGenericAvroConsumer {

    public static void main(String[] args) {
        Properties props = new Properties();
        // Kafka Broker 地址
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        // 消费者组ID
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-avro-consumer-group");
        // Key的反序列化器
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // Value的反序列化器，使用Confluent的KafkaAvroDeserializer
        props.put(ConsumerConfig.VALUE_DESERIAL