Hello, everyone, val result: DataStream[OutOfLimitSpeedInfo_Par2] = stream1.connect(stream2) .process(new BroadcastProcessFunction[TrafficInfo_Part2, MonitorInfo_Part2, OutOfLimitSpeedInfo_Par2] { override def processElement(value: TrafficInfo_Part2, ctx: BroadcastProcessFunction[TrafficInfo_Part2, MonitorInfo_Part2, OutOfLimitSpeedInfo_Par2]#ReadOnlyContext, out: Collector[OutOfLimitSpeedInfo_Par2]) = { if (value.monitorId == "5815") { println(">>>>>>>TrafficInfo_Part2 :" + value) } val info: MonitorInfo_ Part2 = ctx.getBroadcastState(GlobalConstants_Part2.MONITOR_STATE_DESCRIPTOR).get(value.monitorId) if (info != null) { val limitSpeed = info.limitSpeed val speed = value.speed if (limitSpeed * 1.1 < speed) { out.collect(new OutOfLimitSpeedInfo_Par2(value.car, value.monitorId, value.roadId, speed, limitSpeed, value.actionTime)) } } } override def processBroadcastElement(value: MonitorInfo_Part2, ctx: BroadcastProcessFunction[TrafficInfo_Part2, MonitorInfo_Part2, OutOfLimitSpeedInfo_Par2]#Context, out: Collector[OutOfLimitSpeedInfo_Par2]) = { if (value.monitorId == "5815") { println(">>>>>>>>>>5815: " + value) } ctx.getBroadcastState(GlobalConstants_Part2.MONITOR_STATE_DESCRIPTOR).put(value.monitorId, value) } }) The printed data shows that there is data in the broadcast stream, but through ctx. getBroadcastState (GlobalConstants_Part2. MONITOR_STATE_DESCRIPTOR). get (value. monitorId) This sentence cannot query the corresponding status data set by processBroadcastElement. What is the reason for this? Flink version: Flink 1.9