Skip to content

[SPARK-18423][STREAMING] ReceiverTracker should close checkpoint dir when stopped even if it was not started #15867

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
HyukjinKwon wants to merge 3 commits into from
Closed

[SPARK-18423][STREAMING] ReceiverTracker should close checkpoint dir when stopped even if it was not started #15867

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
76f2718
Fix ReceiverTracker
HyukjinKwon Nov 12, 2016
8b05ba7
Fix comment better
HyukjinKwon Nov 12, 2016
4fe4c1b
clean up
HyukjinKwon Nov 14, 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
10 changes: 10 additions & 0 deletions streaming/src/main/scala/org/apache/spark/streaming/scheduler/ReceiverTracker.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -197,6 +197,13 @@ class ReceiverTracker(ssc: StreamingContext, skipReceiverLaunch: Boolean = false
receivedBlockTracker.stop()
logInfo("ReceiverTracker stopped")
trackerState = Stopped
} else if (isTrackerInitialized) {
trackerState = Stopping
// `ReceivedBlockTracker` is open when this instance is created. We should
// close this even if this `ReceiverTracker` is not started.
receivedBlockTracker.stop()
logInfo("ReceiverTracker stopped")
trackerState = Stopped
}
}

Expand Down Expand Up @@ -446,6 +453,9 @@ class ReceiverTracker(ssc: StreamingContext, skipReceiverLaunch: Boolean = false
endpoint.send(StartAllReceivers(receivers))
}

/** Check if tracker has been marked for initiated */
private def isTrackerInitialized: Boolean = trackerState == Initialized

/** Check if tracker has been marked for starting */
private def isTrackerStarted: Boolean = trackerState == Started

Expand Down