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

use map for lookup of retriable errors

Browse source
This commit is contained in:
Harlow Ward 2019-07-30 16:44:01 -07:00
parent 49b9bfa4e7
commit 1df9189517
1 changed files with 6 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
consumer.go
View file

@ -151,7 +151,7 @@ func (c *Consumer) ScanShard(ctx context.Context, shardID string, fn ScanFunc) e
c.logger.Log("[CONSUMER] get records error:", err.Error()) c.logger.Log("[CONSUMER] get records error:", err.Error())
if awserr, ok := err.(awserr.Error); ok { if awserr, ok := err.(awserr.Error); ok {
if awserr.Code() != kinesis.ErrCodeExpiredIteratorException { if _, ok := retriableErrors[awserr.Code()]; !ok {
return fmt.Errorf("get records error: %v", awserr.Message()) return fmt.Errorf("get records error: %v", awserr.Message())
} }
} }
@ -196,6 +196,11 @@ func (c *Consumer) ScanShard(ctx context.Context, shardID string, fn ScanFunc) e
} }
} }
var retriableErrors = map[string]struct{}{
kinesis.ErrCodeExpiredIteratorException: struct{}{},
kinesis.ErrCodeProvisionedThroughputExceededException: struct{}{},
}
func isShardClosed(nextShardIterator, currentShardIterator *string) bool { func isShardClosed(nextShardIterator, currentShardIterator *string) bool {
return nextShardIterator == nil || currentShardIterator == nextShardIterator return nextShardIterator == nil || currentShardIterator == nextShardIterator
} }