source-kinesis: don't try to make more than 5 requests per second

Kinesis has a 5 transactions per second limit for a single shard, and a
GetRecords call counts against that.

Previously the backoff & retry logic of the SDK was used exclusively for
managing this, but that's not ideal since the retry behavior is rather
aggressive, and any time you exceed the provisioned throughput the server will
close the connection. This was actually causing a lot of connection churn and
occasional DNS resolution failures.

The fix here is to use a client-side rate limiter that respects the 5 TPS limit.
Experimentally I have observed that this nearly eliminates the connection churn,
and should improve the stability of the connector.

Author: williamhbaker

source-kinesis/capture.go

@@ -14,6 +14,7 @@ import (
 	log "github.com/sirupsen/logrus"
 	"golang.org/x/exp/maps"
 	"golang.org/x/sync/errgroup"
+	"golang.org/x/time/rate"
 )
 
 type capture struct {
@@ -345,8 +346,13 @@ func (c *capture) readShard(
 		break
 	}
 
+	// Respect the kinesis 5 TPS rate limit.
+	limiter := rate.NewLimiter(rate.Every(time.Second), 5)
 	var didLogNoData bool
 	for {
+		if err := limiter.Wait(ctx); err != nil {
+			return fmt.Errorf("waiting for rate limiter: %w", err)
+		}
 		res, err := c.client.GetRecords(ctx, &kinesis.GetRecordsInput{
 			ShardIterator: iterator,
 			StreamARN:     &stream.arn,
@@ -375,19 +381,11 @@ func (c *capture) readShard(
 		if *res.MillisBehindLatest != 0 && len(res.Records) == 0 {
 			ll.WithField("MillisBehindLatest", *res.MillisBehindLatest).Info("shard is not caught up but returned no new data")
 			didLogNoData = true
-		}
-
-		if *res.MillisBehindLatest == 0 && len(res.Records) == 0 {
+		} else if *res.MillisBehindLatest == 0 && len(res.Records) == 0 {
 			if didLogNoData {
 				ll.Info("shard is caught up")
 				didLogNoData = false
 			}
-			select {
-			case <-ctx.Done():
-				return ctx.Err()
-			case <-time.After(1 * time.Second):
-				// Small delay to avoid hot-looping on a shard with no new data.
-			}
 		}
 	}
 }

source-kinesis/connection.go

@@ -7,12 +7,12 @@ import (
 	"slices"
 	"sync"
 
+	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/aws/ratelimit"
 	"github.com/aws/aws-sdk-go-v2/aws/retry"
 	"github.com/aws/aws-sdk-go-v2/service/kinesis"
 	"golang.org/x/sync/errgroup"
 
-	"github.com/aws/aws-sdk-go-v2/aws"
 	awsConfig "github.com/aws/aws-sdk-go-v2/config"
 	"github.com/aws/aws-sdk-go-v2/credentials"
 )