StreamingQueryManagerTests test fix (#907)

src/csharp/Microsoft.Spark.E2ETest/IpcTests/Sql/Streaming/DataStreamWriterTests.cs

@@ -87,7 +87,9 @@ namespace Microsoft.Spark.E2ETest.IpcTests
                         .WriteStream()
                         .Format("parquet")
                         .Option("checkpointLocation", tempDirectory.Path);
-                    Assert.IsType<StreamingQuery>(dsw.ToTable(tableName));
+
+                    StreamingQuery sq = dsw.ToTable(tableName);
+                    sq.Stop();
                 });
         }
 

src/csharp/Microsoft.Spark.E2ETest/IpcTests/Sql/Streaming/StreamingQueryManagerTests.cs

@@ -31,20 +31,18 @@ namespace Microsoft.Spark.E2ETest.IpcTests
             var intMemoryStream = new MemoryStream<int>(_spark);
             StreamingQuery sq1 = intMemoryStream
                 .ToDF().WriteStream().QueryName("intQuery").Format("console").Start();
-            string id1 = sq1.Id;
 
             var stringMemoryStream = new MemoryStream<string>(_spark);
             StreamingQuery sq2 = stringMemoryStream
                 .ToDF().WriteStream().QueryName("stringQuery").Format("console").Start();
-            string id2 = sq2.Id;
 
             StreamingQueryManager sqm = _spark.Streams();
 
             StreamingQuery[] streamingQueries = sqm.Active().ToArray();
             Assert.Equal(2, streamingQueries.Length);
 
-            Assert.IsType<StreamingQuery>(sqm.Get(id1));
-            Assert.IsType<StreamingQuery>(sqm.Get(id2));
+            Assert.IsType<StreamingQuery>(sqm.Get(sq1.Id));
+            Assert.IsType<StreamingQuery>(sqm.Get(sq2.Id));
 
             sqm.ResetTerminated();
 

src/csharp/Microsoft.Spark.E2ETest/IpcTests/Sql/Streaming/StreamingQueryTests.cs

@@ -29,7 +29,15 @@ namespace Microsoft.Spark.E2ETest.IpcTests
         {
             var intMemoryStream = new MemoryStream<int>(_spark);
             StreamingQuery sq = intMemoryStream
-                .ToDF().WriteStream().QueryName("testQuery").Format("console").Start();
+                .ToDF()
+                .WriteStream()
+                .QueryName("testQuery")
+                .Format("console")
+                .Trigger(Trigger.Once())
+                .Start();
+
+            sq.AwaitTermination();
+            Assert.IsType<bool>(sq.AwaitTermination(10));
 
             Assert.IsType<string>(sq.Name);
 
@@ -39,8 +47,6 @@ namespace Microsoft.Spark.E2ETest.IpcTests
 
             Assert.IsType<bool>(sq.IsActive());
 
-            Assert.IsType<bool>(sq.AwaitTermination(10));
-
             sq.Explain();
 
             Assert.Null(sq.Exception());