Fix temp

scio-core/src/main/scala/com/spotify/scio/ScioContext.scala

@@ -20,6 +20,7 @@ package com.spotify.scio
 import java.beans.Introspector
 import java.io.File
 import java.net.{URI, URLClassLoader}
+import java.nio.file.Files
 import java.util.concurrent.TimeUnit
 
 import com.google.api.services.bigquery.model.TableReference
@@ -157,6 +158,13 @@ class ScioContext private[scio] (val options: PipelineOptions,
       Try(optionsAs[DataflowPipelineWorkerPoolOptions])
         .foreach(_.setFilesToStage(getFilesToStage(artifacts).asJava))
       _pipeline = if (testId.isEmpty) {
+        // if in local runner, temp location may be needed, but is not currently required by
+        // the runner, which may end up with NPE. If not set but user generate new temp dir
+        if (ScioUtil.isLocalRunner(options) && options.getTempLocation == null) {
+          val tmpDir = Files.createTempDirectory("scio-temp-")
+          logger.debug(s"New temp directory at $tmpDir")
+          options.setTempLocation(tmpDir.toString)
+        }
         Pipeline.create(options)
       } else {
         val tp = TestPipeline.create()

scio-core/src/main/scala/com/spotify/scio/util/ScioUtil.scala

@@ -20,6 +20,9 @@ package com.spotify.scio.util
 import java.net.URI
 
 import com.google.cloud.dataflow.sdk.coders.{Coder, CoderRegistry}
+import com.google.cloud.dataflow.sdk.options.PipelineOptions
+import com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner
+import com.google.cloud.dataflow.sdk.runners.inprocess.InProcessPipelineRunner
 
 import scala.reflect.ClassTag
 
@@ -40,4 +43,13 @@ private[scio] object ScioUtil {
 
     coderRegistry.getScalaCoder[T]
   }
+
+  def isLocalRunner(options: PipelineOptions): Boolean = {
+    val runner = options.getRunner
+
+    require(runner != null, "Pipeline runner not set!")
+
+    runner.isAssignableFrom(classOf[DirectPipelineRunner]) ||
+      runner.isAssignableFrom(classOf[InProcessPipelineRunner])
+  }
 }

scio-test/src/test/scala/com/spotify/scio/ScioContextTest.scala

@@ -17,10 +17,14 @@
 
 package com.spotify.scio
 
+import com.google.cloud.dataflow.sdk.options.{DataflowPipelineOptions, PipelineOptionsFactory}
+import com.google.cloud.dataflow.sdk.runners.{DataflowPipelineRunner, DirectPipelineRunner}
+import com.google.cloud.dataflow.sdk.runners.inprocess.InProcessPipelineRunner
 import com.google.cloud.dataflow.sdk.testing.DataflowAssert
 import com.google.cloud.dataflow.sdk.transforms.Create
 import com.google.common.collect.Lists
 import com.spotify.scio.testing.PipelineSpec
+import org.apache.commons.lang.exception.ExceptionUtils
 
 import scala.collection.JavaConverters._
 
@@ -39,4 +43,42 @@ class ScioContextTest extends PipelineSpec {
     pipeline.run()
   }
 
+  it should "have temp directory for default runner" in {
+    val pipeline = ScioContext().pipeline
+    pipeline.getOptions.getTempLocation should not be null
+  }
+
+  it should "have temp directory for default InProcessPipelineRunner" in {
+    val opts = PipelineOptionsFactory.create()
+    opts.setRunner(classOf[InProcessPipelineRunner])
+    val pipeline = ScioContext(opts).pipeline
+    pipeline.getOptions.getTempLocation should not be null
+  }
+
+  it should "have temp directory for default DirectPipelineRunner" in {
+    val opts = PipelineOptionsFactory.create()
+    opts.setRunner(classOf[DirectPipelineRunner])
+    val pipeline = ScioContext(opts).pipeline
+    pipeline.getOptions.getTempLocation should not be null
+  }
+
+  it should "use user specified temp directory" in {
+    val expected = "/expected"
+    val opts = PipelineOptionsFactory.create()
+    opts.setTempLocation(expected)
+    val pipeline = ScioContext(opts).pipeline
+    pipeline.getOptions.getTempLocation shouldEqual expected
+  }
+
+  it should "fail without temp/staging dir for DataflowPipelineRunner " in {
+    val opts = PipelineOptionsFactory.create()
+    opts.setRunner(classOf[DataflowPipelineRunner])
+    val dfOpts = opts.as(classOf[DataflowPipelineOptions])
+    dfOpts.setProject("foobar")
+    val sc = ScioContext(dfOpts)
+    val e = intercept[RuntimeException] { sc.pipeline }
+    ExceptionUtils.getFullStackTrace(e) should
+      include ("at least one of tempLocation or stagingLocation must be set.")
+  }
+
 }