[SPARK-53345][SS][TESTS] Use withTempDir for consistent directory across restarts in streaming test

### What changes were proposed in this pull request?

Use withTempDir for consistent directory across restarts in test.

### Why are the changes needed?

Test was flaky due to using an unpredictable temp directory for a streaming query that tests against restarts. This test was introduced in SPARK-53069.

### Does this PR introduce _any_ user-facing change?

No

### How was this patch tested?

Test only change.

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

No

Closes #52088 from dylanwong250/SPARK-53345.

Authored-by: Dylan Wong <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>

Author: Dylan Wong

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

@@ -750,42 +750,47 @@ class TransformWithListStateTTLSuite extends TransformWithStateTTLTest
     withSQLConf(SQLConf.STATE_STORE_PROVIDER_CLASS.key ->
       classOf[RocksDBStateStoreProvider].getName,
       SQLConf.SHUFFLE_PARTITIONS.key -> "1") {
-      val ttlConfig = TTLConfig(ttlDuration = Duration.ofMinutes(10))
-      val inputStream = MemoryStream[String]
-      val result = inputStream.toDS()
-        .groupByKey(x => x)
-        .transformWithState(
-          new MultiStatefulVariableTTLProcessor(ttlConfig),
-          TimeMode.ProcessingTime(),
-          OutputMode.Append())
-      val clock = new StreamManualClock
+      withTempDir { checkpointLocation =>
+        val ttlConfig = TTLConfig(ttlDuration = Duration.ofMinutes(10))
+        val inputStream = MemoryStream[String]
+        val result = inputStream.toDS()
+          .groupByKey(x => x)
+          .transformWithState(
+            new MultiStatefulVariableTTLProcessor(ttlConfig),
+            TimeMode.ProcessingTime(),
+            OutputMode.Append())
+        val clock = new StreamManualClock
 
-      testStream(result)(
-        StartStream(Trigger.ProcessingTime("1 second"), triggerClock = clock),
+        testStream(result)(
+          StartStream(Trigger.ProcessingTime("1 second"), triggerClock = clock,
+            checkpointLocation = checkpointLocation.getAbsolutePath),
 
-        AddData(inputStream, "k1"),
-        AdvanceManualClock(1 * 1000),
-        CheckNewAnswer(("k1", 1)),
-        assertNumStateRows(total = 3, updated = 3),
+          AddData(inputStream, "k1"),
+          AdvanceManualClock(1 * 1000),
+          CheckNewAnswer(("k1", 1)),
+          assertNumStateRows(total = 3, updated = 3),
 
-        StopStream,
-        StartStream(Trigger.ProcessingTime("1 second"), triggerClock = clock),
+          StopStream,
+          StartStream(Trigger.ProcessingTime("1 second"), triggerClock = clock,
+            checkpointLocation = checkpointLocation.getAbsolutePath),
 
-        AddData(inputStream, "k1"),
-        AdvanceManualClock(1 * 1000),
-        CheckNewAnswer(("k1", 2)),
-        assertNumStateRows(total = 4, updated = 3),
+          AddData(inputStream, "k1"),
+          AdvanceManualClock(1 * 1000),
+          CheckNewAnswer(("k1", 2)),
+          assertNumStateRows(total = 4, updated = 3),
 
-        StopStream,
-        StartStream(Trigger.ProcessingTime("1 second"), triggerClock = clock),
+          StopStream,
+          StartStream(Trigger.ProcessingTime("1 second"), triggerClock = clock,
+            checkpointLocation = checkpointLocation.getAbsolutePath),
 
-        AddData(inputStream, "k1"),
-        AdvanceManualClock(1 * 1000),
-        CheckNewAnswer(("k1", 3)),
-        assertNumStateRows(total = 5, updated = 3),
+          AddData(inputStream, "k1"),
+          AdvanceManualClock(1 * 1000),
+          CheckNewAnswer(("k1", 3)),
+          assertNumStateRows(total = 5, updated = 3),
 
-        StopStream
-      )
+          StopStream
+        )
+      }
     }
   }
 }