Skip to content

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

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
jerryshao wants to merge 23 commits into from
Closed

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

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
23 commits
Select commit Hold shift + click to select a range
dd9aeeb
Initial commit for reliable Kafka receiver
jerryshao Oct 23, 2014
77c3e50
Code refactor and add some unit tests
jerryshao Oct 27, 2014
0894aef
Add some comments
jerryshao Oct 28, 2014
16bfe78
Change the ordering of imports
jerryshao Oct 28, 2014
a949741
Address the comments
jerryshao Oct 29, 2014
8135d31
Fix flaky test
jerryshao Oct 29, 2014
96c7a1d
Update the ReliableKafkaReceiver unit test
jerryshao Oct 31, 2014
4854ee9
Address all the comments
jerryshao Nov 10, 2014
98f3d07
Fix comment style
jerryshao Nov 10, 2014
ea873e4
Further address the comments
jerryshao Nov 12, 2014
e5e21c1
Change to while loop
jerryshao Nov 12, 2014
b798535
Fix the missed issue
jerryshao Nov 12, 2014
e501b3c
Add Mima excludes
jerryshao Nov 12, 2014
b2b2f84
Refactored Kafka receiver logic and Kafka testsuites
tdas Nov 13, 2014
9f636b3
Merge pull request #5 from tdas/kafka-refactor
jerryshao Nov 13, 2014
2a20a01
Address some comments
jerryshao Nov 13, 2014
ec2e95e
Removed the receiver's locks and essentially reverted to Saisai's ori…
tdas Nov 13, 2014
d9a452c
Minor updates.
tdas Nov 13, 2014
14630aa
Minor updates.
tdas Nov 13, 2014
149948b
Fixed mistake
tdas Nov 13, 2014
fab14c7
minor update.
tdas Nov 13, 2014
eae4ad6
Refectored KafkaStreamSuiteBased to eliminate KafkaTestUtils and made…
tdas Nov 13, 2014
5461f1c
Merge pull request #8 from tdas/kafka-refactor3
jerryshao Nov 14, 2014
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
Prev Previous commit
Next Next commit
Change to while loop
  • Loading branch information
@jerryshao
jerryshao committed Nov 12, 2014
commit e5e21c1b54222cc5595e42933f739ca3f078ed36
4 changes: 3 additions & 1 deletion external/kafka/src/main/scala/org/apache/spark/streaming/kafka/KafkaInputDStream.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -128,7 +128,9 @@ class KafkaReceiver[
def run() {
logInfo("Starting MessageHandler.")
try {
for (msgAndMetadata <- stream) {
val streamIterator = stream.iterator()
while (streamIterator.hasNext()) {
val msgAndMetadata = streamIterator.next()
store((msgAndMetadata.key, msgAndMetadata.message))
}
} catch {
Expand Down
4 changes: 3 additions & 1 deletion external/kafka/src/main/scala/org/apache/spark/streaming/kafka/ReliableKafkaReceiver.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -206,7 +206,9 @@ class ReliableKafkaReceiver[
override def run(): Unit = {
logInfo(s"Starting message process thread ${Thread.currentThread().getId}.")
try {
for (msgAndMetadata <- stream) {
val streamIterator = stream.iterator()
while (streamIterator.hasNext()) {
val msgAndMetadata = streamIterator.next()
val topicAndPartition = TopicAndPartition(
msgAndMetadata.topic, msgAndMetadata.partition)
blockGenerator.synchronized {
Expand Down