Implementation for local samza. still need to create kafka streams