luciano/vmware-go-kcl-v2
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge pull request #17 from vmware/spentakota/sendLeaseRenewedMetric

Browse source 
feat: Sending renewed lease metric
This commit is contained in:
spentakota 2023-01-12 11:56:21 -08:00 committed by GitHub
commit c5bc6c4ded
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 4 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
clientlibrary/worker/fan-out-shard-consumer.go
View file

@ -103,6 +103,8 @@ func (sc *FanOutShardConsumer) getRecords() error {
return err return err
} }
refreshLeaseTimer = time.After(time.Until(sc.shard.LeaseTimeout.Add(-time.Duration(sc.kclConfig.LeaseRefreshPeriodMillis) * time.Millisecond))) 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(): case event, ok := <-shardSub.GetStream().Events():
if !ok { if !ok {
// need to resubscribe to shard // need to resubscribe to shard

2
clientlibrary/worker/polling-shard-consumer.go
View file

@ -122,6 +122,8 @@ func (sc *PollingShardConsumer) getRecords() error {
sc.shard.ID, sc.consumerID, err) sc.shard.ID, sc.consumerID, err)
return err return err
} }
// log metric for renewed lease for worker
sc.mService.LeaseRenewed(sc.shard.ID)
} }
getRecordsStartTime := time.Now() getRecordsStartTime := time.Now()