Add partitionable kafka input operator