如何将Protobuf数据从Flink转发到Kafka和stdout？

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;
import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema;
import org.apache.flink.streaming.util.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;

import java.util.Properties;

public class ProtobufFlinkKafkaExample {

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置Kafka相关配置
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092");

        // 从Kafka读取Protobuf数据
        FlinkKafkaConsumer<MyProtobufMessage> kafkaConsumer = new FlinkKafkaConsumer<>(
                "input-topic",
                new ProtobufDeserializationSchema<>(MyProtobufMessage.class),
                properties
        );
        DataStream<MyProtobufMessage> dataStream = env.addSource(kafkaConsumer);

        // 将Protobuf数据转发到Kafka
        FlinkKafkaProducer<MyProtobufMessage> kafkaProducer = new FlinkKafkaProducer<>(
                "output-topic",
                new ProtobufSerializationSchema<>(MyProtobufMessage.class),
                properties
        );
        dataStream.addSink(kafkaProducer);

        // 将Protobuf数据打印到stdout
        dataStream.addSink(new ProtobufPrintSinkFunction<>());

        env.execute("Protobuf Flink Kafka Example");
    }

    // 自定义Protobuf的反序列化器
    public static class ProtobufDeserializationSchema<T> implements DeserializationSchema<T> {
        private final Class<T> clazz;

        public ProtobufDeserializationSchema(Class<T> clazz) {
            this.clazz = clazz;
        }

        @Override
        public T deserialize(byte[] bytes) {
            // 使用Protobuf的反序列化方法将字节数组转换为Protobuf对象
            return ProtobufUtils.deserialize(bytes, clazz);
        }

        @Override
        public boolean isEndOfStream(T t) {
            return false;
        }

        @Override
        public TypeInformation<T> getProducedType() {
            return TypeInformation.of(clazz);
        }
    }

    // 自定义Protobuf的序列化器
    public static class ProtobufSerializationSchema<T> implements SerializationSchema<T> {
        private final Class<T> clazz;

        public ProtobufSerializationSchema(Class<T> clazz) {
            this.clazz = clazz;
        }

        @Override
        public byte[] serialize(T t) {
            // 使用Protobuf的序列化方法将Protobuf对象转换为字节数组
            return ProtobufUtils.serialize(t);
        }
    }

    // 自定义打印Sink函数
    public static class ProtobufPrintSinkFunction<T> extends RichSinkFunction<T> {
        @Override
        public void invoke(T value) {
            System.out.println(value.toString());
        }
    }
}