BAEL-1994: Moved the code to libraries module

libraries/src/main/java/com/baeldung/flink/schema/InputMessageDeserializationSchema.java

@@ -0,0 +1,33 @@
+package com.baeldung.schema;
+
+import com.baeldung.model.InputMessage;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+
+import java.io.IOException;
+
+public class InputMessageDeserializationSchema implements
+      DeserializationSchema<InputMessage> {
+
+    ObjectMapper objectMapper;
+
+    @Override
+    public InputMessage deserialize(byte[] bytes) throws IOException {
+        if(objectMapper == null) {
+            objectMapper = new ObjectMapper().registerModule(new JavaTimeModule());
+        }
+        return objectMapper.readValue(bytes, InputMessage.class);
+    }
+
+    @Override
+    public boolean isEndOfStream(InputMessage inputMessage) {
+        return false;
+    }
+
+    @Override
+    public TypeInformation<InputMessage> getProducedType() {
+        return TypeInformation.of(InputMessage.class);
+    }
+}