Below is the expeption -
Caused by: java.io.IOException: Failed to deserialize an element from the source. If you are using user-defined serialization (Value and Writable types), check the serialization functions. Serializer is org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer@599fcdda at org.apache.flink.streaming.api.functions.source.FromElementsFunction.run(FromElementsFunction.java:121) at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:58) at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.run(SourceStreamTask.java:55) at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:218) at org.apache.flink.runtime.taskmanager.Task.run(Task.java:584) at java.lang.Thread.run(Thread.java:745)
Thanks @Ali Bajwa
@Robert Metzger @Stephan Ewen I was trying to read a Web log which is accessible through a HTTP URL. I used a InputStreamReader to read the content from the HttpConnection. As InputStreamReader is not serializable, I got the above exception.
Should we prepare a custom SourceFunction to handle HTTP Streams or does Flink provide any methods/objects in order to enable users to read HTTP streams?
Right now, there are not methods for reading HTTP streams. But implementing a custom "SourceFunction" (as you suggested) is the recommended appropach. Please let me know if you have further questions.