You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by dragos <gi...@git.apache.org> on 2015/01/29 17:18:41 UTC
[GitHub] spark pull request: [SPARK-4631][streaming] Wait for a receiver to...
GitHub user dragos opened a pull request:
https://github.com/apache/spark/pull/4270
[SPARK-4631][streaming] Wait for a receiver to start before publishing test data.
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.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/dragos/spark issue/fix-flaky-test-SPARK-4631
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/4270.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #4270
----
commit d408a8eb36fe59f64c7b3ce9eb9d22cee6dbd684
Author: Iulian Dragos <ja...@gmail.com>
Date: 2015-01-29T16:06:42Z
Install callback before connecting to MQTT broker.
According to the MQTT client docs the callback needs to be installed
before the client connects or subscribes to a topic. Otherwise
messages may be lost.
commit 261653fad1abd4b40931663e9fc8aa7510f635db
Author: Iulian Dragos <ja...@gmail.com>
Date: 2015-01-29T16:11:14Z
[SPARK-4631][streaming] Wait for a receiver to start before publishing test data.
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.
----
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4631][streaming] Wait for a receiver to...
Posted by dragos <gi...@git.apache.org>.
Github user dragos commented on a diff in the pull request:
https://github.com/apache/spark/pull/4270#discussion_r23828743
--- Diff: external/mqtt/src/test/scala/org/apache/spark/streaming/mqtt/MQTTStreamSuite.scala ---
@@ -121,8 +130,18 @@ class MQTTStreamSuite extends FunSuite with Eventually with BeforeAndAfter {
val message: MqttMessage = new MqttMessage(data.getBytes("utf-8"))
message.setQos(1)
message.setRetained(true)
+
+ def tryPublish(times: Int): Unit =
+ if (times > 0) try {
+ msgTopic.publish(message)
+ } catch {
+ case e: MqttException if e.getReasonCode == MqttException.REASON_CODE_MAX_INFLIGHT =>
+ Thread.sleep(50) // wait for Spark streaming to consume something from the message queue
+ tryPublish(times - 1)
+ }
+
for (i <- 0 to 100) {
- msgTopic.publish(message)
+ tryPublish(3)
--- End diff --
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.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4631][streaming][FIX] Wait for a receiv...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on the pull request:
https://github.com/apache/spark/pull/4270#issuecomment-72539246
Merging this. Thanks for catching and fixing this.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4631][streaming] Wait for a receiver to...
Posted by dragos <gi...@git.apache.org>.
Github user dragos commented on a diff in the pull request:
https://github.com/apache/spark/pull/4270#discussion_r23828701
--- Diff: external/mqtt/src/main/scala/org/apache/spark/streaming/mqtt/MQTTInputDStream.scala ---
@@ -55,14 +55,14 @@ class MQTTInputDStream(
brokerUrl: String,
topic: String,
storageLevel: StorageLevel
- ) extends ReceiverInputDStream[String](ssc_) with Logging {
--- End diff --
No, it was not necessary, I just noticed that it wasn't needed, so I thought I should be a boy scout :)
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4631][streaming] Wait for a receiver to...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/4270#issuecomment-72149096
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.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4631][streaming] Wait for a receiver to...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/4270#issuecomment-72149088
[Test build #26366 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/26366/consoleFull) for PR 4270 at commit [`261653f`](https://github.com/apache/spark/commit/261653fad1abd4b40931663e9fc8aa7510f635db).
* This patch **passes all tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4631][streaming][FIX] Wait for a receiv...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/4270
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4631][streaming] Wait for a receiver to...
Posted by prabeesh <gi...@git.apache.org>.
Github user prabeesh commented on a diff in the pull request:
https://github.com/apache/spark/pull/4270#discussion_r23827773
--- Diff: external/mqtt/src/test/scala/org/apache/spark/streaming/mqtt/MQTTStreamSuite.scala ---
@@ -121,8 +130,18 @@ class MQTTStreamSuite extends FunSuite with Eventually with BeforeAndAfter {
val message: MqttMessage = new MqttMessage(data.getBytes("utf-8"))
message.setQos(1)
message.setRetained(true)
+
+ def tryPublish(times: Int): Unit =
+ if (times > 0) try {
+ msgTopic.publish(message)
+ } catch {
+ case e: MqttException if e.getReasonCode == MqttException.REASON_CODE_MAX_INFLIGHT =>
+ Thread.sleep(50) // wait for Spark streaming to consume something from the message queue
+ tryPublish(times - 1)
+ }
+
for (i <- 0 to 100) {
- msgTopic.publish(message)
+ tryPublish(3)
--- End diff --
for (i <- 0 to 100) {
try {
msgtopic.publish(message)
println("Published data. topic: %s; Message: %s".format(msgtopic.getName(), message))
} catch {
case e: MqttException => Thread.sleep(1)
}
}
I think tread sleep is 1 second is fine.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4631][streaming][FIX] Wait for a receiv...
Posted by dragos <gi...@git.apache.org>.
Github user dragos commented on the pull request:
https://github.com/apache/spark/pull/4270#issuecomment-72177497
OK, I went for the amend (the way we do in our projects). This way the commit history is cleaner.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4631][streaming] Wait for a receiver to...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on the pull request:
https://github.com/apache/spark/pull/4270#issuecomment-72138023
Can you add "[FIX]" to the title, so that it is clear that this fixes SPARK-631
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4631][streaming] Wait for a receiver to...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on the pull request:
https://github.com/apache/spark/pull/4270#issuecomment-72137958
Jenkins, this is okay to test.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4631][streaming] Wait for a receiver to...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/4270#issuecomment-72053982
Can one of the admins verify this patch?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4631][streaming] Wait for a receiver to...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/4270#discussion_r23818318
--- Diff: external/mqtt/src/main/scala/org/apache/spark/streaming/mqtt/MQTTInputDStream.scala ---
@@ -55,14 +55,14 @@ class MQTTInputDStream(
brokerUrl: String,
topic: String,
storageLevel: StorageLevel
- ) extends ReceiverInputDStream[String](ssc_) with Logging {
--- End diff --
This is a good change, but this change?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4631][streaming] Wait for a receiver to...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/4270#discussion_r23822317
--- Diff: external/mqtt/src/test/scala/org/apache/spark/streaming/mqtt/MQTTStreamSuite.scala ---
@@ -121,8 +130,18 @@ class MQTTStreamSuite extends FunSuite with Eventually with BeforeAndAfter {
val message: MqttMessage = new MqttMessage(data.getBytes("utf-8"))
message.setQos(1)
message.setRetained(true)
+
+ def tryPublish(times: Int): Unit =
--- End diff --
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.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4631][streaming] Wait for a receiver to...
Posted by Bilna <gi...@git.apache.org>.
Github user Bilna commented on a diff in the pull request:
https://github.com/apache/spark/pull/4270#discussion_r23831606
--- Diff: external/mqtt/src/test/scala/org/apache/spark/streaming/mqtt/MQTTStreamSuite.scala ---
@@ -121,8 +130,18 @@ class MQTTStreamSuite extends FunSuite with Eventually with BeforeAndAfter {
val message: MqttMessage = new MqttMessage(data.getBytes("utf-8"))
message.setQos(1)
message.setRetained(true)
+
+ def tryPublish(times: Int): Unit =
+ if (times > 0) try {
+ msgTopic.publish(message)
+ } catch {
+ case e: MqttException if e.getReasonCode == MqttException.REASON_CODE_MAX_INFLIGHT =>
+ Thread.sleep(50) // wait for Spark streaming to consume something from the message queue
+ tryPublish(times - 1)
+ }
+
for (i <- 0 to 100) {
- msgTopic.publish(message)
+ tryPublish(3)
--- End diff --
Please try this also.
Along with waitForReceiverToStart(), change the 'for loop' range from 0 to 9 . It is just for testing.
I am getting this error only if I change the MqttDefaultFilePersistence to memoryPersistence. I solved the problem by applying the proposed solution above.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4631][streaming][FIX] Wait for a receiv...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/4270#issuecomment-72177736
[Test build #26397 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/26397/consoleFull) for PR 4270 at commit [`f66c482`](https://github.com/apache/spark/commit/f66c482878a22a3085f1de311a4ace9a9a957070).
* This patch merges cleanly.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4631][streaming][FIX] Wait for a receiv...
Posted by srowen <gi...@git.apache.org>.
Github user srowen commented on the pull request:
https://github.com/apache/spark/pull/4270#issuecomment-72191363
@dragos you can just add a commit. It is easier to review and commits are squashed on merge anyway.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4631][streaming][FIX] Wait for a receiv...
Posted by dragos <gi...@git.apache.org>.
Github user dragos commented on the pull request:
https://github.com/apache/spark/pull/4270#issuecomment-72421376
I think I addressed all reviewer's comments. @tdas @Bilna Is there anything else I should do?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4631][streaming] Wait for a receiver to...
Posted by Bilna <gi...@git.apache.org>.
Github user Bilna commented on a diff in the pull request:
https://github.com/apache/spark/pull/4270#discussion_r23828884
--- Diff: external/mqtt/src/test/scala/org/apache/spark/streaming/mqtt/MQTTStreamSuite.scala ---
@@ -121,8 +130,18 @@ class MQTTStreamSuite extends FunSuite with Eventually with BeforeAndAfter {
val message: MqttMessage = new MqttMessage(data.getBytes("utf-8"))
message.setQos(1)
message.setRetained(true)
+
+ def tryPublish(times: Int): Unit =
+ if (times > 0) try {
+ msgTopic.publish(message)
+ } catch {
+ case e: MqttException if e.getReasonCode == MqttException.REASON_CODE_MAX_INFLIGHT =>
+ Thread.sleep(50) // wait for Spark streaming to consume something from the message queue
+ tryPublish(times - 1)
+ }
+
for (i <- 0 to 100) {
- msgTopic.publish(message)
+ tryPublish(3)
--- End diff --
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.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4631][streaming] Wait for a receiver to...
Posted by Bilna <gi...@git.apache.org>.
Github user Bilna commented on a diff in the pull request:
https://github.com/apache/spark/pull/4270#discussion_r23828369
--- Diff: external/mqtt/src/main/scala/org/apache/spark/streaming/mqtt/MQTTInputDStream.scala ---
@@ -103,7 +97,15 @@ class MQTTReceiver(
}
}
- // Set up callback for MqttClient
+ // Set up callback for MqttClient. This needs to happen before
+ // connecting or subscribing, otherwise messages may be lost
client.setCallback(callback)
--- End diff --
I agree with this modification.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4631][streaming][FIX] Wait for a receiv...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/4270#issuecomment-72185292
[Test build #26397 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/26397/consoleFull) for PR 4270 at commit [`f66c482`](https://github.com/apache/spark/commit/f66c482878a22a3085f1de311a4ace9a9a957070).
* This patch **passes all tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4631][streaming][FIX] Wait for a receiv...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/4270#issuecomment-72185297
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.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4631][streaming] Wait for a receiver to...
Posted by dragos <gi...@git.apache.org>.
Github user dragos commented on a diff in the pull request:
https://github.com/apache/spark/pull/4270#discussion_r23829317
--- Diff: external/mqtt/src/test/scala/org/apache/spark/streaming/mqtt/MQTTStreamSuite.scala ---
@@ -121,8 +130,18 @@ class MQTTStreamSuite extends FunSuite with Eventually with BeforeAndAfter {
val message: MqttMessage = new MqttMessage(data.getBytes("utf-8"))
message.setQos(1)
message.setRetained(true)
+
+ def tryPublish(times: Int): Unit =
+ if (times > 0) try {
+ msgTopic.publish(message)
+ } catch {
+ case e: MqttException if e.getReasonCode == MqttException.REASON_CODE_MAX_INFLIGHT =>
+ Thread.sleep(50) // wait for Spark streaming to consume something from the message queue
+ tryPublish(times - 1)
+ }
+
for (i <- 0 to 100) {
- msgTopic.publish(message)
+ tryPublish(3)
--- End diff --
Ok, I'll check that.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4631][streaming][FIX] Wait for a receiv...
Posted by dragos <gi...@git.apache.org>.
Github user dragos commented on the pull request:
https://github.com/apache/spark/pull/4270#issuecomment-72202747
@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.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4631][streaming] Wait for a receiver to...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on the pull request:
https://github.com/apache/spark/pull/4270#issuecomment-72143482
Jenkins, this is ok to test.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4631][streaming] Wait for a receiver to...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/4270#issuecomment-72143897
[Test build #26366 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/26366/consoleFull) for PR 4270 at commit [`261653f`](https://github.com/apache/spark/commit/261653fad1abd4b40931663e9fc8aa7510f635db).
* This patch merges cleanly.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4631][streaming] Wait for a receiver to...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on the pull request:
https://github.com/apache/spark/pull/4270#issuecomment-72148231
@blina @prabeesh Please take a look.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4631][streaming][FIX] Wait for a receiv...
Posted by dragos <gi...@git.apache.org>.
Github user dragos commented on the pull request:
https://github.com/apache/spark/pull/4270#issuecomment-72176963
@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?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4631][streaming] Wait for a receiver to...
Posted by dragos <gi...@git.apache.org>.
Github user dragos commented on a diff in the pull request:
https://github.com/apache/spark/pull/4270#discussion_r23828752
--- Diff: external/mqtt/src/test/scala/org/apache/spark/streaming/mqtt/MQTTStreamSuite.scala ---
@@ -121,8 +130,18 @@ class MQTTStreamSuite extends FunSuite with Eventually with BeforeAndAfter {
val message: MqttMessage = new MqttMessage(data.getBytes("utf-8"))
message.setQos(1)
message.setRetained(true)
+
+ def tryPublish(times: Int): Unit =
--- End diff --
Ok, will do.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org