Hello everyone, I am encountering an issue while using my JsonDeserializationSchema. I am attempting to utilize a generic approach for correlating data from two streams, but I'm having trouble understanding why KafkaSource always requires a String type for setValueOnlyDeserializer(new JsonDeserializationSchema<T>(messageType)). My IDE suggests casting to String, but in my case, I am working with data from two POJO classes, UserA and UserB, and I need to send objects of these classes as JSON strings to the Kafka topics. My code snippet : public static <T>KafkaSource<T> create_source(String server,String sourceTopic, Class<T> messageType){ KafkaSource<T> source = (KafkaSource<T>) KafkaSource.<String>_builder_() .setBootstrapServers(server) .setTopics(sourceTopic) .setGroupId("my-group") .setStartingOffsets(OffsetsInitializer._earliest_()) .setValueOnlyDeserializer( new JsonDeserializationSchema<T>(messageType)) .build(); return source; } private static class JsonDeserializationSchema<T> extends AbstractDeserializationSchema<T> { private final ObjectMapper objectMapper = new ObjectMapper(); private final Class<T> targetType; public JsonDeserializationSchema(Class<T> targetType) { this.targetType = targetType; } @Override public T deserialize(byte[] message) throws IOException { return objectMapper.readValue(message, targetType); } @Override public TypeInformation<T> getProducedType() { return TypeInformation._of_(targetType); } } But still maybe is so complicated and using Avro and schema registry is the right approach for serialization...