I proposed adding 'withEndReadTime()' in jira about better support for bounded reads (BEAM-2185 <https://issues.apache.org/jira/browse/BEAM-2185>). startReadTime() already exists.
I think it is fairly straight forward to add. At this time it is fine to expect Kafka server side timestamps. We haven't given much attention for batch use cases of KafkaIO. I think it we should improve support for batch pipelines. ​Raghu.
