Open bamos opened 10 years ago
cloudify
commented
10 years ago Mmh this is weird, I haven't seen this issue yet, have you tried using different horizons? I'm actually not using Kinesis in production yet, so I haven't done any proper testing. The whole shard iterator thing for fetching the messages is not super super clear in the docs, I may have made a mistake in the fetch logic...
bamos
commented
10 years ago Hi Federico, I've found a solution to this. Instead of getting records by only the first shard iterator, an application should continuously use the nextIterator and poll for new records. The following code is working well for me.
// Initialize the shard iterators.
val initialShardIteratorsRequest = for {
shards <- stream.get.shards.list
initialShardIterator <- Future.sequence(shards.map {
shard => implicitExecute(shard.iterator)
})
} yield initialShardIterator
var shardIterators = Await.result(initialShardIteratorsRequest,
30.seconds).asInstanceOf[List[ShardIterator]]
// Continuously poll for records.
while (shardIterators != null) {
val recordChunksRequest = for {
recordChunkIterator <- Future.sequence(shardIterators.map {
iterator => implicitExecute(iterator.nextRecords)
})
} yield recordChunkIterator
val recordChunks = Await.result(recordChunksRequest, 30.seconds)
//println("recordChunks: " + recordChunks.toString)
val nextShardIterators = new MutableList[ShardIterator]()
for (recordChunk <- recordChunks) {
//println("==Record chunk:" + recordChunk.toString)
for (record <- recordChunk.records) {
println("sequenceNumber: " + record.sequenceNumber)
//printData(record.data.array())
println("partitionKey: " + record.partitionKey)
}
nextShardIterators += recordChunk.nextIterator
}
shardIterators = nextShardIterators.toList
Thread.sleep(1000)
}CC @alexanderdean
Hi Federico,
I've modified your example to add 10 items to Kinesis and then attempt to retrieve them 10 separate times to illustrate a bug I'm seeing in some of my other projects where not all of the records are retrieved:
Do you know if this is expected behavior from Kinesis? Am I doing something subtly wrong when retrieving the data?
Below, some attempts retrieve all 10 records, but other attempts, such as the last one, read less than 10 records.