About Kafka08Fetcher and Kafka010Fetcher

In Kafka08Fetcher, it use Map to manage multi-threads. But I notice in Kafka09Fetcher or Kafka010Fetcher, it's gone. So how Kafka09Fetcher implements multi-threads read partitions from kafka?

How to stop FlinkKafkaConsumer and make job finished?

I would like to stop FlinkKafkaConsumer consuming data from kafka manually. But I find it won't be close when I invoke "cancel()" method. What I am trying to do is add an EOF symbol meaning the end of my kafka data, and when the FlatMap operator read the symbol it will invoke FlinkKafkaConsumer "ca