How can I mark the end of the stream with an end message while using the new KafkaSource?

[LIU Xiao]

With the legacy FlinkKafkaConsumer, overriding the isEndOfStream method of
DeserializationSchema can solve the problem.
But the new KafkaSource ignores the method (never been called), and it
seems the setUnbounded method only accepts offset or time.