[streaming] SPARK-10955. Disable dynamic allocation for Streaming applications.

Dynamic allocation can be painful for streaming apps and can lose data. The one drawback though
is that apps which run Streaming and non-streaming form the same context will also end up not
being able to do dynamic allocation.

Author: harishreedharan

streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala

@@ -44,7 +44,7 @@ import org.apache.spark.streaming.dstream._
 import org.apache.spark.streaming.receiver.{ActorReceiver, ActorSupervisorStrategy, Receiver}
 import org.apache.spark.streaming.scheduler.{JobScheduler, StreamingListener}
 import org.apache.spark.streaming.ui.{StreamingJobProgressListener, StreamingTab}
-import org.apache.spark.util.{CallSite, ShutdownHookManager, ThreadUtils}
+import org.apache.spark.util.{Utils, CallSite, ShutdownHookManager, ThreadUtils}
 
 /**
  * Main entry point for Spark Streaming functionality. It provides methods used to create
@@ -564,6 +564,15 @@ class StreamingContext private[streaming] (
           )
       }
     }
+
+    if (Utils.isDynamicAllocationEnabled(sc.conf)) {
+      val maxExecutors = sc.conf.getInt("spark.dynamicAllocation.maxExecutors", 2)
+      sc.conf.set("spark.dynamicAllocation.enabled", false.toString)
+      sc.conf.set("spark.executor.instances", maxExecutors.toString)
+      logWarning("Dynamic allocation is not supported with Spark Streaming currently, since it " +
+        s"could lead to data loss in some cases. " +
+        s"The number of executors is being set to $maxExecutors")
+    }
   }
 
   /**