Hi everyone, I have this json file which comes from KafkaSource, how can I parse nestej json data. I tried below codes but they didn't work. I have before key which is before data and after is only changed columns. I want to map after data to before data (for example before.STATUS == after.STATUS) and obtain new POJO class. How can parse nested json which comes from kafka source and maps changes ?

{
"table":"DBNAME.MYTABLE",
"op_type":"U",
"op_ts":"2023-10-21 08:13:01.826497",
"before":{
"ID":"123",
"CREATED":"2023-10-17 15:28:38"
"LAST_UPD":"2023-10-21 11:10:28",
"ACT_CLOSE_DT":"2023-10-21 11:10:28",
"DB_LAST_UPD":"2023-10-21 11:10:28",
"STATUS":"Closed",
"SEND_NUM":"123456",
"UPDATE":null
},
"after":{
"ID":"123",
"LAST_UPD":"2023-10-21 11:12:28",
"ACT_CLOSE_DT":"2023-10-21 11:12:28",
"DB_LAST_UPD":"2023-10-21 11:12:28",
"STATUS":"Closed" 
}}

KafkaSource<String> kfkSource = KafkaSource.<String>builder()
		.setBootstrapServers("localhost:9092")
		.setTopics("mykafkatopic")
		.setStartingOffsets(OffsetsInitializer.latest())
		.setValueOnlyDeserializer(new SimpleStringSchema())
		.build();

DataStream<String> kafkaStream = env.fromSource(kfkSource, WatermarkStrategy.noWatermarks(), "KafkaExactlyOnce");

ObjectMapper objectMapper = new ObjectMapper();

DataStream<MyRequestData> parsedStream = kafkaStream
		.map(new MapFunction<String, MyRequestData>() {
			@Override
			public MyRequestData map(String jsonString) throws Exception {
				JsonNode jsonNode = objectMapper.readTree(jsonString);
				return objectMapper.treeToValue(jsonNode, MyRequestData.class);
			}
		});

parsedStream.print();

What’s the error you are facing ?

This the error what I get