[BEAM-3371] Fix directories not being staged to classpath issue

Directories are now packaged to .jar files and placed in temporary
location. They can be staged from there later by the runner.

runners/spark/src/main/java/org/apache/beam/runners/spark/SparkRunner.java

@@ -19,6 +19,7 @@
 package org.apache.beam.runners.spark;
 
 import static org.apache.beam.runners.core.construction.PipelineResources.detectClassPathResourcesToStage;
+import static org.apache.beam.runners.core.construction.PipelineResources.prepareFilesForStaging;
 
 import com.google.common.collect.Iterables;
 import java.util.Collection;
@@ -164,6 +165,8 @@ public SparkPipelineResult run(final Pipeline pipeline) {
 
     pipeline.replaceAll(SparkTransformOverrides.getDefaultOverrides(mOptions.isStreaming()));
 
+    prepareFilesToStageForRemoteClusterExecution();
+
     if (mOptions.isStreaming()) {
       CheckpointDir checkpointDir = new CheckpointDir(mOptions.getCheckpointDir());
       SparkRunnerStreamingContextFactory streamingContextFactory =
@@ -272,6 +275,18 @@ private void detectTranslationMode(Pipeline pipeline) {
     }
   }
 
+  /**
+   * Local configurations work in the same JVM and have no problems with improperly formatted files
+   * on classpath (eg. directories with .class files or empty directories). Prepare files for
+   * staging only when using remote cluster.
+   */
+  private void prepareFilesToStageForRemoteClusterExecution() {
+    if (!mOptions.getSparkMaster().matches("local\\[?\\d*\\]?")) {
+      mOptions.setFilesToStage(
+          prepareFilesForStaging(mOptions.getFilesToStage(), mOptions.getTempLocation()));
+    }
+  }
+
   /** Evaluator that update/populate the cache candidates. */
   public static void updateCacheCandidates(
       Pipeline pipeline, SparkPipelineTranslator translator, EvaluationContext evaluationContext) {
@@ -435,7 +450,7 @@ <TransformT extends PTransform<? super PInput, POutput>> void doVisitTransform(
     protected <TransformT extends PTransform<? super PInput, POutput>>
         TransformEvaluator<TransformT> translate(
             TransformHierarchy.Node node, TransformT transform, Class<TransformT> transformClass) {
-      //--- determine if node is bounded/unbounded.
+      // --- determine if node is bounded/unbounded.
       // usually, the input determines if the PCollection to apply the next transformation to
       // is BOUNDED or UNBOUNDED, meaning RDD/DStream.
       Map<TupleTag<?>, PValue> pValues;