Skip to content

[SPARK-17038][Streaming] fix metrics retrieval source of 'lastReceivedBatch' #14681

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
keypointt wants to merge 2 commits into from
Closed

[SPARK-17038][Streaming] fix metrics retrieval source of 'lastReceivedBatch' #14681

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f0e39a0
[SPARK-17038] fix metrics retrieval source
keypointt Aug 15, 2016
ccd211f
[SPARK-17038] add some tests
keypointt Aug 17, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions streaming/src/main/scala/org/apache/spark/streaming/StreamingSource.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,11 +87,11 @@ private[streaming] class StreamingSource(ssc: StreamingContext) extends Source {
// Gauge for last received batch, useful for monitoring the streaming job's running status,
// displayed data -1 for any abnormal condition.
registerGaugeWithOption("lastReceivedBatch_submissionTime",
_.lastCompletedBatch.map(_.submissionTime), -1L)
_.lastReceivedBatch.map(_.submissionTime), -1L)
registerGaugeWithOption("lastReceivedBatch_processingStartTime",
_.lastCompletedBatch.flatMap(_.processingStartTime), -1L)
_.lastReceivedBatch.flatMap(_.processingStartTime), -1L)
registerGaugeWithOption("lastReceivedBatch_processingEndTime",
_.lastCompletedBatch.flatMap(_.processingEndTime), -1L)
_.lastReceivedBatch.flatMap(_.processingEndTime), -1L)

// Gauge for last received batch records.
registerGauge("lastReceivedBatch_records", _.lastReceivedBatchRecords.values.sum, 0L)
Expand Down
3 changes: 3 additions & 0 deletions ...ming/src/test/scala/org/apache/spark/streaming/ui/StreamingJobProgressListenerSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,7 @@ class StreamingJobProgressListenerSuite extends TestSuiteBase with Matchers {
listener.waitingBatches should be (List(BatchUIData(batchInfoSubmitted)))
listener.runningBatches should be (Nil)
listener.retainedCompletedBatches should be (Nil)
listener.lastReceivedBatch should be (Some(BatchUIData(batchInfoSubmitted)))
Copy link
Contributor

@jaceklaskowski jaceklaskowski Aug 17, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What do you think about should contain (BatchUIData(batchInfoSubmitted)) instead? See http://www.scalatest.org/user_guide/using_matchers#workingWithContainers.

Copy link
Contributor Author

@keypointt keypointt Aug 17, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm just following https://github.com/keypointt/spark/blob/ccd211f5d37d8d9c2d44511a766090aa6095a463/streaming/src/test/scala/org/apache/spark/streaming/ui/StreamingJobProgressListenerSuite.scala#L129, to keep the style consistent with others

listener.lastCompletedBatch should be (None)
listener.numUnprocessedBatches should be (1)
listener.numTotalCompletedBatches should be (0)
Expand All @@ -81,6 +82,7 @@ class StreamingJobProgressListenerSuite extends TestSuiteBase with Matchers {
listener.waitingBatches should be (Nil)
listener.runningBatches should be (List(BatchUIData(batchInfoStarted)))
listener.retainedCompletedBatches should be (Nil)
listener.lastReceivedBatch should be (Some(BatchUIData(batchInfoStarted)))
listener.lastCompletedBatch should be (None)
listener.numUnprocessedBatches should be (1)
listener.numTotalCompletedBatches should be (0)
Expand Down Expand Up @@ -123,6 +125,7 @@ class StreamingJobProgressListenerSuite extends TestSuiteBase with Matchers {
listener.waitingBatches should be (Nil)
listener.runningBatches should be (Nil)
listener.retainedCompletedBatches should be (List(BatchUIData(batchInfoCompleted)))
listener.lastReceivedBatch should be (Some(BatchUIData(batchInfoCompleted)))
listener.lastCompletedBatch should be (Some(BatchUIData(batchInfoCompleted)))
listener.numUnprocessedBatches should be (0)
listener.numTotalCompletedBatches should be (1)
Expand Down