如何初始化Flink广播状态？

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class BroadcastStateInitializationExample {

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

        // 创建一个MapStateDescriptor，用于定义广播状态的名称和类型
        MapStateDescriptor<String, Integer> broadcastStateDescriptor =
                new MapStateDescriptor<>("broadcast-state", String.class, Integer.class);

        // 创建一个数据流作为广播流
        BroadcastStream<Tuple2<String, Integer>> broadcastStream = env
                .fromElements(Tuple2.of("key1", 1), Tuple2.of("key2", 2))
                .broadcast(broadcastStateDescriptor);

        // 创建一个主数据流
        DataStream<Tuple2<String, Integer>> mainStream = env
                .fromElements(Tuple2.of("key1", 10), Tuple2.of("key2", 20));

        // 将广播流与主数据流连接起来
        ConnectedStreams<Tuple2<String, Integer>, Tuple2<String, Integer>> connectedStreams =
                mainStream.connect(broadcastStream);

        // 处理广播流和主数据流的元素
        connectedStreams.process(new BroadcastProcessFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, String>() {
            @Override
            public void processElement(Tuple2<String, Integer> value, ReadOnlyContext ctx, Collector<String> out) throws Exception {
                // 处理主数据流的元素
                // ...

                // 从广播状态中获取初始值
                Integer initialValue = ctx.getBroadcastState(broadcastStateDescriptor).get(value.f0);

                // 使用初始值进行处理
                // ...
            }

            @Override
            public void processBroadcastElement(Tuple2<String, Integer> value, Context ctx, Collector<String> out) throws Exception {
                // 处理广播流的元素
                // ...

                // 将初始值写入广播状态
                BroadcastState<String, Integer> broadcastState = ctx.getBroadcastState(broadcastStateDescriptor);
                broadcastState.put(value.f0, value.f1);
            }
        });

        env.execute("Broadcast State Initialization Example");
    }
}