Fix AwaitTermination(int timeout) error, and change to AwaitTerminationOrTimeout(long timeout)

csharp/Adapter/Microsoft.Spark.CSharp/Proxy/IStreamingContextProxy.cs

@@ -25,7 +25,7 @@ namespace Microsoft.Spark.CSharp.Proxy
         IDStreamProxy DirectKafkaStreamWithRepartition(List<string> topics, Dictionary<string, string> kafkaParams, Dictionary<string, long> fromOffsets, int numPartitions);
         IDStreamProxy Union(IDStreamProxy firstDStreams, IDStreamProxy[] otherDStreams);
         void AwaitTermination();
-        void AwaitTermination(int timeout);
+        void AwaitTerminationOrTimeout(long timeout);
         IDStreamProxy CreateCSharpDStream(IDStreamProxy jdstream, byte[] func, string serializationMode);
         IDStreamProxy CreateCSharpTransformed2DStream(IDStreamProxy jdstream, IDStreamProxy jother, byte[] func, string serializationMode, string serializationModeOther);
         IDStreamProxy CreateCSharpReducedWindowedDStream(IDStreamProxy jdstream, byte[] func, byte[] invFunc, int windowSeconds, int slideSeconds, string serializationMode);

csharp/Adapter/Microsoft.Spark.CSharp/Proxy/Ipc/StreamingContextIpcProxy.cs

@@ -263,9 +263,9 @@ namespace Microsoft.Spark.CSharp.Proxy.Ipc
             SparkCLRIpcProxy.JvmBridge.CallNonStaticJavaMethod(jvmStreamingContextReference, "awaitTermination");
         }
 
-        public void AwaitTermination(int timeout)
+        public void AwaitTerminationOrTimeout(long timeout)
         {
-            SparkCLRIpcProxy.JvmBridge.CallNonStaticJavaMethod(jvmStreamingContextReference, "awaitTermination", new object[] { timeout });
+            SparkCLRIpcProxy.JvmBridge.CallNonStaticJavaMethod(jvmStreamingContextReference, "awaitTerminationOrTimeout", new object[] { timeout });
         }
 
         private void ProcessCallbackRequest(object socket)

csharp/Adapter/Microsoft.Spark.CSharp/Streaming/StreamingContext.cs

@@ -163,10 +163,10 @@ namespace Microsoft.Spark.CSharp.Streaming
         /// <summary>
         /// Wait for the execution to stop.
         /// </summary>
-        /// <param name="timeout">time to wait in seconds</param>
-        public void AwaitTerminationOrTimeout(int timeout)
+        /// <param name="timeout">time to wait in milliseconds</param>
+        public void AwaitTerminationOrTimeout(long timeout)
         {
-            streamingContextProxy.AwaitTermination(timeout);
+            streamingContextProxy.AwaitTerminationOrTimeout(timeout);
         }
 
         /// <summary>

csharp/AdapterTest/Mocks/MockStreamingContextProxy.cs

@@ -63,7 +63,7 @@ namespace AdapterTest.Mocks
         {
         }
 
-        public void AwaitTermination(int timeout)
+        public void AwaitTerminationOrTimeout(long timeout)
         {
         }
 

csharp/AdapterTest/StreamingContextTest.cs

@@ -46,5 +46,28 @@ namespace AdapterTest
             ssc.AwaitTermination();
             ssc.Stop();
         }
+
+        [Test]
+        public void TestStreamingAwaitTimeout()
+        {
+            var ssc = new StreamingContext(new SparkContext("", ""), 1000);
+            Assert.IsNotNull((ssc.streamingContextProxy as MockStreamingContextProxy));
+
+            ssc.Start();
+            ssc.Remember(1000);
+            ssc.Checkpoint(Path.GetTempPath());
+
+            var textFile = ssc.TextFileStream(Path.GetTempPath());
+            Assert.IsNotNull(textFile.DStreamProxy);
+
+            var socketStream = ssc.SocketTextStream(IPAddress.Loopback.ToString(), 12345);
+            Assert.IsNotNull(socketStream.DStreamProxy);
+
+            var union = ssc.Union(textFile, socketStream);
+            Assert.IsNotNull(union.DStreamProxy);
+
+            ssc.AwaitTerminationOrTimeout(3000);
+            ssc.Stop();
+        }
     }
 }