luciano/kinesis-consumer
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

if there are no records available, keep going until the nextiterator returns nothing

Browse source
This commit is contained in:
dan 2015-04-07 13:42:50 -07:00
parent 2ff748a2d4
commit 969ba18824
1 changed files with 7 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
pipeline.go
View file

@ -33,7 +33,7 @@ var pipelineRecoverableErrorCodes = map[string]bool{
func (p Pipeline) isRecoverableError(err error) bool { func (p Pipeline) isRecoverableError(err error) bool {
r := false r := false
l4g.Debug("isRecoverableError, type %s, value (+%v)", reflect.TypeOf(err).Name(), err) log.Printf("isRecoverableError, type %s, value (+%v)\n", reflect.TypeOf(err).String(), err)
cErr, ok := err.(*kinesis.Error) cErr, ok := err.(*kinesis.Error)
if ok && pipelineRecoverableErrorCodes[cErr.Code] == true { if ok && pipelineRecoverableErrorCodes[cErr.Code] == true {
@ -48,8 +48,11 @@ func (p Pipeline) handleAwsWaitTimeExp(attempts int) {
//http://docs.aws.amazon.com/general/latest/gr/api-retries.html //http://docs.aws.amazon.com/general/latest/gr/api-retries.html
// wait up to 5 minutes based on the aws exponential backoff algorithm // wait up to 5 minutes based on the aws exponential backoff algorithm
waitTime := time.Duration(math.Min(100*math.Pow(2, float64(attempts)), 300000)) * time.Millisecond if attempts > 0 {
time.Sleep(waitTime) waitTime := time.Duration(math.Min(100*math.Pow(2, float64(attempts)), 300000)) * time.Millisecond
l4g.Finest("handleAwsWaitTimeExp:%s", waitTime.String())
time.Sleep(waitTime)
}
} }
@ -119,7 +122,7 @@ func (p Pipeline) ProcessShard(ksis *kinesis.Kinesis, shardID string) {
l4g.Error("NextShardIterator ERROR: %v", err) l4g.Error("NextShardIterator ERROR: %v", err)
break break
} else { } else {
time.Sleep(5 * time.Second) //time.Sleep(5 * time.Second)
} }
if p.Buffer.ShouldFlush() { if p.Buffer.ShouldFlush() {