Merge branch 'master' into test

nebula-exchange/src/main/scala/com/vesoft/nebula/exchange/processor/EdgeProcessor.scala

@@ -156,6 +156,8 @@ class EdgeProcessor(data: DataFrame,
 
  val ranking: Long = if (edgeConfig.rankingField.isDefined) {
  val rankIndex = row.schema.fieldIndex(edgeConfig.rankingField.get)
+ assert(rankIndex >= 0 && !row.isNullAt(rankIndex),
+ s"rank must exist and cannot be null, your row data is $row")
  row.get(rankIndex).toString.toLong
  } else {
  0

nebula-exchange/src/main/scala/com/vesoft/nebula/exchange/utils/HDFSUtils.scala

@@ -22,36 +22,23 @@ object HDFSUtils {
  conf.set("fs.default.name", namenode)
  conf.set("fs.defaultFS", namenode)
  }
- conf.set("fs.hdfs.impl.disable.cache", "true")
  FileSystem.get(conf)
  }
 
  def list(path: String): List[String] = {
  val system = getFileSystem()
- try {
- system.listStatus(new Path(path)).map(_.getPath.getName).toList
- } finally {
- system.close()
- }
+ system.listStatus(new Path(path)).map(_.getPath.getName).toList
  }
 
  def exists(path: String): Boolean = {
  val system = getFileSystem()
- try {
- system.exists(new Path(path))
- } finally {
- system.close()
- }
+ system.exists(new Path(path))
  }
 
  def getContent(path: String): String = {
  val system = getFileSystem()
  val inputStream = system.open(new Path(path))
- try {
- Source.fromInputStream(inputStream).mkString
- } finally {
- system.close()
- }
+ Source.fromInputStream(inputStream).mkString
  }
 
  def saveContent(path: String,
@@ -79,10 +66,6 @@ object HDFSUtils {
  e)
  }
  val system = getFileSystem(namenode)
- try {
- system.copyFromLocalFile(new Path(localPath), new Path(remotePath))
- } finally {
- system.close()
- }
+ system.copyFromLocalFile(new Path(localPath), new Path(remotePath))
  }
 }