如何从kstream应用中读取Kafka消息头

import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;

import java.util.Iterator;

public class KafkaStreamApp {
    public static void main(String[] args) {
        // 创建Kafka Streams实例并设置配置参数
        KafkaStreams streams = new KafkaStreams(getTopology(), getProperties());

        // 启动Kafka Streams应用
        streams.start();
    }

    private static StreamsBuilder getTopology() {
        StreamsBuilder builder = new StreamsBuilder();

        // 创建一个KStream对象，表示从Kafka主题中读取的消息流
        KStream<String, String> stream = builder.stream("your_topic");

        // 对每条消息进行处理
        stream.map((key, value) -> {
            // 获取消息的头部信息
            Headers headers = stream.headers();

            // 遍历头部信息的键值对
            Iterator<Header> iterator = headers.iterator();
            while (iterator.hasNext()) {
                Header header = iterator.next();
                String headerKey = header.key();
                byte[] headerValue = header.value();

                // 处理头部信息，例如打印
                System.out.println("Header: " + headerKey + " = " + new String(headerValue));
            }

            // 返回处理后的消息
            return value;
        });

        return builder;
    }

    private static Properties getProperties() {
        Properties props = new Properties();
        // 设置Kafka集群的地址
        props.put("bootstrap.servers", "your_bootstrap_servers");
        // 设置消费者组ID
        props.put("group.id", "your_consumer_group_id");
        // 其他配置参数...

        return props;
    }
}