diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSink.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSink.scala index 638da08d0fd9..d561ee1ef730 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSink.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSink.scala @@ -49,8 +49,8 @@ object FileStreamSink extends Logging { path match { case Seq(singlePath) => - val hdfsPath = new Path(singlePath) try { + val hdfsPath = new Path(singlePath) val fs = hdfsPath.getFileSystem(hadoopConf) if (fs.getFileStatus(hdfsPath).isDirectory) { val metadataPath = getMetadataLogPath(fs, hdfsPath, sqlConf) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/FileStreamSinkSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/FileStreamSinkSuite.scala index 04193d5189ae..2e80588fb282 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/FileStreamSinkSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/FileStreamSinkSuite.scala @@ -651,6 +651,19 @@ abstract class FileStreamSinkSuite extends StreamTest { } } } + + test("SPARK-48991: Move path initialization into try-catch block") { + val logAppender = new LogAppender("Assume no metadata directory.") + Seq(null, "", "file:tmp").foreach { path => + withLogAppender(logAppender) { + assert(!FileStreamSink.hasMetadata(Seq(path), spark.sessionState.newHadoopConf(), conf)) + } + + assert(logAppender.loggingEvents.map(_.getMessage.getFormattedMessage).contains( + "Assume no metadata directory. Error while looking for metadata directory in the path:" + + s" $path.")) + } + } } object PendingCommitFilesTrackingManifestFileCommitProtocol {