ProvisionedThroughputExceededException Error

[chenyin0126]

We have 2 pods, each pod is a consumer and relying on this library. We used .Scan to consume from all shards. We have 8 Kinesis shards, but we meet this error when 2 consumers runs together. We got

shard shardId-000000000010 error: get records error: operation error Kinesis: GetRecords, exceeded maximum number of attempts, 3, https response error StatusCode: 400, RequestID: f450bd05-0d8a-739b-aa98-773a29ee48de, ProvisionedThroughputExceededException: Rate exceeded for Shard - 715119903224/command-data-stream/shardId-000000000010

I think the reason would be the consumer get records every 0.25 seconds as default, but one shard at AWS only allow 5 transactions per second. When 2 consumers works at the same time, both of them scan all shards, so it might happen that one shard got getRecords for 8 times per second, so it raises this error: ProvisionedThroughputExceededException

Any idea what is the best way to solve this?

[chenyin0126]

cc: @harlow

[luanruisong]

mayby there is something wrong in func isRetriableError

error log

shard {shardId} error: get records error: operation error Kinesis: GetRecords, exceeded maximum number of attempts, 3, https response error StatusCode: 400, RequestID: {requestId}, ProvisionedThroughputExceededException: Rate exceeded for Shard - {shard}

func isRetriableError(err error) bool {
	switch err.(type) {
	case *types.ExpiredIteratorException:
		return true
	case *types.ProvisionedThroughputExceededException:
		return true
	}
	return false
}

func Scan got error ProvisionedThroughputExceededException but this error maybe retry?

https://github.com/harlow/kinesis-consumer/blob/master/consumer.go#L188

i think it want to retry (wait the ticker),but it is finish the goroutine

maybe should change to

        if oe := (*types.ProvisionedThroughputExceededException)(nil); errors.As(err, &oe) {
		return true
	}

[luanruisong]

BTW
there are some problems in commit: 6720a01
It starts the goroutine repeatedly to process the same shard until ProvisionedThroughputExceededException

error log (There are only two shards)

{"level":"info","caller":"/kinesis/logger.go:22","ts":1718213184044,"msg":"[CONSUMER] start scan: {shardId} 49647579764499847879996118179013785811305901676174508770"}
{"level":"info","caller":"/kinesis/logger.go:22","ts":1718213184046,"msg":"[CONSUMER] start scan: {shardId} 49651624479223759978416358002305044641501698271515509490"}
{"level":"info","caller":"/kinesis/logger.go:22","ts":1718213214045,"msg":"[CONSUMER] start scan: {shardId} 49651624479223759978416358045822747370169506734998029042"}
{"level":"info","caller":"/kinesis/logger.go:22","ts":1718213214045,"msg":"[CONSUMER] start scan: {shardId} 49647579764499847879996118207531136970195391339902796514"}
{"level":"info","caller":"/kinesis/logger.go:22","ts":1718213244044,"msg":"[CONSUMER] start scan: {shardId} 49647579764499847879996118236082338052034090689242333922"}
{"level":"info","caller":"/kinesis/logger.go:22","ts":1718213244046,"msg":"[CONSUMER] start scan: {shardId} 49651624479223759978416358091472995244637521131381719794"}
{"level":"info","caller":"/kinesis/logger.go:22","ts":1718213274045,"msg":"[CONSUMER] start scan: {shardId} 49647579764499847879996118265124363016636329277354148578"}
{"level":"info","caller":"/kinesis/logger.go:22","ts":1718213274046,"msg":"[CONSUMER] start scan: {shardId} 49651624479223759978416358130257753389514056238450606834"}
{"level":"info","caller":"/kinesis/logger.go:22","ts":1718213285292,"msg":"[CONSUMER] get records error: operation error Kinesis: GetRecords, exceeded maximum number of attempts, 3, https response error StatusCode: 400, RequestID: {requestId}, ProvisionedThroughputExceededException: Rate exceeded for Shard {{shardId}}"}
{"level":"info","caller":"/kinesis/logger.go:22","ts":1718213288257,"msg":"[CONSUMER] get records error: operation error Kinesis: GetRecords, exceeded maximum number of attempts, 3, https response error StatusCode: 400, RequestID: {requestId}, ProvisionedThroughputExceededException: Rate exceeded for Shard {{shardId}}"}
{"level":"info","caller":"/kinesis/logger.go:22","ts":1718213294575,"msg":"[CONSUMER] get records error: operation error Kinesis: GetRecords, exceeded maximum number of attempts, 3, https response error StatusCode: 400, RequestID: {requestId}, ProvisionedThroughputExceededException: Rate exceeded for Shard {{shardId}}"}
{"level":"info","caller":"/kinesis/logger.go:22","ts":1718213298295,"msg":"[CONSUMER] get records error: operation error Kinesis: GetRecords, exceeded maximum number of attempts, 3, https response error StatusCode: 400, RequestID: {requestId}, ProvisionedThroughputExceededException: Rate exceeded for Shard {{shardId}}"}

i think the problems in allgroup.go:121

on ticker channel triggered, it is also put the on runing shard into the channel shardc(no parentShard) in the second loop

	for _, shard := range shards {
		if _, ok := g.shards[*shard.ShardId]; ok {
			continue
		}
		g.shards[*shard.ShardId] = shard
		g.shardsClosed[*shard.ShardId] = make(chan struct{})
	}
	for _, shard := range shards {
		shard := shard // Shadow shard, since we use it in goroutine
		var parent1, parent2 <-chan struct{}
		if shard.ParentShardId != nil {
			parent1 = g.shardsClosed[*shard.ParentShardId]
		}
		if shard.AdjacentParentShardId != nil {
			parent2 = g.shardsClosed[*shard.AdjacentParentShardId]
		}
		go func() {
			// Asynchronously wait for all parents of this shard to be processed
			// before providing it out to our client.  Kinesis guarantees that a
			// given partition key's data will be provided to clients in-order,
			// but when splits or joins happen, we need to process all parents prior
			// to processing children or that ordering guarantee is not maintained.
			if waitForCloseChannel(ctx, parent1) && waitForCloseChannel(ctx, parent2) {
				shardc <- shard
			}
		}()
	}

[mskonovalov]

Fix #161

[luanruisong]

good job!