Implement the automate partitioned kafka input operator(1:1) #113