[SPARK-4062][Streaming]Add ReliableKafkaReceiver in Spark Streaming Kafka connector #2991

jerryshao · 2014-10-29T05:16:37Z

Add ReliableKafkaReceiver in Kafka connector to prevent data loss if WAL in Spark Streaming is enabled. Details and design doc can be seen in SPARK-4062.

jerryshao · 2014-10-29T05:17:42Z

Hi @tdas , would you mind taking a look at this? Thanks a lot.

SparkQA · 2014-10-29T05:19:44Z

Test build #22425 has started for PR 2991 at commit 5cc4cb1.

This patch merges cleanly.

SparkQA · 2014-10-29T06:14:37Z

Test build #22425 has finished for PR 2991 at commit 5cc4cb1.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

AmplabJenkins · 2014-10-29T06:14:40Z

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

SparkQA · 2014-10-29T07:04:46Z

Test build #22438 has started for PR 2991 at commit 09d57c5.

This patch merges cleanly.

SparkQA · 2014-10-29T08:16:14Z

Test build #22438 has finished for PR 2991 at commit 09d57c5.

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

AmplabJenkins · 2014-10-29T08:16:18Z

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

SparkQA · 2014-10-29T09:14:43Z

Test build #22446 has started for PR 2991 at commit bb57d05.

This patch merges cleanly.

SparkQA · 2014-10-29T10:27:54Z

Test build #22446 has finished for PR 2991 at commit bb57d05.

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

AmplabJenkins · 2014-10-29T10:27:57Z

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

tdas · 2014-10-30T21:01:39Z

@jerryshao Thanks for this change! I am taking a look at this. But the code does not merge. So please update the code to the master branch.

jerryshao · 2014-10-31T00:51:06Z

OK, will do :).

SparkQA · 2014-10-31T05:32:31Z

Test build #22595 has started for PR 2991 at commit 48752b2.

This patch merges cleanly.

SparkQA · 2014-10-31T06:44:52Z

Test build #22595 has finished for PR 2991 at commit 48752b2.

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

AmplabJenkins · 2014-10-31T06:44:55Z

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

jerryshao · 2014-11-06T05:42:46Z

Hi @tdas , would you mind reviewing this code? Thanks a lot.

tdas · 2014-11-07T03:00:31Z

streaming/src/main/scala/org/apache/spark/streaming/receiver/BlockGenerator.scala

    waitToPush()
    currentBuffer += data
+    listener.onStoreData(data, metadata)


What is the point of calling onStore data right after inserting into the buffer? If something needs to be done right after insertion then that can be done in the caller class.

The reason of putting onStoreData here is that this is synchronized with the same mutex, if we put this into caller class, we should use another mutex, this will add some other synchronization logic.

Actually adding synch logic in caller is more obvious, and introduces minimum extra logic in the BlockGenerator, keeping the BlockGenerator more generic. It is tricky to document stuff like "method 1 and method 2 of BlockGenerator are called in the same lock, but not other methods".

tdas · 2014-11-07T04:19:19Z

I have made some high level comments, but I am going to do another deeper roudn to understand the logic and correctness. In the mean time, since the WAL has been merged, could you test whether this patch is working as expected with the WAL turned on?

jerryshao · 2014-11-07T04:24:31Z

Thanks a lot for your review, I will address the above comments, besides we are under test with StreamBench, the result will return to you later.

tdas · 2014-11-07T10:30:07Z

external/kafka/src/main/scala/org/apache/spark/streaming/kafka/ReliableKafkaReceiver.scala

+  private lazy val blockOffsetMap =
+    new ConcurrentHashMap[StreamBlockId, Map[TopicAndPartition, Long]]
+
+  private lazy val blockGeneratorListener = new BlockGeneratorListener {


Good to define the named class for this generator listener.

SparkQA · 2014-11-10T06:45:02Z

Test build #23142 has started for PR 2991 at commit c174454.

This patch merges cleanly.

tdas · 2014-11-13T22:15:58Z

@jerryshao Here is another round of changes from me.
You correctly identified a flaw in the lock logic in the last change I made. I played around with different implementations, and I came up with two implementation that I think are correct, and tries to preserve the BlockGenerator performance for existing receivers. Both of them are pulls requests on your repo.

Refactor 1, Refactor 1: Achieve deadlock-free lock ordering by subclassing BlockGenerator and taking receiver lock first in BlockGenerator.updateCurrentBuffer jerryshao/apache-spark#7: Here I attempt to fix the flaw by taking receiver lock before updating the buffer. And this is done by extending the BlockGenerator and overriding the updateCurrentBuffer method to take receiver lock first. This ensures deadlock free locking by always taking locks in the same order - receiver lock followed by block generator lock. The default block generator code path is not affected, so other receiver should not be affected either.
Refactor2, Refactor 2: Removed the receiver's locks and essentially reverted to Saisai's original design jerryshao/apache-spark#6: I essentially reverted back to your original proposal. :) As I tried out various different implementations, and finally got "Refactor 1", I realized that it was more complicated than what you proposed. So I reverted back to that, and added a lot of scala docs explaining the behavior. Personally I am in favor of this now.

Besides I also eliminated more duplicate code form the unit tests and merged two unit tests to reduce test run times. Please take a look.

… Java more robust.

tdas · 2014-11-13T23:02:57Z

I did more refactoring for Refactoring 2, to create jerryshao#8 . This is what I finally recommend for merging. Please take a look. I have run the 3 kafka testsuites repeatedly for a long time, and they seemed consistently pass.

jerryshao · 2014-11-14T00:43:40Z

OK, I will, thanks a lot, greatly appreciated.

Refactor 3 = Refactor 2 + refactored KafkaStreamSuite further to elimite KafkaTestUtils, and made Java testsuite more robust

SparkQA · 2014-11-14T01:24:55Z

Test build #23345 has started for PR 2991 at commit 5461f1c.

This patch merges cleanly.

tdas · 2014-11-14T01:30:58Z

Lets see if this passes jenkins, I hadnt tried that yet

jerryshao · 2014-11-14T01:43:47Z

Hi TD, this test is so flaky, it fails several times in my local test:

- block addition, block to batch allocation and cleanup with write ahead log *** FAILED *** (21 milliseconds)
[info]   java.io.FileNotFoundException: File /tmp/1415929501402-0/receivedBlockMetadata/log-0-1000 does not exist.
[info]   at org.apache.hadoop.fs.RawLocalFileSystem.getFileStatus(RawLocalFileSystem.java:397)
[info]   at org.apache.hadoop.fs.RawLocalFileSystem.listStatus(RawLocalFileSystem.java:324)
[info]   at org.apache.spark.streaming.util.WriteAheadLogSuite$.getLogFilesInDirectory(WriteAheadLogSuite.scala:344)
[info]   at org.apache.spark.streaming.ReceivedBlockTrackerSuite.getWriteAheadLogFiles(ReceivedBlockTrackerSuite.scala:226)
[info]   at org.apache.spark.streaming.ReceivedBlockTrackerSuite$$anonfun$4.apply$mcV$sp(ReceivedBlockTrackerSuite.scala:171)
[info]   at org.apache.spark.streaming.ReceivedBlockTrackerSuite$$anonfun$4.apply(ReceivedBlockTrackerSuite.scala:96)
[info]   at org.apache.spark.streaming.ReceivedBlockTrackerSuite$$anonfun$4.apply(ReceivedBlockTrackerSuite.scala:96)
[info]   at 
....

SparkQA · 2014-11-14T02:45:26Z

Test build #23345 has finished for PR 2991 at commit 5461f1c.

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

AmplabJenkins · 2014-11-14T02:45:30Z

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

jerryshao · 2014-11-14T06:57:02Z

Jenkins, retest this please.

SparkQA · 2014-11-14T06:57:42Z

Test build #23352 has started for PR 2991 at commit 5461f1c.

This patch merges cleanly.

SparkQA · 2014-11-14T08:22:47Z

Test build #23352 has finished for PR 2991 at commit 5461f1c.

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

AmplabJenkins · 2014-11-14T08:22:51Z

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

jerryshao · 2014-11-14T08:52:11Z

Jenkins, retest this please.

SparkQA · 2014-11-14T08:59:51Z

Test build #23362 has started for PR 2991 at commit 5461f1c.

This patch merges cleanly.

SparkQA · 2014-11-14T10:59:51Z

Test build #23362 timed out for PR 2991 at commit 5461f1c after a configured wait of 120m.

AmplabJenkins · 2014-11-14T10:59:55Z

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

tdas · 2014-11-14T19:38:24Z

Jenkins, retest this please.

SparkQA · 2014-11-14T19:44:56Z

Test build #23380 has started for PR 2991 at commit 5461f1c.

This patch merges cleanly.

SparkQA · 2014-11-14T21:13:06Z

Test build #23380 has finished for PR 2991 at commit 5461f1c.

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

AmplabJenkins · 2014-11-14T21:13:10Z

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

tdas · 2014-11-14T22:32:05Z

Alright, I am merging this! Thanks @jerryshao for all the hard work. Though, I encourage you to continue testing this receiver on your side prior to the spark 1.2 release for stability and correctness.

…afka connector Add ReliableKafkaReceiver in Kafka connector to prevent data loss if WAL in Spark Streaming is enabled. Details and design doc can be seen in [SPARK-4062](https://issues.apache.org/jira/browse/SPARK-4062). Author: jerryshao <saisai.shao@intel.com> Author: Tathagata Das <tathagata.das1565@gmail.com> Author: Saisai Shao <saisai.shao@intel.com> Closes #2991 from jerryshao/kafka-refactor and squashes the following commits: 5461f1c [Saisai Shao] Merge pull request #8 from tdas/kafka-refactor3 eae4ad6 [Tathagata Das] Refectored KafkaStreamSuiteBased to eliminate KafkaTestUtils and made Java more robust. fab14c7 [Tathagata Das] minor update. 149948b [Tathagata Das] Fixed mistake 14630aa [Tathagata Das] Minor updates. d9a452c [Tathagata Das] Minor updates. ec2e95e [Tathagata Das] Removed the receiver's locks and essentially reverted to Saisai's original design. 2a20a01 [jerryshao] Address some comments 9f636b3 [Saisai Shao] Merge pull request #5 from tdas/kafka-refactor b2b2f84 [Tathagata Das] Refactored Kafka receiver logic and Kafka testsuites e501b3c [jerryshao] Add Mima excludes b798535 [jerryshao] Fix the missed issue e5e21c1 [jerryshao] Change to while loop ea873e4 [jerryshao] Further address the comments 98f3d07 [jerryshao] Fix comment style 4854ee9 [jerryshao] Address all the comments 96c7a1d [jerryshao] Update the ReliableKafkaReceiver unit test 8135d31 [jerryshao] Fix flaky test a949741 [jerryshao] Address the comments 16bfe78 [jerryshao] Change the ordering of imports 0894aef [jerryshao] Add some comments 77c3e50 [jerryshao] Code refactor and add some unit tests dd9aeeb [jerryshao] Initial commit for reliable Kafka receiver (cherry picked from commit 5930f64) Signed-off-by: Tathagata Das <tathagata.das1565@gmail.com>

jerryshao · 2014-11-15T01:26:33Z

Yes, I will, thanks a lot, greatly appreciate your help.

jerryshao force-pushed the kafka-refactor branch from 09d57c5 to bb57d05 Compare October 29, 2014 09:07

jerryshao force-pushed the kafka-refactor branch from bb57d05 to 48752b2 Compare October 31, 2014 05:29

tdas reviewed Nov 7, 2014
View reviewed changes

Fixed mistake

149948b

tdas added 2 commits November 13, 2014 14:21

minor update.

fab14c7

Refectored KafkaStreamSuiteBased to eliminate KafkaTestUtils and made…

eae4ad6

… Java more robust.

Merge pull request #8 from tdas/kafka-refactor3

5461f1c

Refactor 3 = Refactor 2 + refactored KafkaStreamSuite further to elimite KafkaTestUtils, and made Java testsuite more robust

asfgit closed this in 5930f64 Nov 14, 2014

[SPARK-4062][Streaming]Add ReliableKafkaReceiver in Spark Streaming Kafka connector #2991

[SPARK-4062][Streaming]Add ReliableKafkaReceiver in Spark Streaming Kafka connector #2991

Conversation

jerryshao commented Oct 29, 2014

jerryshao commented Oct 29, 2014

SparkQA commented Oct 29, 2014

SparkQA commented Oct 29, 2014

AmplabJenkins commented Oct 29, 2014

SparkQA commented Oct 29, 2014

SparkQA commented Oct 29, 2014

AmplabJenkins commented Oct 29, 2014

SparkQA commented Oct 29, 2014

SparkQA commented Oct 29, 2014

AmplabJenkins commented Oct 29, 2014

tdas commented Oct 30, 2014

jerryshao commented Oct 31, 2014

SparkQA commented Oct 31, 2014

SparkQA commented Oct 31, 2014

AmplabJenkins commented Oct 31, 2014

jerryshao commented Nov 6, 2014

tdas Nov 7, 2014

Choose a reason for hiding this comment

jerryshao Nov 7, 2014

Choose a reason for hiding this comment

tdas Nov 11, 2014

Choose a reason for hiding this comment

tdas commented Nov 7, 2014

jerryshao commented Nov 7, 2014

tdas Nov 7, 2014

Choose a reason for hiding this comment

SparkQA commented Nov 10, 2014

tdas commented Nov 13, 2014

tdas commented Nov 13, 2014

jerryshao commented Nov 14, 2014

SparkQA commented Nov 14, 2014

tdas commented Nov 14, 2014

jerryshao commented Nov 14, 2014

SparkQA commented Nov 14, 2014

AmplabJenkins commented Nov 14, 2014

jerryshao commented Nov 14, 2014

SparkQA commented Nov 14, 2014

SparkQA commented Nov 14, 2014

AmplabJenkins commented Nov 14, 2014

jerryshao commented Nov 14, 2014

SparkQA commented Nov 14, 2014

SparkQA commented Nov 14, 2014

AmplabJenkins commented Nov 14, 2014

tdas commented Nov 14, 2014

SparkQA commented Nov 14, 2014

SparkQA commented Nov 14, 2014

AmplabJenkins commented Nov 14, 2014

tdas commented Nov 14, 2014

jerryshao commented Nov 15, 2014