Source API: to poll the internal producer regularly

[daniil-quix]

Currently, Sources poll the internal Producer only while producing a message.
It works fine if the Source produces data steadily without gaps.

But if the source doesn't receive any data, the Producer will only be polled once new data comes in.
It may lead to delayed errors if some previous messages fail to be delivered to Kafka.

We need to investigate how to poll the internal producer regularly or how to let Source implementations do that.