[MTCES-1264] Fetch upstream updates (#3)

magiusdarrigo · mrmonaghan · calebstewart · web-flow · commit d32b83b41a1f · 2023-06-20T15:33:20.000-04:00
* fixing infinite worker loop Signed-off-by: Mike Monaghan <mike_monaghan@live.ca> * Automatically resolve default KinesisEndpoint This commit fixes vmware-archive#5 by returning `aws.EndpointNotFoundError` from the endpoint resolver when no `KinesisEndpoint` is defined, which will resolve the default AWS endpoint. This is the same process used by the DynamoDB checkpointer to resolve the default endpoint. Signed-off-by: Caleb Stewart <caleb.stewart94@gmail.com> * fix: catch DynamoDB Scan error when trying to scan nonexistent table/index in syncLeases() Signed-off-by: Shiva Pentakota <spentakota@vmware.com> * chore: Adding periods to copyright comment to satisfy gofmt Signed-off-by: Shiva Pentakota <spentakota@vmware.com> * feat: Sending renewed lease metric Signed-off-by: Shiva Pentakota <spentakota@vmware.com> * fix: add DeleteMetricMillisBehindLatest for error case Signed-off-by: Shiva Pentakota <spentakota@vmware.com> * Refactor in prep for testing rate limiting improvements Signed-off-by: John Calixto <jcalixto@vmware.com> * fix: add getRecords TPS rate limiting Signed-off-by: Shiva Pentakota <spentakota@vmware.com> * fix: add hard cap maxRetries for getRecord errors Signed-off-by: Shiva Pentakota <spentakota@vmware.com> * fix: add maxBytes per second getRecord check Signed-off-by: Shiva Pentakota <spentakota@vmware.com> * chore: log RemoveLeaseOwner errors with debug instead of error Signed-off-by: Shiva Pentakota <spentakota@vmware.com> * fix: add check for GetRecords error within callGetRecordsAPI Signed-off-by: Shiva Pentakota <spentakota@vmware.com> * fix: use nanosecond precision in lease comparisons Signed-off-by: Shiva Pentakota <spentakota@vmware.com> * chore: add info logs in sleep case for kinesis backoff errors Signed-off-by: Shiva Pentakota <spentakota@vmware.com> * fix: Check token bucket corner cases correctly. Signed-off-by: John Calixto <jcalixto@vmware.com> * Bump github.com/prometheus/client_golang from 1.11.0 to 1.11.1 Bumps [github.com/prometheus/client_golang](https://github.com/prometheus/client_golang) from 1.11.0 to 1.11.1. - [Release notes](https://github.com/prometheus/client_golang/releases) - [Changelog](https://github.com/prometheus/client_golang/blob/main/CHANGELOG.md) - [Commits](prometheus/client_golang@v1.11.0...v1.11.1) --- updated-dependencies: - dependency-name: github.com/prometheus/client_golang dependency-type: direct:production ... Signed-off-by: dependabot[bot] <support@github.com> * Bump golang.org/x/sys from 0.0.0-20211216021012-1d35b9e2eb4e to 0.1.0 Bumps [golang.org/x/sys](https://github.com/golang/sys) from 0.0.0-20211216021012-1d35b9e2eb4e to 0.1.0. - [Release notes](https://github.com/golang/sys/releases) - [Commits](https://github.com/golang/sys/commits/v0.1.0) --- updated-dependencies: - dependency-name: golang.org/x/sys dependency-type: indirect ... Signed-off-by: dependabot[bot] <support@github.com> * fix: add shutdown and leaseExpired error cases for checkpoint function Signed-off-by: Shiva Pentakota <spentakota@vmware.com> * feat: make lease renewal async Signed-off-by: Shiva Pentakota <spentakota@vmware.com> * fix: return err log in case of ErrLeaseNotAcquired Signed-off-by: Shiva Pentakota <spentakota@vmware.com> * chore: Remove extraneous err check After checking the scan result above this line, checking err here no longer has any effect. Signed-off-by: John Calixto <jcalixto@vmware.com> * fix: pass in ctx with cancel for renewLease Signed-off-by: Shiva Pentakota <spentakota@vmware.com> * ran go mod tidy --------- Signed-off-by: Mike Monaghan <mike_monaghan@live.ca> Signed-off-by: Caleb Stewart <caleb.stewart94@gmail.com> Signed-off-by: Shiva Pentakota <spentakota@vmware.com> Signed-off-by: John Calixto <jcalixto@vmware.com> Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: Mike Monaghan <mike_monaghan@live.ca> Co-authored-by: Caleb Stewart <caleb.stewart94@gmail.com> Co-authored-by: Shiva Pentakota <spentakota@vmware.com> Co-authored-by: spentakota <120056013+spentakota@users.noreply.github.com> Co-authored-by: John Calixto <jcalixto@vmware.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: cmckelvey-vmware <85366153+cmckelvey-vmware@users.noreply.github.com> Co-authored-by: vmwjc <108959326+vmwjc@users.noreply.github.com>
diff --git a/clientlibrary/checkpoint/checkpointer.go b/clientlibrary/checkpoint/checkpointer.go
@@ -20,7 +20,7 @@
 // Package checkpoint
 // The implementation is derived from https://github.com/patrobinson/gokini
 //
-// # Copyright 2018 Patrick robinson
+// Copyright 2018 Patrick robinson.
 //
 // Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
 //
@@ -79,6 +79,9 @@ type Checkpointer interface {
 	// RemoveLeaseOwner to remove lease owner for the shard entry to make the shard available for reassignment
 	RemoveLeaseOwner(string) error
 
+	// GetLeaseOwner to get current owner of lease for shard
+	GetLeaseOwner(string) (string, error)
+
 	// ListActiveWorkers returns active workers and their shards (New Lease Stealing Methods)
 	ListActiveWorkers(map[string]*par.ShardStatus) (map[string][]*par.ShardStatus, error)
 
diff --git a/clientlibrary/checkpoint/dynamodb-checkpointer.go b/clientlibrary/checkpoint/dynamodb-checkpointer.go
@@ -20,7 +20,7 @@
 // Package checkpoint
 // The implementation is derived from https://github.com/patrobinson/gokini
 //
-// # Copyright 2018 Patrick robinson
+// Copyright 2018 Patrick robinson.
 //
 // Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
 //
@@ -51,6 +51,10 @@ const (
 	NumMaxRetries = 10
 )
 
+var (
+	NoLeaseOwnerErr = errors.New("no LeaseOwner in checkpoints table")
+)
+
 // DynamoCheckpoint implements the Checkpoint interface using DynamoDB as a backend
 type DynamoCheckpoint struct {
 	log                     logger.Logger
@@ -129,7 +133,7 @@ func (checkpointer *DynamoCheckpoint) Init() error {
 // GetLease attempts to gain a lock on the given shard
 func (checkpointer *DynamoCheckpoint) GetLease(shard *par.ShardStatus, newAssignTo string) error {
 	newLeaseTimeout := time.Now().Add(time.Duration(checkpointer.LeaseDuration) * time.Millisecond).UTC()
-	newLeaseTimeoutString := newLeaseTimeout.Format(time.RFC3339)
+	newLeaseTimeoutString := newLeaseTimeout.Format(time.RFC3339Nano)
 	currentCheckpoint, err := checkpointer.getItem(shard.ID)
 	if err != nil {
 		return err
@@ -161,7 +165,7 @@ func (checkpointer *DynamoCheckpoint) GetLease(shard *par.ShardStatus, newAssign
 		assignedTo := assignedVar.(*types.AttributeValueMemberS).Value
 		leaseTimeout := leaseVar.(*types.AttributeValueMemberS).Value
 
-		currentLeaseTimeout, err := time.Parse(time.RFC3339, leaseTimeout)
+		currentLeaseTimeout, err := time.Parse(time.RFC3339Nano, leaseTimeout)
 		if err != nil {
 			return err
 		}
@@ -246,7 +250,7 @@ func (checkpointer *DynamoCheckpoint) GetLease(shard *par.ShardStatus, newAssign
 
 // CheckpointSequence writes a checkpoint at the designated sequence ID
 func (checkpointer *DynamoCheckpoint) CheckpointSequence(shard *par.ShardStatus) error {
-	leaseTimeout := shard.GetLeaseTimeout().UTC().Format(time.RFC3339)
+	leaseTimeout := shard.GetLeaseTimeout().UTC().Format(time.RFC3339Nano)
 	marshalledCheckpoint := map[string]types.AttributeValue{
 		LeaseKeyKey: &types.AttributeValueMemberS{
 			Value: shard.ID,
@@ -290,7 +294,7 @@ func (checkpointer *DynamoCheckpoint) FetchCheckpoint(shard *par.ShardStatus) er
 
 	// Use up-to-date leaseTimeout to avoid ConditionalCheckFailedException when claiming
 	if leaseTimeout, ok := checkpoint[LeaseTimeoutKey]; ok && leaseTimeout.(*types.AttributeValueMemberS).Value != "" {
-		currentLeaseTimeout, err := time.Parse(time.RFC3339, leaseTimeout.(*types.AttributeValueMemberS).Value)
+		currentLeaseTimeout, err := time.Parse(time.RFC3339Nano, leaseTimeout.(*types.AttributeValueMemberS).Value)
 		if err != nil {
 			return err
 		}
@@ -336,6 +340,23 @@ func (checkpointer *DynamoCheckpoint) RemoveLeaseOwner(shardID string) error {
 	return err
 }
 
+// GetLeaseOwner returns current lease owner of given shard in checkpoints table
+func (checkpointer *DynamoCheckpoint) GetLeaseOwner(shardID string) (string, error) {
+	currentCheckpoint, err := checkpointer.getItem(shardID)
+	if err != nil {
+		return "", err
+	}
+
+	assignedVar, assignedToOk := currentCheckpoint[LeaseOwnerKey]
+
+	if !assignedToOk {
+		return "", NoLeaseOwnerErr
+	}
+
+	return assignedVar.(*types.AttributeValueMemberS).Value, nil
+
+}
+
 // ListActiveWorkers returns a map of workers and their shards
 func (checkpointer *DynamoCheckpoint) ListActiveWorkers(shardStatus map[string]*par.ShardStatus) (map[string][]*par.ShardStatus, error) {
 	err := checkpointer.syncLeases(shardStatus)
@@ -370,7 +391,7 @@ func (checkpointer *DynamoCheckpoint) ClaimShard(shard *par.ShardStatus, claimID
 	if err != nil && err != ErrSequenceIDNotFound {
 		return err
 	}
-	leaseTimeoutString := shard.GetLeaseTimeout().Format(time.RFC3339)
+	leaseTimeoutString := shard.GetLeaseTimeout().Format(time.RFC3339Nano)
 
 	conditionalExpression := `ShardID = :id AND LeaseTimeout = :lease_timeout AND attribute_not_exists(ClaimRequest)`
 	expressionAttributeValues := map[string]types.AttributeValue{
@@ -441,6 +462,12 @@ func (checkpointer *DynamoCheckpoint) syncLeases(shardStatus map[string]*par.Sha
 	}
 
 	scanOutput, err := checkpointer.svc.Scan(context.TODO(), input)
+
+	if err != nil {
+		log.Debugf("Error performing DynamoDB Scan. Error: %+v ", err)
+		return err
+	}
+
 	results := scanOutput.Items
 	for _, result := range results {
 		shardId, foundShardId := result[LeaseKeyKey]
@@ -456,10 +483,6 @@ func (checkpointer *DynamoCheckpoint) syncLeases(shardStatus map[string]*par.Sha
 		}
 	}
 
-	if err != nil {
-		log.Debugf("Error performing SyncLeases. Error: %+v ", err)
-		return err
-	}
 	log.Debugf("Lease sync completed. Next lease sync will occur in %s", time.Duration(checkpointer.kclConfig.LeaseSyncingTimeIntervalMillis)*time.Millisecond)
 	return nil
 }
diff --git a/clientlibrary/checkpoint/mock-dynamodb_test.go b/clientlibrary/checkpoint/mock-dynamodb_test.go
@@ -19,7 +19,7 @@
 
 // The implementation is derived from https://github.com/patrobinson/gokini
 //
-// Copyright 2018 Patrick robinson
+// Copyright 2018 Patrick robinson.
 //
 // Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
 //
diff --git a/clientlibrary/config/config.go b/clientlibrary/config/config.go
@@ -69,6 +69,9 @@ const (
 	// DefaultLeaseRefreshPeriodMillis Period before the end of lease during which a lease is refreshed by the owner.
 	DefaultLeaseRefreshPeriodMillis = 5000
 
+	// DefaultLeaseRefreshWaitTime is the period of time to wait before async lease renewal attempt
+	DefaultLeaseRefreshWaitTime = 2500
+
 	// DefaultMaxRecords Max records to fetch from Kinesis in a single GetRecords call.
 	DefaultMaxRecords = 10000
 
@@ -136,6 +139,9 @@ const (
 
 	// DefaultLeaseSyncingIntervalMillis Number of milliseconds to wait before syncing with lease table (dynamodDB)
 	DefaultLeaseSyncingIntervalMillis = 60000
+
+	// DefaultMaxRetryCount The default maximum number of retries in case of error
+	DefaultMaxRetryCount = 5
 )
 
 type (
@@ -213,6 +219,9 @@ type (
 		// LeaseRefreshPeriodMillis is the period before the end of lease during which a lease is refreshed by the owner.
 		LeaseRefreshPeriodMillis int
 
+		// LeaseRefreshWaitTime is the period of time to wait before async lease renewal attempt
+		LeaseRefreshWaitTime int
+
 		// MaxRecords Max records to read per Kinesis getRecords() call
 		MaxRecords int
 
@@ -283,6 +292,9 @@ type (
 
 		// LeaseSyncingTimeInterval The number of milliseconds to wait before syncing with lease table (dynamoDB)
 		LeaseSyncingTimeIntervalMillis int
+
+		// MaxRetryCount The maximum number of retries in case of error
+		MaxRetryCount int
 	}
 )
 
diff --git a/clientlibrary/config/kcl-config.go b/clientlibrary/config/kcl-config.go
@@ -102,6 +102,8 @@ func NewKinesisClientLibConfigWithCredentials(applicationName, streamName, regio
 		LeaseStealingIntervalMillis:                      DefaultLeaseStealingIntervalMillis,
 		LeaseStealingClaimTimeoutMillis:                  DefaultLeaseStealingClaimTimeoutMillis,
 		LeaseSyncingTimeIntervalMillis:                   DefaultLeaseSyncingIntervalMillis,
+		LeaseRefreshWaitTime:                             DefaultLeaseRefreshWaitTime,
+		MaxRetryCount:                                    DefaultMaxRetryCount,
 		Logger:                                           logger.GetDefaultLogger(),
 	}
 }
@@ -148,6 +150,12 @@ func (c *KinesisClientLibConfiguration) WithLeaseRefreshPeriodMillis(leaseRefres
 	return c
 }
 
+func (c *KinesisClientLibConfiguration) WithLeaseRefreshWaitTime(leaseRefreshWaitTime int) *KinesisClientLibConfiguration {
+	checkIsValuePositive("LeaseRefreshWaitTime", leaseRefreshWaitTime)
+	c.LeaseRefreshWaitTime = leaseRefreshWaitTime
+	return c
+}
+
 func (c *KinesisClientLibConfiguration) WithShardSyncIntervalMillis(shardSyncIntervalMillis int) *KinesisClientLibConfiguration {
 	checkIsValuePositive("ShardSyncIntervalMillis", shardSyncIntervalMillis)
 	c.ShardSyncIntervalMillis = shardSyncIntervalMillis
@@ -211,6 +219,13 @@ func (c *KinesisClientLibConfiguration) WithLogger(logger logger.Logger) *Kinesi
 	return c
 }
 
+// WithMaxRetryCount sets the max retry count in case of error.
+func (c *KinesisClientLibConfiguration) WithMaxRetryCount(maxRetryCount int) *KinesisClientLibConfiguration {
+	checkIsValuePositive("maxRetryCount", maxRetryCount)
+	c.MaxRetryCount = maxRetryCount
+	return c
+}
+
 // WithMonitoringService sets the monitoring service to use to publish metrics.
 func (c *KinesisClientLibConfiguration) WithMonitoringService(mService metrics.MonitoringService) *KinesisClientLibConfiguration {
 	// Nil case is handled downward (at worker creation) so no need to do it here.
diff --git a/clientlibrary/metrics/cloudwatch/cloudwatch.go b/clientlibrary/metrics/cloudwatch/cloudwatch.go
@@ -20,7 +20,7 @@
 // Package cloudwatch
 // The implementation is derived from https://github.com/patrobinson/gokini
 //
-// # Copyright 2018 Patrick robinson
+// Copyright 2018 Patrick robinson.
 //
 // Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
 //
diff --git a/clientlibrary/metrics/interfaces.go b/clientlibrary/metrics/interfaces.go
@@ -20,7 +20,7 @@
 // Package metrics
 // The implementation is derived from https://github.com/patrobinson/gokini
 //
-// Copyright 2018 Patrick robinson
+// Copyright 2018 Patrick robinson.
 //
 // Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
 //
@@ -35,6 +35,7 @@ type MonitoringService interface {
 	IncrRecordsProcessed(shard string, count int)
 	IncrBytesProcessed(shard string, count int64)
 	MillisBehindLatest(shard string, milliSeconds float64)
+	DeleteMetricMillisBehindLatest(shard string)
 	LeaseGained(shard string)
 	LeaseLost(shard string)
 	LeaseRenewed(shard string)
@@ -53,6 +54,7 @@ func (NoopMonitoringService) Shutdown()                 {}
 func (NoopMonitoringService) IncrRecordsProcessed(_ string, _ int)         {}
 func (NoopMonitoringService) IncrBytesProcessed(_ string, _ int64)         {}
 func (NoopMonitoringService) MillisBehindLatest(_ string, _ float64)       {}
+func (NoopMonitoringService) DeleteMetricMillisBehindLatest(_ string)      {}
 func (NoopMonitoringService) LeaseGained(_ string)                         {}
 func (NoopMonitoringService) LeaseLost(_ string)                           {}
 func (NoopMonitoringService) LeaseRenewed(_ string)                        {}
diff --git a/clientlibrary/metrics/prometheus/prometheus.go b/clientlibrary/metrics/prometheus/prometheus.go
@@ -20,7 +20,7 @@
 // Package prometheus
 // The implementation is derived from https://github.com/patrobinson/gokini
 //
-// # Copyright 2018 Patrick robinson
+// Copyright 2018 Patrick robinson.
 //
 // Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
 //
@@ -147,6 +147,10 @@ func (p *MonitoringService) MillisBehindLatest(shard string, millSeconds float64
 	p.behindLatestMillis.With(prom.Labels{"shard": shard, "kinesisStream": p.streamName}).Set(millSeconds)
 }
 
+func (p *MonitoringService) DeleteMetricMillisBehindLatest(shard string) {
+	p.behindLatestMillis.Delete(prom.Labels{"shard": shard, "kinesisStream": p.streamName})
+}
+
 func (p *MonitoringService) LeaseGained(shard string) {
 	p.leasesHeld.With(prom.Labels{"shard": shard, "kinesisStream": p.streamName, "workerID": p.workerID}).Inc()
 }
diff --git a/clientlibrary/partition/partition.go b/clientlibrary/partition/partition.go
@@ -20,7 +20,7 @@
 // Package partition
 // The implementation is derived from https://github.com/patrobinson/gokini
 //
-// # Copyright 2018 Patrick robinson
+// Copyright 2018 Patrick robinson.
 //
 // Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
 //
diff --git a/clientlibrary/worker/common-shard-consumer.go b/clientlibrary/worker/common-shard-consumer.go
@@ -21,6 +21,7 @@
 package worker
 
 import (
+	"context"
 	"sync"
 	"time"
 
@@ -40,29 +41,36 @@ type shardConsumer interface {
 	getRecords() error
 }
 
+type KinesisSubscriberGetter interface {
+	SubscribeToShard(ctx context.Context, params *kinesis.SubscribeToShardInput, optFns ...func(*kinesis.Options)) (*kinesis.SubscribeToShardOutput, error)
+	GetShardIterator(ctx context.Context, params *kinesis.GetShardIteratorInput, optFns ...func(*kinesis.Options)) (*kinesis.GetShardIteratorOutput, error)
+	GetRecords(ctx context.Context, params *kinesis.GetRecordsInput, optFns ...func(*kinesis.Options)) (*kinesis.GetRecordsOutput, error)
+}
+
 // commonShardConsumer implements common functionality for regular and enhanced fan-out consumers
 type commonShardConsumer struct {
 	shard           *par.ShardStatus
-	kc              *kinesis.Client
+	kc              KinesisSubscriberGetter
 	checkpointer    chk.Checkpointer
 	recordProcessor kcl.IRecordProcessor
 	kclConfig       *config.KinesisClientLibConfiguration
 	mService        metrics.MonitoringService
 }
 
 // Cleanup the internal lease cache
-func (sc *commonShardConsumer) releaseLease() {
+func (sc *commonShardConsumer) releaseLease(shard string) {
 	log := sc.kclConfig.Logger
 	log.Infof("Release lease for shard %s", sc.shard.ID)
 	sc.shard.SetLeaseOwner("")
 
 	// Release the lease by wiping out the lease owner for the shard
 	// Note: we don't need to do anything in case of error here and shard lease will eventually be expired.
 	if err := sc.checkpointer.RemoveLeaseOwner(sc.shard.ID); err != nil {
-		log.Errorf("Failed to release shard lease or shard: %s Error: %+v", sc.shard.ID, err)
+		log.Debugf("Failed to release shard lease or shard: %s Error: %+v", sc.shard.ID, err)
 	}
 
 	// reporting lease lose metrics
+	sc.mService.DeleteMetricMillisBehindLatest(shard)
 	sc.mService.LeaseLost(sc.shard.ID)
 }
 
@@ -165,7 +173,6 @@ func (sc *commonShardConsumer) processRecords(getRecordsStartTime time.Time, rec
 		input.CacheEntryTime = &getRecordsStartTime
 		input.CacheExitTime = &processRecordsStartTime
 		sc.recordProcessor.ProcessRecords(input)
-
 		processedRecordsTiming := time.Since(processRecordsStartTime).Milliseconds()
 		sc.mService.RecordProcessRecordsTime(sc.shard.ID, float64(processedRecordsTiming))
 	}
diff --git a/clientlibrary/worker/fan-out-shard-consumer.go b/clientlibrary/worker/fan-out-shard-consumer.go
@@ -46,7 +46,7 @@ type FanOutShardConsumer struct {
 // getRecords subscribes to a shard and reads events from it.
 // Precondition: it currently has the lease on the shard.
 func (sc *FanOutShardConsumer) getRecords() error {
-	defer sc.releaseLease()
+	defer sc.releaseLease(sc.shard.ID)
 
 	log := sc.kclConfig.Logger
 
@@ -103,6 +103,8 @@ func (sc *FanOutShardConsumer) getRecords() error {
 				return err
 			}
 			refreshLeaseTimer = time.After(time.Until(sc.shard.LeaseTimeout.Add(-time.Duration(sc.kclConfig.LeaseRefreshPeriodMillis) * time.Millisecond)))
+			// log metric for renewed lease for worker
+			sc.mService.LeaseRenewed(sc.shard.ID)
 		case event, ok := <-shardSub.GetStream().Events():
 			if !ok {
 				// need to resubscribe to shard
diff --git a/clientlibrary/worker/polling-shard-consumer.go b/clientlibrary/worker/polling-shard-consumer.go
diff --git a/clientlibrary/worker/polling-shard-consumer_test.go b/clientlibrary/worker/polling-shard-consumer_test.go
diff --git a/clientlibrary/worker/worker.go b/clientlibrary/worker/worker.go
diff --git a/go.mod b/go.mod
diff --git a/go.sum b/go.sum
