[SPARK-4631][streaming][FIX] Wait for a receiver to start before publishing test data.

[dragos]

This fixes two sources of non-deterministic failures in this test:

• wait for a receiver to be up before pushing data through MQTT
• gracefully handle the case where the MQTT client is overloaded. There’s
  a hard-coded limit of 10 in-flight messages, and this test may hit it.
  Instead of crashing, we retry sending the message.

Both of these are needed to make the test pass reliably on my machine.

[AmplabJenkins]

Can one of the admins verify this patch?

[tdas]

Jenkins, this is okay to test.

[tdas]

Can you add "[FIX]" to the title, so that it is clear that this fixes SPARK-631

[tdas]

This is a good change, but why this change for this fix? Was it necessary?

[dragos]

No, it was not necessary, I just noticed that it wasn't needed, so I thought I should be a boy scout :)

[tdas]

Jenkins, this is ok to test.

[SparkQA]

Test build #26366 has started for PR 4270 at commit 261653f.

• This patch merges cleanly.

[tdas]

@Blina @prabeesh Please take a look.

[tdas]

Defining this function inside the "if" looks a little weird, too nested. Might be better to just have it in for loop. Code would look less complicated.

[dragos]

Ok, will do.

[SparkQA]

Test build #26366 has finished for PR 4270 at commit 261653f.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[AmplabJenkins]

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/26366/
Test PASSed.

[prabeesh]

You can modify the code like this.

for (i <- 0 to 100) {
  try {
    msgtopic.publish(message)
  } catch {
    case e: MqttException => Thread.sleep(1)
  }
}

I think tread sleep 1 second is fine.

[dragos]

I can do that, but it won't retry the same message, instead if would just ignore a failure. In that case the total number of messages will be less than 101.

[Bilna]

I think waitForReceiverToStart() method is sufficient. Once the receiver is started, since the qos is set to 1 , as soon as the receiver receives the message, message will be flushed out from queue. If still problem exists the Thread.Sleep in the catch can solve the problem. There is no need for tryPublish (I guess)

[dragos]

Ok, I'll check that.

[Bilna]

Please try this also.
Along with waitForReceiverToStart(), change the 'for loop' range from 0 to 9 . It is just for testing.
I am getting the error only if I change the MqttDefaultFilePersistence to memoryPersistence. I solved the problem by applying the solution provided above.

[dragos]

@Bilna, @tdas what is the preferred protocol for reviews? Should I push a new commit, or amend the existing one and force-push to my repo?

[dragos]

OK, I went for the amend (the way we do in our projects). This way the commit history is cleaner.

[SparkQA]

Test build #26397 has started for PR 4270 at commit f66c482.

• This patch merges cleanly.

[SparkQA]

Test build #26397 has finished for PR 4270 at commit f66c482.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[AmplabJenkins]

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/26397/
Test PASSed.

[srowen]

@dragos you can just add a commit. It is easier to review and commits are squashed on merge anyway.

[dragos]

@srowen if it's not too bad I'll do that the next time. I already amended the last commit, and the change is pretty small, it shouldn't be too hard to re-review.

[dragos]

I think I addressed all reviewer's comments. @tdas @Bilna Is there anything else I should do?

[tdas]

Merging this. Thanks for catching and fixing this.