Add partitionable kafka input operator #113