Detect checkpoint seqnum pointing to expired data #126
Conversation
| }, | ||
| ) | ||
| if err != nil { | ||
| if err.(awserr.Error).Code() != "AccessDeniedException" { |
There was a problem hiding this comment.
There is no constant for this that I could find. kinesis.ErrCodeAccessDeniedException isn't a thing, and while this constant is defined in other packages, I didn't think it made sense to import one of them just for that.
| if err.(awserr.Error).Code() != "AccessDeniedException" { | ||
| return err | ||
| } | ||
| c.logger.Log("[CONSUMER] IAM entity lacks kinesis:DescribeStreamSummary permissions, skipping extra sanity checks") |
There was a problem hiding this comment.
Existing users may have IAM policies crafted to work with kinesis-consumer that don't allow this permission, so it's optional.
| @@ -228,7 +270,7 @@ func isShardClosed(nextShardIterator, currentShardIterator *string) bool { | |||
| return nextShardIterator == nil || currentShardIterator == nextShardIterator | |||
| } | |||
|
|
|||
| func (c *Consumer) getShardIterator(ctx context.Context, streamName, shardID, seqNum string) (*string, error) { | |||
| func (c *Consumer) getShardIterator(ctx context.Context, streamName, shardID, seqNum string, oldest bool) (*string, error) { | |||
| params := &kinesis.GetShardIteratorInput{ | |||
| ShardId: aws.String(shardID), | |||
| StreamName: aws.String(streamName), | |||
There was a problem hiding this comment.
Could we reference c.streamName here instead and get rid of the streamName argument to this function?
|
|
||
| for { | ||
| nextTicker := scanTicker | ||
| resp, err := c.client.GetRecords(&kinesis.GetRecordsInput{ |
There was a problem hiding this comment.
BTW, should we be using GetRecordsWithContext here?
|
Hi! What's the status of this? It hasn't been merge right? I am running into the same issue |
Implementation for #125 for consideration/discussion.
CI will fail but I can add a mock for
DescribeStreamSummaryWithContextif you think this is worthy of merging.This also adds an extra ticker
fastTickerwhich is based on the 5TPS rate limit, and which is used for those cases where we want a fast return back to Kinesis. I can amend #122 to use this as well, if/when the time comes. Alternatively, to squeeze more throughput, we could use the behavior from #122 (continue immediately) and only start backing off if we hitErrCodeProvisionedThroughputExceededException.