introduced AsyncSemaphore and FuturePool in KafkaStore.