[SPARK-11462][STREAMING] Add JavaStreamingListener

Currently, StreamingListener is not Java friendly because it exposes some Scala collections to Java users directly, such as Option, Map.

This PR added a Java version of StreamingListener and a bunch of Java friendly classes for Java users.

Author: zsxwing <zsxwing@gmail.com>
Author: Shixiong Zhu <shixiong@databricks.com>

Closes #9420 from zsxwing/java-streaming-listener.

2 files changed, 290 insertions, 0 deletions

diff --git a/streaming/src/main/scala/org/apache/spark/streaming/api/java/JavaStreamingListener.scala b/streaming/src/main/scala/org/apache/spark/streaming/api/java/JavaStreamingListener.scala
new file mode 100644
index 0000000000..c86c7101ff
--- /dev/null
+++ b/streaming/src/main/scala/org/apache/spark/streaming/api/java/JavaStreamingListener.scala
@@ -0,0 +1,168 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.streaming.api.java
+
+import org.apache.spark.streaming.Time
+
+/**
+ * A listener interface for receiving information about an ongoing streaming  computation.
+ */
+private[streaming] class JavaStreamingListener {
+
+  /** Called when a receiver has been started */
+  def onReceiverStarted(receiverStarted: JavaStreamingListenerReceiverStarted): Unit = { }
+
+  /** Called when a receiver has reported an error */
+  def onReceiverError(receiverError: JavaStreamingListenerReceiverError): Unit = { }
+
+  /** Called when a receiver has been stopped */
+  def onReceiverStopped(receiverStopped: JavaStreamingListenerReceiverStopped): Unit = { }
+
+  /** Called when a batch of jobs has been submitted for processing. */
+  def onBatchSubmitted(batchSubmitted: JavaStreamingListenerBatchSubmitted): Unit = { }
+
+  /** Called when processing of a batch of jobs has started.  */
+  def onBatchStarted(batchStarted: JavaStreamingListenerBatchStarted): Unit = { }
+
+  /** Called when processing of a batch of jobs has completed. */
+  def onBatchCompleted(batchCompleted: JavaStreamingListenerBatchCompleted): Unit = { }
+
+  /** Called when processing of a job of a batch has started. */
+  def onOutputOperationStarted(
+      outputOperationStarted: JavaStreamingListenerOutputOperationStarted): Unit = { }
+
+  /** Called when processing of a job of a batch has completed. */
+  def onOutputOperationCompleted(
+      outputOperationCompleted: JavaStreamingListenerOutputOperationCompleted): Unit = { }
+}
+
+/**
+ * Base trait for events related to JavaStreamingListener
+ */
+private[streaming] sealed trait JavaStreamingListenerEvent
+
+private[streaming] class JavaStreamingListenerBatchSubmitted(val batchInfo: JavaBatchInfo)
+  extends JavaStreamingListenerEvent
+
+private[streaming] class JavaStreamingListenerBatchCompleted(val batchInfo: JavaBatchInfo)
+  extends JavaStreamingListenerEvent
+
+private[streaming] class JavaStreamingListenerBatchStarted(val batchInfo: JavaBatchInfo)
+  extends JavaStreamingListenerEvent
+
+private[streaming] class JavaStreamingListenerOutputOperationStarted(
+    val outputOperationInfo: JavaOutputOperationInfo) extends JavaStreamingListenerEvent
+
+private[streaming] class JavaStreamingListenerOutputOperationCompleted(
+    val outputOperationInfo: JavaOutputOperationInfo) extends JavaStreamingListenerEvent
+
+private[streaming] class JavaStreamingListenerReceiverStarted(val receiverInfo: JavaReceiverInfo)
+  extends JavaStreamingListenerEvent
+
+private[streaming] class JavaStreamingListenerReceiverError(val receiverInfo: JavaReceiverInfo)
+  extends JavaStreamingListenerEvent
+
+private[streaming] class JavaStreamingListenerReceiverStopped(val receiverInfo: JavaReceiverInfo)
+  extends JavaStreamingListenerEvent
+
+/**
+ * Class having information on batches.
+ *
+ * @param batchTime Time of the batch
+ * @param streamIdToInputInfo A map of input stream id to its input info
+ * @param submissionTime Clock time of when jobs of this batch was submitted to the streaming
+ *                       scheduler queue
+ * @param processingStartTime Clock time of when the first job of this batch started processing.
+ *                            `-1` means the batch has not yet started
+ * @param processingEndTime Clock time of when the last job of this batch finished processing. `-1`
+ *                          means the batch has not yet completed.
+ * @param schedulingDelay Time taken for the first job of this batch to start processing from the
+ *                        time this batch was submitted to the streaming scheduler. Essentially, it
+ *                        is `processingStartTime` - `submissionTime`. `-1` means the batch has not
+ *                        yet started
+ * @param processingDelay Time taken for the all jobs of this batch to finish processing from the
+ *                        time they started processing. Essentially, it is
+ *                        `processingEndTime` - `processingStartTime`. `-1` means the batch has not
+ *                        yet completed.
+ * @param totalDelay Time taken for all the jobs of this batch to finish processing from the time
+ *                   they were submitted.  Essentially, it is `processingDelay` + `schedulingDelay`.
+ *                   `-1` means the batch has not yet completed.
+ * @param numRecords The number of recorders received by the receivers in this batch
+ * @param outputOperationInfos The output operations in this batch
+ */
+private[streaming] case class JavaBatchInfo(
+    batchTime: Time,
+    streamIdToInputInfo: java.util.Map[Int, JavaStreamInputInfo],
+    submissionTime: Long,
+    processingStartTime: Long,
+    processingEndTime: Long,
+    schedulingDelay: Long,
+    processingDelay: Long,
+    totalDelay: Long,
+    numRecords: Long,
+    outputOperationInfos: java.util.Map[Int, JavaOutputOperationInfo])
+
+/**
+ * Track the information of input stream at specified batch time.
+ *
+ * @param inputStreamId the input stream id
+ * @param numRecords the number of records in a batch
+ * @param metadata metadata for this batch. It should contain at least one standard field named
+ *                 "Description" which maps to the content that will be shown in the UI.
+ * @param metadataDescription description of this input stream
+ */
+private[streaming] case class JavaStreamInputInfo(
+    inputStreamId: Int,
+    numRecords: Long,
+    metadata: java.util.Map[String, Any],
+    metadataDescription: String)
+
+/**
+ * Class having information about a receiver
+ */
+private[streaming] case class JavaReceiverInfo(
+    streamId: Int,
+    name: String,
+    active: Boolean,
+    location: String,
+    lastErrorMessage: String,
+    lastError: String,
+    lastErrorTime: Long)
+
+/**
+ * Class having information on output operations.
+ *
+ * @param batchTime Time of the batch
+ * @param id Id of this output operation. Different output operations have different ids in a batch.
+ * @param name The name of this output operation.
+ * @param description The description of this output operation.
+ * @param startTime Clock time of when the output operation started processing. `-1` means the
+ *                  output operation has not yet started
+ * @param endTime Clock time of when the output operation started processing. `-1` means the output
+ *                operation has not yet completed
+ * @param failureReason Failure reason if this output operation fails. If the output operation is
+ *                      successful, this field is `null`.
+ */
+private[streaming] case class JavaOutputOperationInfo(
+    batchTime: Time,
+    id: Int,
+    name: String,
+    description: String,
+    startTime: Long,
+    endTime: Long,
+    failureReason: String)
diff --git a/streaming/src/main/scala/org/apache/spark/streaming/api/java/JavaStreamingListenerWrapper.scala b/streaming/src/main/scala/org/apache/spark/streaming/api/java/JavaStreamingListenerWrapper.scala
new file mode 100644
index 0000000000..2c60b396a6
--- /dev/null
+++ b/streaming/src/main/scala/org/apache/spark/streaming/api/java/JavaStreamingListenerWrapper.scala
@@ -0,0 +1,122 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.streaming.api.java
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.streaming.scheduler._
+
+/**
+ * A wrapper to convert a [[JavaStreamingListener]] to a [[StreamingListener]].
+ */
+private[streaming] class JavaStreamingListenerWrapper(javaStreamingListener: JavaStreamingListener)
+  extends StreamingListener {
+
+  private def toJavaReceiverInfo(receiverInfo: ReceiverInfo): JavaReceiverInfo = {
+    JavaReceiverInfo(
+      receiverInfo.streamId,
+      receiverInfo.name,
+      receiverInfo.active,
+      receiverInfo.location,
+      receiverInfo.lastErrorMessage,
+      receiverInfo.lastError,
+      receiverInfo.lastErrorTime
+    )
+  }
+
+  private def toJavaStreamInputInfo(streamInputInfo: StreamInputInfo): JavaStreamInputInfo = {
+    JavaStreamInputInfo(
+      streamInputInfo.inputStreamId,
+      streamInputInfo.numRecords: Long,
+      streamInputInfo.metadata.asJava,
+      streamInputInfo.metadataDescription.orNull
+    )
+  }
+
+  private def toJavaOutputOperationInfo(
+      outputOperationInfo: OutputOperationInfo): JavaOutputOperationInfo = {
+    JavaOutputOperationInfo(
+      outputOperationInfo.batchTime,
+      outputOperationInfo.id,
+      outputOperationInfo.name,
+      outputOperationInfo.description: String,
+      outputOperationInfo.startTime.getOrElse(-1),
+      outputOperationInfo.endTime.getOrElse(-1),
+      outputOperationInfo.failureReason.orNull
+    )
+  }
+
+  private def toJavaBatchInfo(batchInfo: BatchInfo): JavaBatchInfo = {
+    JavaBatchInfo(
+      batchInfo.batchTime,
+      batchInfo.streamIdToInputInfo.mapValues(toJavaStreamInputInfo(_)).asJava,
+      batchInfo.submissionTime,
+      batchInfo.processingStartTime.getOrElse(-1),
+      batchInfo.processingEndTime.getOrElse(-1),
+      batchInfo.schedulingDelay.getOrElse(-1),
+      batchInfo.processingDelay.getOrElse(-1),
+      batchInfo.totalDelay.getOrElse(-1),
+      batchInfo.numRecords,
+      batchInfo.outputOperationInfos.mapValues(toJavaOutputOperationInfo(_)).asJava
+    )
+  }
+
+  override def onReceiverStarted(receiverStarted: StreamingListenerReceiverStarted): Unit = {
+    javaStreamingListener.onReceiverStarted(
+      new JavaStreamingListenerReceiverStarted(toJavaReceiverInfo(receiverStarted.receiverInfo)))
+  }
+
+  override def onReceiverError(receiverError: StreamingListenerReceiverError): Unit = {
+    javaStreamingListener.onReceiverError(
+      new JavaStreamingListenerReceiverError(toJavaReceiverInfo(receiverError.receiverInfo)))
+  }
+
+  override def onReceiverStopped(receiverStopped: StreamingListenerReceiverStopped): Unit = {
+    javaStreamingListener.onReceiverStopped(
+      new JavaStreamingListenerReceiverStopped(toJavaReceiverInfo(receiverStopped.receiverInfo)))
+  }
+
+  override def onBatchSubmitted(batchSubmitted: StreamingListenerBatchSubmitted): Unit = {
+    javaStreamingListener.onBatchSubmitted(
+      new JavaStreamingListenerBatchSubmitted(toJavaBatchInfo(batchSubmitted.batchInfo)))
+  }
+
+  override def onBatchStarted(batchStarted: StreamingListenerBatchStarted): Unit = {
+    javaStreamingListener.onBatchStarted(
+      new JavaStreamingListenerBatchStarted(toJavaBatchInfo(batchStarted.batchInfo)))
+  }
+
+  override def onBatchCompleted(batchCompleted: StreamingListenerBatchCompleted): Unit = {
+    javaStreamingListener.onBatchCompleted(
+      new JavaStreamingListenerBatchCompleted(toJavaBatchInfo(batchCompleted.batchInfo)))
+  }
+
+  override def onOutputOperationStarted(
+      outputOperationStarted: StreamingListenerOutputOperationStarted): Unit = {
+    javaStreamingListener.onOutputOperationStarted(new JavaStreamingListenerOutputOperationStarted(
+      toJavaOutputOperationInfo(outputOperationStarted.outputOperationInfo)))
+  }
+
+  override def onOutputOperationCompleted(
+      outputOperationCompleted: StreamingListenerOutputOperationCompleted): Unit = {
+    javaStreamingListener.onOutputOperationCompleted(
+      new JavaStreamingListenerOutputOperationCompleted(
+        toJavaOutputOperationInfo(outputOperationCompleted.outputOperationInfo)))
+  }
+
+}