[FLINK-14327] Getting "Could not forward element to next operator" error

• Type: Bug
• Status: Closed
• Priority: Minor
• Resolution: Invalid
• Affects Version/s: 1.9.0
• Component/s: API / DataStream
• Labels: auto-deprioritized-major
• External issue URL: https://stackoverflow.com/questions/58233895/how-to-omit-a-null-value-exception-in-flink-kafka
• Assignee: Unassigned
• Reporter: ASK5
• Created: 05/Oct/19 06:13
• Updated: 06/May/21 08:35
• Resolved: 06/May/21 08:35

Description

val TEMPERATURE_THRESHOLD: Double = 50.00

val see: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

val properties = new Properties()
properties.setProperty("zookeeper.connect", "localhost:2181")
properties.setProperty("bootstrap.servers", "localhost:9092")

val src = see.addSource(new FlinkKafkaConsumer010[ObjectNode]("broadcast",
new JSONKeyValueDeserializationSchema(false), properties)).name("kafkaSource")
case class Event(locationID: String, temp: Double)

var data = src.map { v => {
val loc = v.get("locationID").asInstanceOf[String]
val temperature = v.get("temp").asDouble()
(loc, temperature)
}}

data = data
.keyBy(
v => v._1
)

data.print()

see.execute()

--*********---

And I'm getting the following error while consuming json file from Kafka:-

 
Exception in thread "main" org.apache.flink.runtime.client.JobExecutionException: Job execution failed.... at flinkBroadcast1$.main(flinkBroadcast1.scala:59) at flinkBroadcast1.main(flinkBroadcast1.scala)Caused by: java.lang.Exception: org.apache.flink.streaming.runtime.tasks.ExceptionInChainedOperatorException: Could not forward element to next operator...Caused by: org.apache.flink.streaming.runtime.tasks.ExceptionInChainedOperatorException: Could not forward element to next operator...Caused by: java.lang.NullPointerException