clean AbstractKafkaInputOperator #73