Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/6294#discussion_r30917648
--- Diff:
streaming/src/main/scala/org/apache/spark/streaming/receiver/ReceiverSupervisor.scala
---
@@ -117,11 +117,15 @@ private[streaming] abstract class ReceiverSupervisor(
/** Start receiver */
def startReceiver(): Unit = synchronized {
try {
- logInfo("Starting receiver")
- receiver.onStart()
- logInfo("Called receiver onStart")
- onReceiverStart()
- receiverState = Started
+ if (onReceiverStart()) {
+ logInfo("Starting receiver")
+ receiverState = Started
+ receiver.onStart()
+ logInfo("Called receiver onStart")
+ } else {
+ // The driver refused us
+ stop("Registered unsuccessfully because the driver refused" +
streamId, None)
+ }
} catch {
case t: Throwable =>
--- End diff --
Can you also convert this `case Throwable` to `case NonFatal(e)`? I had
caught Throwable in a number of places but later learnt that we should not
catch all throwable as Scala uses throwables for some control logic. Best to
always catch `NonFatal(e)`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
