[GitHub] spark pull request #22952: [SPARK-20568][SS] Provide option to clean up comp...

[gaborgsomogyi <gi...@git.apache.org>]

Github user gaborgsomogyi commented on a diff in the pull request:

    https://github.com/apache/spark/pull/22952#discussion_r235314493
  
    --- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala ---
    @@ -257,16 +258,64 @@ class FileStreamSource(
        * equal to `end` and will only request offsets greater than `end` in the future.
        */
       override def commit(end: Offset): Unit = {
    -    // No-op for now; FileStreamSource currently garbage-collects files based on timestamp
    -    // and the value of the maxFileAge parameter.
    +    def move(entry: FileEntry, baseArchiveDirPath: String): Unit = {
    +      val curPath = new Path(entry.path)
    +      val curPathUri = curPath.toUri
    +
    +      val newPath = new Path(baseArchiveDirPath + curPathUri.getPath)
    +      if (!fs.exists(newPath.getParent)) {
    --- End diff --
    
    These fs operation can also throw exception. Why not covered these as well with try?


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org