[SPARK-53961][SQL][TESTS] Fix FileStreamSinkSuite flakiness by using walkFileTree instead of walk

[dongjoon-hyun]

What changes were proposed in this pull request?

This PR aims to fix FileStreamSinkSuite flakiness by using walkFileTree instead of walk.

Why are the changes needed?

Files.walk is flaky like the following when the directory has a race condition. walkFileTree has more robust error handling.

spark/sql/core/src/test/scala/org/apache/spark/sql/streaming/FileStreamSinkSuite.scala

Lines 543 to 547 in 2bb73fb

	// there would be possible to have race condition:
	// - some tasks complete while abortJob is being called
	// we can't delete complete files for these tasks (it's OK since this is a best effort)
	assert(outputFileNames.intersect(trackingFileNames).isEmpty,
	"abortJob should clean up files reported as successful.")

[info] - cleanup complete but invalid output for aborted job *** FAILED *** (438 milliseconds)
     [info]   java.io.UncheckedIOException: java.nio.file.NoSuchFileException: ***/spark-4c7ad10b-5848-45d7-ba43-dae4020ad011/output @#output/part-00007-e582f3e3-87e3-40fa-8269-7fac9b545775-c000.snappy.parquet
     [info]   at java.base/java.nio.file.FileTreeIterator.fetchNextIfNeeded(FileTreeIterator.java:87)
     [info]   at java.base/java.nio.file.FileTreeIterator.hasNext(FileTreeIterator.java:103)
     [info]   at java.base/java.util.Spliterators$IteratorSpliterator.tryAdvance(Spliterators.java:1855)
     [info]   at java.base/java.util.stream.StreamSpliterators$WrappingSpliterator.lambda$initPartialTraversalState$0(StreamSpliterators.java:292)
     [info]   at java.base/java.util.stream.StreamSpliterators$AbstractWrappingSpliterator.fillBuffer(StreamSpliterators.java:206)
     [info]   at java.base/java.util.stream.StreamSpliterators$AbstractWrappingSpliterator.doAdvance(StreamSpliterators.java:169)
     [info]   at java.base/java.util.stream.StreamSpliterators$WrappingSpliterator.tryAdvance(StreamSpliterators.java:298)
     [info]   at java.base/java.util.Spliterators$1Adapter.hasNext(Spliterators.java:681)
     [info]   at scala.collection.convert.JavaCollectionWrappers$JIteratorWrapper.hasNext(JavaCollectionWrappers.scala:46)
     [info]   at scala.collection.Iterator$$anon$6.hasNext(Iterator.scala:480)
     [info]   at scala.collection.Iterator$$anon$9.hasNext(Iterator.scala:583)
     [info]   at scala.collection.mutable.Growable.addAll(Growable.scala:61)
     [info]   at scala.collection.mutable.Growable.addAll$(Growable.scala:57)
     [info]   at scala.collection.immutable.SetBuilderImpl.addAll(Set.scala:405)
     [info]   at scala.collection.immutable.Set$.from(Set.scala:362)
     [info]   at scala.collection.IterableOnceOps.toSet(IterableOnce.scala:1469)
     [info]   at scala.collection.IterableOnceOps.toSet$(IterableOnce.scala:1469)
     [info]   at scala.collection.AbstractIterator.toSet(Iterator.scala:1306)
     [info]   at org.apache.spark.sql.streaming.FileStreamSinkSuite.$anonfun$new$52(FileStreamSinkSuite.scala:537)

Does this PR introduce any user-facing change?

No, this is a test case change.

How was this patch tested?

Pass the CIs.

Was this patch authored or co-authored using generative AI tooling?

No.

[dongjoon-hyun]

It's okay to ignore the deleted files because those are supposed to be cleaned up in a success case.

[dongjoon-hyun]

The updated test passed.

• https://github.com/dongjoon-hyun/spark/actions/runs/18664914877/job/53221841383

FileStreamSinkV1Suite
...
[info] - cleanup complete but invalid output for aborted job (260 milliseconds)
...
FileStreamSinkV2Suite
...
[info] - cleanup complete but invalid output for aborted job (269 milliseconds)
...

[dongjoon-hyun]

Thank you, @HyukjinKwon . Merged to master/4.0/3.5.