fix: Handle ProvisionedThroughputExceededException throttling #19

spentakota · 2023-01-23T20:02:12Z

Signed-off-by: Shiva Pentakota spentakota@vmware.com

Devika-Shanbhag · 2023-01-23T21:25:19Z

clientlibrary/config/config.go

@@ -283,6 +283,9 @@ type (

 		// LeaseSyncingTimeInterval The number of milliseconds to wait before syncing with lease table (dynamoDB)
 		LeaseSyncingTimeIntervalMillis int
+
+		// MaxRetryCount The number of retries for getRecords in error case


Nit: Since this exportable field is in a general config file for the client library, we could have a comment like
// MaxRetryCount The maximum number of retries in case of error ?

That way it could be a general config field not necessarily only linked to getRecords()

Devika-Shanbhag · 2023-01-23T21:33:09Z

clientlibrary/worker/polling-shard-consumer.go

@@ -134,15 +143,53 @@ func (sc *PollingShardConsumer) getRecords() error {
 			ShardIterator: shardIterator,
 		}

+		// Each shard can support up to five read transactions per second.
+		if transactionNum == 6 {


Could we make this magic number 6 a constant like MaxReadTransactionsPerSecond = 5?

The condition then could be if transactionNum > MaxReadTransactionsPerSecond

Devika-Shanbhag · 2023-01-23T21:43:27Z

clientlibrary/worker/polling-shard-consumer.go

+				// If there is insufficient provisioned throughput on the stream,
+				// subsequent calls made within the next 1 second throw ProvisionedThroughputExceededException.
+				// ref: https://docs.aws.amazon.com/streams/latest/dev/service-sizes-and-limits.html
+				waitTime := time.Since(getRecordsTransactionTime)


There's scope to refactor this block (line 176-179) into a method - something like waitASecond(waitTimePassed)
Then reuse it in multiple places where we need to waitASecond()

vmwclabot · 2023-01-23T21:49:48Z

@spentakota, you must sign every commit in this pull request acknowledging our Developer Certificate of Origin before your changes are merged. This can be done by adding Signed-off-by: John Doe <john.doe@email.org> to the last line of each Git commit message. The e-mail address used to sign must match the e-mail address of the Git author. Click here to view the Developer Certificate of Origin agreement.

Commit d5bded3 must be signed by spentakota@vmware.com
Commit a5348d7 must be signed by spentakota@vmware.com
Commit 27ba073 must be signed by spentakota@vmware.com

Devika-Shanbhag

Thanks for a quick yet comprehensive fix to handle ProvisionedThroughputExceededException throttling to getRecords()!

Overall, looks good to me with few review comments on lines and below over-arching suggestions which could be skipped or addressed later if not now:

From code quality perspective, getRecords() is becoming a brain overload - it might help to break it down into smaller logical methods that preferably - do one thing, do it well, do it only.
This method should be tested so we can assert that it is doing what we expect it to do - which is more important given its complex logic and calculations owing to underlying behaviors that you've rightly linked in comments. A quick basic test would really help in building confidence for these changes.

Devika-Shanbhag · 2023-01-23T21:53:16Z

clientlibrary/worker/polling-shard-consumer.go

+		// Calculate size of records from read transaction
+		numBytes := 0
+		for _, record := range getResp.Records {
+			numBytes = numBytes + len(record.Data)


Nit: Could use Compound Assignment Operator += such that numBytes += len(record.Data) for better notational convenience and readability (although the difference is only syntactical, this could possibly avoid a bug)

Devika-Shanbhag · 2023-01-23T21:59:13Z

clientlibrary/worker/polling-shard-consumer.go

+		}
+
+		// Add to number of getRecords successful transactions
+		transactionNum++


Nit: Possibly rename this variable to something like NumSuccessfulTransactions so it's a clear increment counter without the need for this comment. Ok to retain as if there are other reasons why it's called transactionNum

Devika-Shanbhag · 2023-01-23T22:11:44Z

clientlibrary/worker/polling-shard-consumer.go

+				// ref: https://docs.aws.amazon.com/streams/latest/dev/service-sizes-and-limits.html
+				waitTime := time.Since(getRecordsTransactionTime)
+				if waitTime < time.Second {
+					time.Sleep(time.Second - waitTime)


Nit: Suggest using time.Sub() instead. Also, consistent with usage of time.Sub() to calculate timePassed in line 234

Devika-Shanbhag · 2023-01-23T22:20:58Z

clientlibrary/worker/polling-shard-consumer.go

+		timePassed := currTime.Sub(lastCheckTime)
+		lastCheckTime = currTime
+
+		remBytes = remBytes + float64(timePassed.Seconds())*(MaxBytes/(float64(time.Second*5)))


Nit: Another instance for replacement with += and could be called remainingBytes for clarity.

Devika-Shanbhag · 2023-01-23T22:25:56Z

clientlibrary/worker/polling-shard-consumer.go

+		timePassed := currTime.Sub(lastCheckTime)
+		lastCheckTime = currTime
+
+		remBytes = remBytes + float64(timePassed.Seconds())*(MaxBytes/(float64(time.Second*5)))


Not evident what the expression does without processing the numbers in the comment. Could you simplify this (MaxBytes/(float64(time.Second*5) to be called something that tells what it represents/evaluates to like readDataRate?

Also, timePassed.Seconds() returns a float64, is the outer float64 casting necessary?

Devika-Shanbhag · 2023-01-23T22:49:41Z

clientlibrary/worker/polling-shard-consumer.go

+		if remBytes <= float64(numBytes) {
+			// Wait until cool down period has passed to prevent ProvisionedThroughputExceededException
+			coolDown := numBytes / MaxBytesPerSecond
+			time.Sleep(time.Duration(coolDown) * time.Second)


I think there is a possibility of loss of precision in this arithmetic operation. coolDown is type float64?

Durations is measured as int64 type in nanoseconds, so time.Duration(f) truncates the floating-point f value to an integer.

We could multiply the number of seconds by the number of duration units per second instead?
time.Sleep(time.Duration(coolDown * float64(time.Second)))

Devika-Shanbhag · 2023-01-23T23:00:27Z

clientlibrary/worker/polling-shard-consumer.go

+			coolDown := numBytes / MaxBytesPerSecond
+			time.Sleep(time.Duration(coolDown) * time.Second)
+		} else {
+			remBytes = remBytes - float64(numBytes)


Nit: Noticed multiple occurrences of casting numBytes as float64(numBytes). Would it make sense for numBytes to have float64 type to begin with? We might only need one casting on line 209 as
numBytes += float64(len(record.Data))

Another instance for replacement with -=.

vmwclabot · 2023-01-24T00:03:54Z

@spentakota, you must sign every commit in this pull request acknowledging our Developer Certificate of Origin before your changes are merged. This can be done by adding Signed-off-by: John Doe <john.doe@email.org> to the last line of each Git commit message. The e-mail address used to sign must match the e-mail address of the Git author. Click here to view the Developer Certificate of Origin agreement.

Commit 5ff9229 must be signed by spentakota@vmware.com

Devika-Shanbhag · 2023-01-24T00:40:33Z

clientlibrary/worker/polling-shard-consumer.go

+	}
+}
+
+func (sc *PollingShardConsumer) callGetRecordsAPI(successfulTransactionNum *int, remBytes *float64,


Refactor suggestion: If all these fields are related or represent a particular struct, you could capture these fields in that struct or in the method's receiver struct (PollingShardConsumer).

You could return the values instead of updating the field references in this method - making callGetRecordsAPI() idempotent, such that it does not modify the fields that are being passed to it as parameters, but instead returns their values - that might clean this up from having multiple pointers.

Signed-off-by: Shiva Pentakota <spentakota@vmware.com>

vmwclabot · 2023-01-24T17:19:09Z

@spentakota, you must sign every commit in this pull request acknowledging our Developer Certificate of Origin before your changes are merged. This can be done by adding Signed-off-by: John Doe <john.doe@email.org> to the last line of each Git commit message. The e-mail address used to sign must match the e-mail address of the Git author. Click here to view the Developer Certificate of Origin agreement.

Commit 2a5b00e must be signed by spentakota@vmware.com

…ordsAPI Signed-off-by: Shiva Pentakota <spentakota@vmware.com>

spentakota · 2023-01-24T20:53:35Z

Closing MR in favor of: #21

vmwjc · 2023-01-24T05:38:43Z

clientlibrary/worker/polling-shard-consumer.go

+	mService      metrics.MonitoringService
+	currTime      time.Time
+	callsLeft     int
+	remBytes      float64


Let's use an int here. It doesn't really make sense to have fractional bytes in this case.

vmwjc · 2023-01-24T05:39:35Z

clientlibrary/worker/polling-shard-consumer.go

+	callsLeft     int
+	remBytes      float64
+	lastCheckTime time.Time
+	bytesRead     float64


Let's use ints, not floats for vars that count bytes.

vmwjc · 2023-01-24T05:42:26Z

clientlibrary/worker/polling-shard-consumer.go

+	MaxBytes            = 10000000.0
+	MaxBytesPerSecond   = 2000000.0
+	BytesToMbConversion = 1000000.0


These can all be ints.

vmwjc · 2023-01-24T05:47:15Z

clientlibrary/worker/polling-shard-consumer.go

+	timePassed := currentTime.Sub(sc.lastCheckTime)
+	sc.lastCheckTime = currentTime
+	sc.remBytes += timePassed.Seconds() * MaxBytesPerSecond
+	transactionReadRate := sc.bytesRead / (timePassed.Seconds() * BytesToMbConversion)


With the byte-counting vars as ints, you can do arithmetic by dividing a time.Duration by time.Second

Suggested change

timePassed := currentTime.Sub(sc.lastCheckTime)

sc.lastCheckTime = currentTime

sc.remBytes += timePassed.Seconds() * MaxBytesPerSecond

transactionReadRate := sc.bytesRead / (timePassed.Seconds() * BytesToMbConversion)

secondsPassed := currentTime.Sub(sc.lastCheckTime) / time.Second

sc.lastCheckTime = currentTime

sc.remBytes += secondsPassed * MaxBytesPerSecond

transactionReadRate := sc.bytesRead / (secondsPassed * BytesToMbConversion)

spentakota force-pushed the spentakota/handleGetRecordsErrors branch from bf812a7 to c72ae54 Compare January 23, 2023 20:04

spentakota requested a review from vmwjc January 23, 2023 20:04

spentakota force-pushed the spentakota/handleGetRecordsErrors branch from c72ae54 to 7b318de Compare January 23, 2023 20:07

Devika-Shanbhag reviewed Jan 23, 2023

View reviewed changes

spentakota force-pushed the spentakota/handleGetRecordsErrors branch from 7b318de to 27ba073 Compare January 23, 2023 21:49

vmwclabot added the dco-required label Jan 23, 2023

spentakota force-pushed the spentakota/handleGetRecordsErrors branch from 27ba073 to 07b65d9 Compare January 23, 2023 21:52

vmwclabot removed the dco-required label Jan 23, 2023

Devika-Shanbhag suggested changes Jan 23, 2023

View reviewed changes

vmwclabot added the dco-required label Jan 24, 2023

spentakota force-pushed the spentakota/handleGetRecordsErrors branch from 5ff9229 to 28246c7 Compare January 24, 2023 00:04

vmwclabot removed the dco-required label Jan 24, 2023

Devika-Shanbhag reviewed Jan 24, 2023

View reviewed changes

spentakota added 5 commits January 23, 2023 18:54

fix: adding MaxRetryCount for capped num of retries

9fcd1a8

Signed-off-by: Shiva Pentakota <spentakota@vmware.com>

fix: rate limiting getRecords tps

b49cbdf

Signed-off-by: Shiva Pentakota <spentakota@vmware.com>

fix: max bytes per second getRecords check

cd1f34e

Signed-off-by: Shiva Pentakota <spentakota@vmware.com>

fix: created seperate rate limit check method callGetRecordsAPI

adcff0b

Signed-off-by: Shiva Pentakota <spentakota@vmware.com>

fix: added callGetRecordsAPI tests

ae3763e

Signed-off-by: Shiva Pentakota <spentakota@vmware.com>

spentakota force-pushed the spentakota/handleGetRecordsErrors branch from 28246c7 to ae3763e Compare January 24, 2023 04:51

vmwclabot added the dco-required label Jan 24, 2023

fix: add case for setting lastCheckTime in first iteration callGetRec…

706e803

…ordsAPI Signed-off-by: Shiva Pentakota <spentakota@vmware.com>

spentakota force-pushed the spentakota/handleGetRecordsErrors branch from 2a5b00e to 706e803 Compare January 24, 2023 17:20

vmwclabot removed the dco-required label Jan 24, 2023

spentakota closed this Jan 24, 2023

vmwjc reviewed Jan 24, 2023

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

fix: Handle ProvisionedThroughputExceededException throttling #19

fix: Handle ProvisionedThroughputExceededException throttling #19

spentakota commented Jan 23, 2023

Devika-Shanbhag Jan 23, 2023

Devika-Shanbhag Jan 23, 2023 •

edited

Devika-Shanbhag Jan 23, 2023

vmwclabot commented Jan 23, 2023

Devika-Shanbhag left a comment

Devika-Shanbhag Jan 23, 2023

Devika-Shanbhag Jan 23, 2023

Devika-Shanbhag Jan 23, 2023

Devika-Shanbhag Jan 23, 2023

Devika-Shanbhag Jan 23, 2023

Devika-Shanbhag Jan 23, 2023

Devika-Shanbhag Jan 23, 2023

vmwclabot commented Jan 24, 2023

Devika-Shanbhag Jan 24, 2023

vmwclabot commented Jan 24, 2023

spentakota commented Jan 24, 2023

vmwjc Jan 24, 2023

vmwjc Jan 24, 2023

vmwjc Jan 24, 2023

vmwjc Jan 24, 2023

fix: Handle ProvisionedThroughputExceededException throttling #19

fix: Handle ProvisionedThroughputExceededException throttling #19

Conversation

spentakota commented Jan 23, 2023

Choose a reason for hiding this comment

Devika-Shanbhag Jan 23, 2023 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vmwclabot commented Jan 23, 2023

Devika-Shanbhag left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vmwclabot commented Jan 24, 2023

Choose a reason for hiding this comment

vmwclabot commented Jan 24, 2023

spentakota commented Jan 24, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Devika-Shanbhag Jan 23, 2023 •

edited