StatusUpdateStreamingListener.scala

class StatusUpdateStreamingListener(
    pipelineName: String, 
    pipelineId: Long) extends StreamingListener {

  @Inject var statusTrackerClient: StatusTrackerClient = _

  override def onBatchCompleted(batch: StreamingListenerBatchCompleted): Unit = {

    val msg = StreamingStatusMessage.newBuilder()
        .setPipelineName(pipelineName)
        .setId(pipelineId)
        .setBatchTime(batch.batchInfo.batchTime.milliseconds)
        .build

    statusTrackerClient.send(msg)
  }
}