Added more checks for BatchInfo

Author: djalova

python/pyspark/streaming/tests.py

@@ -443,24 +443,39 @@ def func(dstream):
 
         self.assertEqual(len(batchInfosSubmitted), 4)
         for info in batchInfosSubmitted:
-
+            self.assertGreaterEqual(info.batchTime().milliseconds(), 0)
+            self.assertGreaterEqual(info.submissionTime(), 0)
+            self.assertTrue(info.streamIdToInputInfo().isEmpty())
+            self.assertFalse(info.outputOperationInfos().isEmpty())
+            self.assertIsNotNone(info.outputOperationInfos().get(0))
             self.assertEqual(info.schedulingDelay(), -1)
             self.assertEqual(info.processingDelay(), -1)
             self.assertEqual(info.totalDelay(), -1)
+            self.assertEqual(info.numRecords(), 0)
 
         self.assertEqual(len(batchInfosStarted), 4)
         for info in batchInfosStarted:
+            self.assertGreaterEqual(info.batchTime().milliseconds(), 0)
+            self.assertGreaterEqual(info.submissionTime(), 0)
+            self.assertTrue(info.streamIdToInputInfo().isEmpty())
+            self.assertFalse(info.outputOperationInfos().isEmpty())
+            self.assertIsNotNone(info.outputOperationInfos().get(0))
             self.assertGreaterEqual(info.schedulingDelay(), 0)
             self.assertEqual(info.processingDelay(), -1)
             self.assertEqual(info.totalDelay(), -1)
+            self.assertEqual(info.numRecords(), 0)
 
         self.assertEqual(len(batchInfosCompleted), 4)
-
         for info in batchInfosCompleted:
-
+            self.assertGreaterEqual(info.batchTime().milliseconds(), 0)
+            self.assertGreaterEqual(info.submissionTime(), 0)
+            self.assertTrue(info.streamIdToInputInfo().isEmpty())
+            self.assertFalse(info.outputOperationInfos().isEmpty())
+            self.assertIsNotNone(info.outputOperationInfos().get(0))
             self.assertGreaterEqual(info.schedulingDelay(), 0)
             self.assertGreaterEqual(info.processingDelay(), 0)
             self.assertGreaterEqual(info.totalDelay(), 0)
+            self.assertEqual(info.numRecords(), 0)
 
 
 class WindowFunctionTests(PySparkStreamingTestCase):

streaming/src/main/scala/org/apache/spark/streaming/api/java/JavaStreamingListener.scala

@@ -49,7 +49,7 @@ private[streaming] trait PythonStreamingListener{
       outputOperationCompleted: JavaStreamingListenerOutputOperationCompleted) { }
 }
 
-class PythonStreamingListenerWrapper(listener: PythonStreamingListener)
+private[streaming] class PythonStreamingListenerWrapper(listener: PythonStreamingListener)
   extends JavaStreamingListener {
 
   /** Called when a receiver has been started */