Skip to content

[SPARK-19304] [Streaming] [Kinesis] fix kinesis slow checkpoint recovery #16842

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
Gauravshah wants to merge 6 commits into from
Closed

[SPARK-19304] [Streaming] [Kinesis] fix kinesis slow checkpoint recovery #16842

Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
b5e544a
added limit to kinesis checkpoint backed rdd to reduce number of reco…
Gauravshah Feb 7, 2017
d3b8c62
fixing test cases on b5e544a8ec326149b7d03773dd7abf8703ee44a2
Gauravshah Feb 8, 2017
274ee27
limiting max records limit to 10k for aws limitations
Gauravshah Feb 22, 2017
4f5edd3
limiting max records limit to 10k for aws limitations
Gauravshah Feb 28, 2017
82499bc
scala coding style fixes, one line per param
Gauravshah Mar 1, 2017
c8efdcf
scala coding style fixes, one line per param
Gauravshah Mar 4, 2017
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
4 changes: 3 additions & 1 deletion ...kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisBackedBlockRDD.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -139,6 +139,8 @@ class KinesisSequenceRangeIterator(
private val client = new AmazonKinesisClient(credentials)
private val streamName = range.streamName
private val shardId = range.shardId
// AWS limits to maximum of 10k records per get call
private val maxGetRecordsLimit = 10000

private var toSeqNumberReceived = false
private var lastSeqNumber: String = null
Expand Down Expand Up @@ -212,7 +214,7 @@ class KinesisSequenceRangeIterator(
val getRecordsRequest = new GetRecordsRequest
getRecordsRequest.setRequestCredentials(credentials)
getRecordsRequest.setShardIterator(shardIterator)
getRecordsRequest.setLimit(recordCount)
getRecordsRequest.setLimit(Math.max(recordCount, this.maxGetRecordsLimit))
Copy link
Contributor

@brkyvz brkyvz Feb 27, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this should be a min not a max

Copy link
Contributor Author

@Gauravshah Gauravshah Feb 28, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

👍

val getRecordsResult = retryOrTimeout[GetRecordsResult](
s"getting records using shard iterator") {
client.getRecords(getRecordsRequest)
Expand Down