-
Notifications
You must be signed in to change notification settings - Fork 12
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: add batch byte size limit configuration
- Loading branch information
Showing
6 changed files
with
238 additions
and
40 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,63 @@ | ||
| package kafka | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "strconv" | ||
| "strings" | ||
| ) | ||
|
|
||
| func ResolveUnionIntOrStringValue(input any) (int, error) { | ||
| switch value := input.(type) { | ||
| case int: | ||
| return value, nil | ||
| case uint: | ||
| return int(value), nil | ||
| case nil: | ||
| return 0, nil | ||
| case string: | ||
| intValue, err := strconv.ParseInt(value, 10, 64) | ||
| if err == nil { | ||
| return int(intValue), nil | ||
|
Check failure Code scanning / CodeQL Incorrect conversion between integer types High
Incorrect conversion of a signed 64-bit integer from
strconv.ParseInt Error loading related location Loading |
||
| } | ||
|
|
||
| result, err := convertSizeUnitToByte(value) | ||
| if err != nil { | ||
| return 0, err | ||
| } | ||
|
|
||
| return result, nil | ||
| } | ||
|
|
||
| return 0, fmt.Errorf("invalid input: %v", input) | ||
| } | ||
|
|
||
| func convertSizeUnitToByte(str string) (int, error) { | ||
| if len(str) < 2 { | ||
| return 0, fmt.Errorf("invalid input: %s", str) | ||
| } | ||
|
|
||
| // Extract the numeric part of the input | ||
| sizeStr := str[:len(str)-2] | ||
| sizeStr = strings.TrimSpace(sizeStr) | ||
| sizeStr = strings.ReplaceAll(sizeStr, ",", ".") | ||
|
|
||
| size, err := strconv.ParseFloat(sizeStr, 64) | ||
| if err != nil { | ||
| return 0, fmt.Errorf("cannot extract numeric part for the input %s, err = %w", str, err) | ||
| } | ||
|
|
||
| // Determine the unit (B, KB, MB, GB) | ||
| unit := str[len(str)-2:] | ||
| switch strings.ToUpper(unit) { | ||
| case "B": | ||
| return int(size), nil | ||
| case "KB": | ||
| return int(size * 1024), nil | ||
| case "MB": | ||
| return int(size * 1024 * 1024), nil | ||
| case "GB": | ||
| return int(size * 1024 * 1024 * 1024), nil | ||
| default: | ||
| return 0, fmt.Errorf("unsupported unit: %s, you can specify one of B, KB, MB and GB", unit) | ||
| } | ||
| } | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,85 @@ | ||
| package kafka | ||
|
|
||
| import "testing" | ||
|
|
||
| func TestDcp_ResolveConnectionBufferSize(t *testing.T) { | ||
| tests := []struct { | ||
| input any | ||
| name string | ||
| want int | ||
| }{ | ||
| { | ||
| name: "When_Client_Gives_Int_Value", | ||
| input: 20971520, | ||
| want: 20971520, | ||
| }, | ||
| { | ||
| name: "When_Client_Gives_UInt_Value", | ||
| input: uint(10971520), | ||
| want: 10971520, | ||
| }, | ||
| { | ||
| name: "When_Client_Gives_StringInt_Value", | ||
| input: "15971520", | ||
| want: 15971520, | ||
| }, | ||
| { | ||
| name: "When_Client_Gives_KB_Value", | ||
| input: "500kb", | ||
| want: 500 * 1024, | ||
| }, | ||
| { | ||
| name: "When_Client_Gives_MB_Value", | ||
| input: "10mb", | ||
| want: 10 * 1024 * 1024, | ||
| }, | ||
| } | ||
| for _, tt := range tests { | ||
| t.Run(tt.name, func(t *testing.T) { | ||
| if got, _ := ResolveUnionIntOrStringValue(tt.input); got != tt.want { | ||
| t.Errorf("ResolveConnectionBufferSize() = %v, want %v", got, tt.want) | ||
| } | ||
| }) | ||
| } | ||
| } | ||
|
|
||
| func TestConvertToBytes(t *testing.T) { | ||
| testCases := []struct { | ||
| input string | ||
| expected int | ||
| err bool | ||
| }{ | ||
| {"1kb", 1024, false}, | ||
| {"5mb", 5 * 1024 * 1024, false}, | ||
| {"5,5mb", 5.5 * 1024 * 1024, false}, | ||
| {"8.5mb", 8.5 * 1024 * 1024, false}, | ||
| {"10,25 mb", 10.25 * 1024 * 1024, false}, | ||
| {"10gb", 10 * 1024 * 1024 * 1024, false}, | ||
| {"1KB", 1024, false}, | ||
| {"5MB", 5 * 1024 * 1024, false}, | ||
| {"12 MB", 12 * 1024 * 1024, false}, | ||
| {"10GB", 10 * 1024 * 1024 * 1024, false}, | ||
| {"123", 0, true}, | ||
| {"15TB", 0, true}, | ||
| {"invalid", 0, true}, | ||
| {"", 0, true}, | ||
| {"123 KB", 123 * 1024, false}, | ||
| {"1 MB", 1 * 1024 * 1024, false}, | ||
| } | ||
|
|
||
| for _, tc := range testCases { | ||
| result, err := convertSizeUnitToByte(tc.input) | ||
|
|
||
| if tc.err && err == nil { | ||
| t.Errorf("Expected an error for input %s, but got none", tc.input) | ||
| } | ||
|
|
||
| if !tc.err && err != nil { | ||
| t.Errorf("Unexpected error for input %s: %v", tc.input, err) | ||
| } | ||
|
|
||
| if result != tc.expected { | ||
| t.Errorf("For input %s, expected %d bytes, but got %d", tc.input, tc.expected, result) | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.