[streaming] revise KafkaConnector's 'receive' method