title: Issuing spark-submit on command line completes tasks but never returns prompt
body: <p>I'm using spark (Java) to read and write to DB. I'm using Spark's built-in cluster manager. The application is bundles as fat jar and run via spark-submit command:</p>
  
  <pre><code>"./spark-submit --class com.tte.site.sector.daily.main.Driver --master spark://ip-xxx-xx-xx-xx:7077 --deploy-mode client /home/ec2-user/jars-dir/site-sector-daily-1.0-jar-with-dependencies.jar
  </code></pre>
  
  <p>The tasks run fine, there are no exceptions in the application log or when spark-submit is issued on the command line. Below is the entire stdout output from running spark-submit: </p>
  
  <blockquote>
    <p>./spark-submit --class com.tte.site.sector.daily.main.Driver --master spark://ip-xxx-xx-xx-xx:7077 --deploy-mode client /home/ec2-user/jars-dir/site->> sector-daily-1.0-jar-with-dependencies.jar</p>
    
    <p>15/12/17 06:42:06 INFO TaskSetManager: Finished task 0.0 in stage 0.0 (TID 0) in 861449 ms on xxx.xx.8.62 (1/2)
    15/12/17 06:46:01 INFO DAGScheduler: ResultStage 0 (foreach at Driver.java:143) finished in 1095.509 s
    15/12/17 06:46:01 INFO TaskSetManager: Finished task 1.0 in stage 0.0 (TID 1) in 1095438 ms on xxx.xx.17.222 (2/2)
    15/12/17 06:46:01 INFO TaskSchedulerImpl: Removed TaskSet 0.0, whose tasks have all completed, from pool 
    15/12/17 06:46:01 INFO DAGScheduler: Job 0 finished: foreach at Driver.java:143, took 1095.7685> 15 s</p>
  </blockquote>
  
  <p>It seems that application logic has completed as far as I can tell since appropriate DB updates have been executed correctly, but Spark UI shows that it's still running and of course spark-submit command does not return prompt back, so something is still holding on to resources. I've tried using JavaSparkContext.close(), although it helps with Spark UI showing the job as finished but spark-submit command still does not return. What is it that I'm doing wrong here?</p>
  
  <p><a href="http://i.stack.imgur.com/Wawza.png" rel="nofollow">spark ui main page</a></p>
  
  <p>Here is the code:</p>
  
  <pre><code>        public class Driver {
  
              private static final Logger logger = LoggerFactory.getLogger(Driver.class);
  
              public static void main(String[] args) {
  
                  Cluster cluster = Cluster.builder().addContactPoint("xxx.xx.xx.xx").build();// aws local
                  Session dbSession = cluster.connect("syringa");
                  SparkConf conf = new SparkConf()
                          .setAppName("sector_site_hourly_daily_job")
                          .setMaster("spark://ip-172-31-29-81:7077");
                  JavaSparkContext sc = new JavaSparkContext(conf);
  
  
                      // share with workers
                      final Broadcast&lt;List&lt;KpiEntity&gt;&gt; kpiFormulaEntityBroadcastVar = sc.broadcast(kpiFormulaEntity);
                      final Broadcast&lt;Set&lt;String&gt;&gt; inputCountersBroadcastVar = sc.broadcast(inputCountersInKPIFormulas);
                      final Broadcast&lt;Map&lt;Integer, List&lt;Date&gt;&gt;&gt; fromTohoursBroadcastVar = sc.broadcast(generateFromToHour2());
                      final Broadcast&lt;Map&lt;Integer, List&lt;DateTime&gt;&gt;&gt; fromTohoursSelectBroadcastVar = sc.broadcast(generateFromToHours());
                      final Broadcast&lt;Map&lt;Integer, List&lt;DateTime&gt;&gt;&gt; noneUTCDatesBroadcastVar = sc.broadcast(generateFromToHoursForSelecting());
  
                      // Alternate approach to cassandraTable - Sites-Sectors RDD
                      ResultSet siteSectorQueryResult = dbSession.execute("select * from kpi.site_sectors_zone_area"); 
                      List&lt;Row&gt; rows = siteSectorQueryResult.all();
                      List&lt;SiteSectorsEntity&gt; siteSectorsEntities = new ArrayList&lt;&gt;();
                      for(Row row: rows) {
                          SiteSectorsEntity siteSectorsEntity = new SiteSectorsEntity();
                          siteSectorsEntity.setSiteName(row.getString("site_name"));
                          siteSectorsEntity.setArea(row.getString("area"));
                          siteSectorsEntity.setLatitude(row.getString("latitude"));
                          siteSectorsEntity.setLongitude(row.getString("longitude"));
                          siteSectorsEntity.setSectorAzimuth(row.getMap("sector_azimuth", String.class, String.class));
                          siteSectorsEntity.setSectors(row.getList("sectors", String.class));
                          siteSectorsEntity.setZone(row.getString("zone"));
                          siteSectorsEntities.add(siteSectorsEntity);
                      }
                      logger.info("*** Number of SiteSectorEntities {}", siteSectorsEntities.size());
  
                      JavaRDD&lt;SiteSectorsEntity&gt; siteSectorsEntityRDD = sc.parallelize(siteSectorsEntities);
  
  
                      doWork(siteSectorsEntityRDD, kpiFormulaEntityBroadcastVar, inputCountersBroadcastVar, 
                              fromTohoursBroadcastVar, fromTohoursSelectBroadcastVar, noneUTCDatesBroadcastVar);
                  } finally {
                       dbSession.close();
                  }
              }
  
  
              public static void doWork(JavaRDD&lt;SiteSectorsEntity&gt; siteSectorsEntityRDD, Broadcast&lt;List&lt;KpiEntity&gt;&gt; kpiFormulaEntityBroadcastVar,
                      Broadcast&lt;Set&lt;String&gt;&gt; inputCountersBroadcastVar, Broadcast&lt;Map&lt;Integer, List&lt;Date&gt;&gt;&gt; fromTohoursBroadcastVar,
                      Broadcast&lt;Map&lt;Integer, List&lt;DateTime&gt;&gt;&gt; fromTohoursSelectBroadcastVar,
                      Broadcast&lt;Map&lt;Integer, List&lt;DateTime&gt;&gt;&gt; noneUTCBroadvaseVar)  {
  
                  // Distribute to workers
                  siteSectorsEntityRDD.foreach(new VoidFunction&lt;SiteSectorsEntity&gt;() {
                      private static final long serialVersionUID = 5219326359281542043L;
  
                      public void call(SiteSectorsEntity siteSectorsEntity) throws Exception { // remove throws
                          Cluster cluster = Cluster.builder().addContactPoint("xxx.xx.xx.xx").build();//aws local
                          Session dbSession = cluster.connect("syringa");
                          Map&lt;Integer, List&lt;Date&gt;&gt; fromTohours = fromTohoursBroadcastVar.value();
                          Map&lt;Integer, List&lt;DateTime&gt;&gt; fromTohoursSelect = fromTohoursSelectBroadcastVar.value();
                          Map&lt;Integer, List&lt;DateTime&gt;&gt; noneUTCDates = noneUTCBroadvaseVar.value();
                          DateTime now = new DateTime();
  
                          //omitting long line of code here...
  
                          logger.info("site-daily-counter-aggregation: compeleted for: {}", siteName);
  
                          logger.info("site_daily_kpi: starting daily kpi generation for {}", siteName);
                          String siteDailyAggregateValue;
                          Map&lt;String, String&gt; mapOfSiteToDailyKpiInputCountersTotal = new HashMap&lt;&gt;();
                          for(KpiEntity kpiEntity: kpiEntityList) {
                              for(String kpiInputCounter : kpiEntity.getFormulaCounterNames()) {
                                  String Zfrom = fromTimestamp.toString();
                                  String[] noZfrom = Zfrom.split("Z");
                                  String from = noZfrom[0]+"-0800";
                                  String Zto = toTimestamp.toString();
                                  String[] noZto = Zto.split("Z");
                                  String to  = noZto[0]+"-0800";
                                  List&lt;Row&gt; siteDailyAggregate = 
                                          DBUtil.selectSiteDailyCounterAggregate(siteName, from, to, kpiInputCounter, dbSession);
                                  siteDailyAggregateValue = siteDailyAggregate.get(0).getString("counter_agg_value");
                                  mapOfSiteToDailyKpiInputCountersTotal.put(kpiInputCounter, siteDailyAggregateValue);
                              }
                              String kpiFormula = kpiEntity.getKpiFormula();
                              for(String counter:kpiEntity.getFormulaCounterNames()) {
                                  kpiFormula = kpiFormula.replaceAll("\\b"+counter+"\\b", mapOfSectorsToKpiInputCountersTotal.get(counter));
                              }
                              System.out.println("site_daily_kpi: KPI FORMULA TO BE EVAL'd :: "+kpiFormula +" for hour::" +fromTimestamp);
                              // create a script engine manager
                              ScriptEngineManager factory = new ScriptEngineManager();
                              // create a Nashorn script engine
                              ScriptEngine engine = factory.getEngineByName("nashorn");
                              // evaluate KPI formula as a JavaScript statement
                              try {
                                  String red = kpiEntity.getKpiStatusRed();
                                  String green = kpiEntity.getKpiStatusGreen();
                                  String yellow = kpiEntity.getKpiStatusYellow();
                                  Map&lt;String, String&gt; thresholdMap = new HashMap&lt;&gt;();
                                  thresholdMap.put("red", red);
                                  thresholdMap.put("yellow", yellow);
                                  thresholdMap.put("green", green);
                                  String[] yellowRange = yellow.split("-"); // assuming that only yellow carries a multi-value (range)
  
                                  BigDecimal dailyKpiValue = evaluateExpression(kpiFormula, engine);
                                  if(compareIfLessThanOneAndGreaterThanZero(dailyKpiValue)) {
                                      dailyKpiValue = dailyKpiValue.setScale(1, RoundingMode.UP);
                                  } else {
                                      dailyKpiValue = dailyKpiValue.setScale(0, RoundingMode.DOWN);
                                  }
                                  System.out.println("site_hourly_kpi: site Hourly "+kpiEntity.getKpiName()+"="+dailyKpiValue.setScale(0, RoundingMode.DOWN) +" "+kpiEntity.getMeasurementUnit());
  
                                  String kpiStatusColor = determineKpiStatusColor(dailyKpiValue, red, green, yellowRange, engine);
  
                                  // populate sector counter aggregation table for a counter
                                  Insert insert = QueryBuilder.insertInto("kpi", "site_daily_kpi")
                                          .value("site_name", siteName)
                                          .value("area", siteSectorsEntity.getArea())
                                          .value("id", UUID.randomUUID())
                                          .value("kpi_name", kpiEntity.getKpiName())
                                          .value("kpi_status", kpiStatusColor)
                                          .value("kpi_value", dailyKpiValue.toString())
                                          .value("measurement", kpiEntity.getMeasurementUnit())
                                          .value("thresholds", thresholdMap)
                                          .value("time_stamp", isoFormat.parse(fromTimestamp.toString()))
                                          .value("category", kpiEntity.getCategory())
                                          .value("zone", siteSectorsEntity.getZone())
                                          ;
                                  ResultSet results = dbSession.execute(insert);
  
                                   } catch (NumberFormatException nfe) {
                                       logger.info("site_daily_kpi: site hourly "+kpiEntity.getKpiName()+"="+0+" "+kpiEntity.getMeasurementUnit());
                                   } catch (ClassCastException cce) {
                                       logger.info("site_daily_kpi: ClassCastException site hourly: "+ siteName );
                                   }
                      }// ends site hourly kpi gen.
                      logger.info("site_daily_kpi: completed site daily kpi genertion for: {}",siteName);
  
              } finally {
                  try {
                      logger.info("in finally - closing DB session.");
                      dbSession.close();
                  } catch (Exception e) {
                      logger.error("Error during db session close", e.getMessage());
                  }
              }
              }
           });
          }
          }
  </code></pre>
  
  <p>Adding jstack output after tasks completed:</p>
  
  <pre><code>   jstack 14073
      2015-12-18 06:30:46
      Full thread dump Java HotSpot(TM) 64-Bit Server VM (25.65-b01   mixed mode):
  
      "ForkJoinPool-3-worker-3" #73 daemon prio=5 os_prio=0 tid=0x00007f59801c6000 nid=0x37fd waiting on condition [0x00007f59741f8000]
         java.lang.Thread.State: WAITING (parking)
          at sun.misc.Unsafe.park(Native Method)
          - parking to wait for  &lt;0x00000000cead31b8&gt; (a scala.concurrent.forkjoin.ForkJoinPool)
          at scala.concurrent.forkjoin.ForkJoinPool.scan(ForkJoinPool.java:2075)
          at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
          at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
  
      "DestroyJavaVM" #72 prio=5 os_prio=0 tid=0x00007f59a8008800 nid=0x3718 waiting on condition [0x0000000000000000]
         java.lang.Thread.State: RUNNABLE
  
      "Attach Listener" #69 daemon prio=9 os_prio=0 tid=0x00007f5980220800 nid=0x376e waiting on condition [0x0000000000000000]
         java.lang.Thread.State: RUNNABLE
  
      "cluster1-nio-worker-1" #10 prio=5 os_prio=0 tid=0x00007f59a85a9000 nid=0x3727 runnable [0x00007f5984625000]
         java.lang.Thread.State: RUNNABLE
          at io.netty.channel.epoll.Native.epollWait0(Native Method)
          at io.netty.channel.epoll.Native.epollWait(Native.java:153)
          at io.netty.channel.epoll.EpollEventLoop.epollWait(EpollEventLoop.java:184)
          at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:209)
          at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:111)
          at java.lang.Thread.run(Thread.java:745)
  
      "threadDeathWatcher-2-1" #15 daemon prio=1 os_prio=0 tid=0x00007f59780ca000 nid=0x3726 waiting on condition [0x00007f5984926000]
         java.lang.Thread.State: TIMED_WAITING (sleeping)
          at java.lang.Thread.sleep(Native Method)
          at io.netty.util.ThreadDeathWatcher$Watcher.run(ThreadDeathWatcher.java:137)
          at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:137)
          at java.lang.Thread.run(Thread.java:745)
  
      "cluster1-timeouter-0" #11 prio=5 os_prio=0 tid=0x00007f597806b800 nid=0x3725 waiting on condition [0x00007f5984a27000]
         java.lang.Thread.State: TIMED_WAITING (sleeping)
          at java.lang.Thread.sleep(Native Method)
          at io.netty.util.HashedWheelTimer$Worker.waitForNextTick(HashedWheelTimer.java:461)
          at io.netty.util.HashedWheelTimer$Worker.run(HashedWheelTimer.java:360)
          at java.lang.Thread.run(Thread.java:745)
  
      "cluster1-nio-worker-0" #9 prio=5 os_prio=0 tid=0x00007f59a8584000 nid=0x3724 runnable [0x00007f5984b28000]
         java.lang.Thread.State: RUNNABLE
          at io.netty.channel.epoll.Native.epollWait0(Native Method)
          at io.netty.channel.epoll.Native.epollWait(Native.java:153)
          at io.netty.channel.epoll.EpollEventLoop.epollWait(EpollEventLoop.java:184)
          at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:209)
          at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:111)
          at java.lang.Thread.run(Thread.java:745)
  
      "cluster1-scheduled-task-worker-0" #13 prio=5 os_prio=0 tid=0x00007f59a854e800 nid=0x3722 waiting on condition [0x00007f59ac103000]
         java.lang.Thread.State: TIMED_WAITING (parking)
          at sun.misc.Unsafe.park(Native Method)
          - parking to wait for  &lt;0x00000000d59c70b8&gt; (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
          at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
          at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
          at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1093)
          at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
          at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
          at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
          at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
          at java.lang.Thread.run(Thread.java:745)
  
      "cluster1-connection-reaper-0" #8 prio=5 os_prio=0 tid=0x00007f59a8483800 nid=0x3721 waiting on condition [0x00007f59ac408000]
         java.lang.Thread.State: TIMED_WAITING (parking)
          at sun.misc.Unsafe.park(Native Method)
          - parking to wait for  &lt;0x00000000d59c8f20&gt; (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
          at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
          at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
          at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1093)
          at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
          at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
          at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
          at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
          at java.lang.Thread.run(Thread.java:745)
  
      "Service Thread" #7 daemon prio=9 os_prio=0 tid=0x00007f59a80bc800 nid=0x371f runnable [0x0000000000000000]
         java.lang.Thread.State: RUNNABLE
  
      "C1 CompilerThread1" #6 daemon prio=9 os_prio=0 tid=0x00007f59a80af800 nid=0x371e waiting on condition [0x0000000000000000]
         java.lang.Thread.State: RUNNABLE
  
      "C2 CompilerThread0" #5 daemon prio=9 os_prio=0 tid=0x00007f59a80ad800 nid=0x371d waiting on condition [0x0000000000000000]
         java.lang.Thread.State: RUNNABLE
  
      "Signal Dispatcher" #4 daemon prio=9 os_prio=0 tid=0x00007f59a80ac000 nid=0x371c runnable [0x0000000000000000]
         java.lang.Thread.State: RUNNABLE
  
      "Finalizer" #3 daemon prio=8 os_prio=0 tid=0x00007f59a8074000 nid=0x371b in Object.wait() [0x00007f59ace39000]
         java.lang.Thread.State: WAITING (on object monitor)
          at java.lang.Object.wait(Native Method)
          - waiting on &lt;0x00000000d5556950&gt; (a java.lang.ref.ReferenceQueue$Lock)
          at java.lang.ref.ReferenceQueue.remove(ReferenceQueue.java:143)
          - locked &lt;0x00000000d5556950&gt; (a java.lang.ref.ReferenceQueue$Lock)
          at java.lang.ref.ReferenceQueue.remove(ReferenceQueue.java:164)
          at java.lang.ref.Finalizer$FinalizerThread.run(Finalizer.java:209)
  
      "Reference Handler" #2 daemon prio=10 os_prio=0 tid=0x00007f59a8072000 nid=0x371a in Object.wait() [0x00007f59acf3a000]
         java.lang.Thread.State: WAITING (on object monitor)
          at java.lang.Object.wait(Native Method)
          - waiting on &lt;0x00000000d55563d0&gt; (a java.lang.ref.Reference$Lock)
          at java.lang.Object.wait(Object.java:502)
          at java.lang.ref.Reference$ReferenceHandler.run(Reference.java:157)
          - locked &lt;0x00000000d55563d0&gt; (a java.lang.ref.Reference$Lock)
  
      "VM Thread" os_prio=0 tid=0x00007f59a806d000 nid=0x3719 runnable 
  
      "VM Periodic Task Thread" os_prio=0 tid=0x00007f59a80c0000 nid=0x3720 waiting on condition 
  
      JNI global references: 278
  </code></pre>