Removing sleep in the `KafkaSourceTest`

[sanha]

At now, in KafkaSourceTest, there are a sleep code between the close of KafkaSource and the collpase of LocalKafkaBroker. This is because that KafkaDataGenerator waits until it's pollTimeout before it is closed. If the test doesn't sleep and just shut down the broker, the generator produces a lots of Exception. Moreover, the KafkaConsumer does not handle these exception properly, so we can not catch them. However, sleep code in test is not good. Therefore, we need to find another way to avoid these exceptions.