KafkaIO SDF: Fetch main data and the end position for each topic-partition tuple in a background thread, reusing consumers.

KafkaIO SDF: Fetch main data and the end position for each topic-partition tuple in a background thread, reusing consumers. #9040