Use commandargs - #378
Merged
Merged
Conversation
📝 WalkthroughWalkthroughThis pull request introduces a protobuf-based framework for connector command-line flag definitions, restructures proto module configuration into separate Changes
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~28 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 7
🧹 Nitpick comments (1)
proto/connectorcommands/adiom/commands/connectors/v1/cosmos.proto (1)
32-34: Keep Cosmos help grouping consistent forcosmos_parallel_partition_workers.Line 32 defines a Cosmos-specific flag but omits the category used by neighboring Cosmos fields, so help output may look fragmented.
Proposed metadata tweak
int32 cosmos_parallel_partition_workers = 5 [(commandargs.v1.flag) = { name: "cosmos-parallel-partition-workers" + category: "Cosmos DB-specific Options" }];🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@proto/connectorcommands/adiom/commands/connectors/v1/cosmos.proto` around lines 32 - 34, The field cosmos_parallel_partition_workers is missing the Cosmos help/category metadata that neighboring Cosmos flags use; update its (commandargs.v1.flag) options to include the same category/grouping metadata (e.g., add the category key/value used by other Cosmos flags) so the help output groups it consistently with other Cosmos options—locate the int32 cosmos_parallel_partition_workers = 5 declaration and extend its flag options to match the metadata format used by adjacent Cosmos fields.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@buf.gen.commands.yaml`:
- Around line 6-8: The buf plugin protoc-gen-commandargs-urfave referenced in
buf.gen.commands.yaml is not being installed by the bootstrap script; update
scripts/devsetup.sh to install protoc-gen-commandargs-urfave (matching the
pattern used for protoc-gen-go, protoc-gen-go-grpc, and protoc-gen-connect-go)
so Buf codegen no longer fails, and add a short note about the new tool to the
development setup docs so developers know it's required when regenerating
protos.
In `@buf.gen.yaml`:
- Around line 17-18: The buf.gen.yaml references a non-existent proto input path
under the inputs block ("directory: proto/dsync"), which breaks Buf generation;
fix it by either creating the missing proto/dsync directory with the expected
.proto files or updating the inputs.directory value to the actual proto root in
the repo (e.g., replace "proto/dsync" with the correct path) so the inputs block
in buf.gen.yaml points to an existing proto directory.
In `@internal/app/options/connectorflags.go`:
- Around line 733-736: The loop over f.Payload (in connectorflags.go) currently
uses strings.Cut(p, ":") but ignores the boolean result, causing entries like
"--payload foo" to become key "foo" with empty value; update the loop in the
function handling f.Payload to validate the Cut result (the third return value,
often named ok) and reject or return an error/log when ok is false (i.e., no ':'
present), rather than inserting an empty value into m; reference the variables
f.Payload, p, strings.Cut, and the map m when making the change so malformed
payload entries are surfaced to the caller.
- Around line 593-606: Validate the --data-type string against the
adiomv1.DataType_value map before converting in kafkaSrcFromFlags and
kafkaDstFromFlags: add a small helper (e.g., validateDataTypeString) that looks
up f.DataType in adiomv1.DataType_value, returns an error if missing, and only
then performs the conversion to adiomv1.DataType; update kafkaSrcFromFlags and
kafkaDstFromFlags to call this helper and return the error when the data type is
invalid so typos or unknown values fail fast instead of defaulting to
DATA_TYPE_UNKNOWN.
- Around line 652-663: fileSettingsFromFlags currently checks delimiter length
using byte-length which rejects valid single UTF-8 characters; change the
validation to use utf8.RuneCountInString(f.Delimiter) != 1 and when assigning
s.Delimiter convert the string to its first rune (e.g., []rune(f.Delimiter)[0])
so multi-byte single characters (like '€') pass and are stored correctly; keep
the same error return (fileconnector.ErrInvalidDelimiter) and preserve existing
fields (Uri, Format, BatchSize).
In `@proto/connectorcommands/adiom/commands/connectors/v1/dynamodb.proto`:
- Around line 23-25: The help text for the proto field declaration string id = 3
[(commandargs.v1.flag) = { usage: "A fixed id for the connector" }]; is
misleading because the field is currently a user-supplied string; either enforce
a fixed value or update the usage text—so either (A) enforce a fixed id by
removing the flag annotation and hardcoding the connector id where the proto is
consumed (or set the server-side default and validate incoming values against
it), or (B) simply change the usage string to something accurate like "A
user-provided/unique id for the connector" (update the usage in the
(commandargs.v1.flag) annotation for the id field).
In `@proto/connectorcommands/adiom/commands/connectors/v1/fakesource.proto`:
- Around line 58-61: The flag help text for the protobuf field
max_updates_per_tick (flag name "max-updates-per-tick") is misleading—change its
usage string from "Number of docs per update" to indicate "per tick" instead
(e.g., "Number of docs per tick") so the flag description matches its semantics;
update the usage value in the options for the int32 max_updates_per_tick field
accordingly.
---
Nitpick comments:
In `@proto/connectorcommands/adiom/commands/connectors/v1/cosmos.proto`:
- Around line 32-34: The field cosmos_parallel_partition_workers is missing the
Cosmos help/category metadata that neighboring Cosmos flags use; update its
(commandargs.v1.flag) options to include the same category/grouping metadata
(e.g., add the category key/value used by other Cosmos flags) so the help output
groups it consistently with other Cosmos options—locate the int32
cosmos_parallel_partition_workers = 5 declaration and extend its flag options to
match the metadata format used by adjacent Cosmos fields.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 2acaf0bb-c841-4928-bb93-18d48598d691
⛔ Files ignored due to path filters (31)
buf.lockis excluded by!**/*.lockgen/adiom/commands/connectors/v1/cosmos.pb.gois excluded by!**/*.pb.go,!**/gen/**gen/adiom/commands/connectors/v1/cosmos_commandargs.gois excluded by!**/gen/**gen/adiom/commands/connectors/v1/devnull.pb.gois excluded by!**/*.pb.go,!**/gen/**gen/adiom/commands/connectors/v1/devnull_commandargs.gois excluded by!**/gen/**gen/adiom/commands/connectors/v1/devrandom.pb.gois excluded by!**/*.pb.go,!**/gen/**gen/adiom/commands/connectors/v1/devrandom_commandargs.gois excluded by!**/gen/**gen/adiom/commands/connectors/v1/dynamodb.pb.gois excluded by!**/*.pb.go,!**/gen/**gen/adiom/commands/connectors/v1/dynamodb_commandargs.gois excluded by!**/gen/**gen/adiom/commands/connectors/v1/fakesource.pb.gois excluded by!**/*.pb.go,!**/gen/**gen/adiom/commands/connectors/v1/fakesource_commandargs.gois excluded by!**/gen/**gen/adiom/commands/connectors/v1/file.pb.gois excluded by!**/*.pb.go,!**/gen/**gen/adiom/commands/connectors/v1/file_commandargs.gois excluded by!**/gen/**gen/adiom/commands/connectors/v1/grpc.pb.gois excluded by!**/*.pb.go,!**/gen/**gen/adiom/commands/connectors/v1/grpc_commandargs.gois excluded by!**/gen/**gen/adiom/commands/connectors/v1/kafka.pb.gois excluded by!**/*.pb.go,!**/gen/**gen/adiom/commands/connectors/v1/kafka_commandargs.gois excluded by!**/gen/**gen/adiom/commands/connectors/v1/mongo.pb.gois excluded by!**/*.pb.go,!**/gen/**gen/adiom/commands/connectors/v1/mongo_commandargs.gois excluded by!**/gen/**gen/adiom/commands/connectors/v1/postgres.pb.gois excluded by!**/*.pb.go,!**/gen/**gen/adiom/commands/connectors/v1/postgres_commandargs.gois excluded by!**/gen/**gen/adiom/commands/connectors/v1/s3.pb.gois excluded by!**/*.pb.go,!**/gen/**gen/adiom/commands/connectors/v1/s3_commandargs.gois excluded by!**/gen/**gen/adiom/commands/connectors/v1/s3vectors.pb.gois excluded by!**/*.pb.go,!**/gen/**gen/adiom/commands/connectors/v1/s3vectors_commandargs.gois excluded by!**/gen/**gen/adiom/commands/connectors/v1/sqlbatch.pb.gois excluded by!**/*.pb.go,!**/gen/**gen/adiom/commands/connectors/v1/sqlbatch_commandargs.gois excluded by!**/gen/**gen/adiom/v1/adiom.pb.gois excluded by!**/*.pb.go,!**/gen/**gen/adiom/v1/messages.pb.gois excluded by!**/*.pb.go,!**/gen/**gen/adiom/v1/vector.pb.gois excluded by!**/*.pb.go,!**/gen/**go.sumis excluded by!**/*.sum
📒 Files selected for processing (23)
buf.gen.commands.yamlbuf.gen.yamlbuf.yamlgo.modinternal/app/options/connectorflags.gointernal/app/options/flags.gointernal/app/options/grpc.goproto/adiom/v1/adiom.protoproto/adiom/v1/messages.protoproto/adiom/v1/vector.protoproto/connectorcommands/adiom/commands/connectors/v1/cosmos.protoproto/connectorcommands/adiom/commands/connectors/v1/devnull.protoproto/connectorcommands/adiom/commands/connectors/v1/devrandom.protoproto/connectorcommands/adiom/commands/connectors/v1/dynamodb.protoproto/connectorcommands/adiom/commands/connectors/v1/fakesource.protoproto/connectorcommands/adiom/commands/connectors/v1/file.protoproto/connectorcommands/adiom/commands/connectors/v1/grpc.protoproto/connectorcommands/adiom/commands/connectors/v1/kafka.protoproto/connectorcommands/adiom/commands/connectors/v1/mongo.protoproto/connectorcommands/adiom/commands/connectors/v1/postgres.protoproto/connectorcommands/adiom/commands/connectors/v1/s3.protoproto/connectorcommands/adiom/commands/connectors/v1/s3vectors.protoproto/connectorcommands/adiom/commands/connectors/v1/sqlbatch.proto
💤 Files with no reviewable changes (3)
- proto/adiom/v1/adiom.proto
- proto/adiom/v1/messages.proto
- proto/adiom/v1/vector.proto
Comment on lines
+17
to
+18
| inputs: | ||
| - directory: proto/dsync No newline at end of file |
There was a problem hiding this comment.
Broken Buf generation input path (blocker).
Line 18 points to proto/dsync, which is missing in the repo context. That will make Buf generation fail immediately.
Please either create proto/dsync or update this input to the actual existing proto root.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@buf.gen.yaml` around lines 17 - 18, The buf.gen.yaml references a
non-existent proto input path under the inputs block ("directory: proto/dsync"),
which breaks Buf generation; fix it by either creating the missing proto/dsync
directory with the expected .proto files or updating the inputs.directory value
to the actual proto root in the repo (e.g., replace "proto/dsync" with the
correct path) so the inputs block in buf.gen.yaml points to an existing proto
directory.
Comment on lines
+593
to
+606
| func kafkaSrcFromFlags(f *connectorsv1.KafkaSrcFlags) (adiomv1connect.ConnectorServiceHandler, error) { | ||
| dataType := adiomv1.DataType(adiomv1.DataType_value[f.DataType]) | ||
| tm := map[string][]string{} | ||
| for _, topic := range topics { | ||
| for _, topic := range f.Topics { | ||
| tm[topic] = nil | ||
| } | ||
| for _, m := range topicMappings { | ||
| for _, m := range f.TopicMappings { | ||
| topic, ns, ok := strings.Cut(m, ":") | ||
| if !ok { | ||
| return nil, fmt.Errorf("invalid topic mapping %v", m) | ||
| } | ||
| tm[topic] = append(tm[topic], ns) | ||
| } | ||
| return kafka.NewKafkaConn(brokers, tm, kafka.DsyncMessageToUpdate, kafka.DsyncMessageToNamespace, user, password, kafkaOffset, dataType), nil | ||
| return kafka.NewKafkaConn(f.Brokers, tm, kafka.DsyncMessageToUpdate, kafka.DsyncMessageToNamespace, f.SaslUser, f.SaslPassword, f.Offset, dataType), nil |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
cd internal/app/options && wc -l connectorflags.goRepository: adiom-data/dsync
Length of output: 81
🏁 Script executed:
sed -n '593,620p' internal/app/options/connectorflags.goRepository: adiom-data/dsync
Length of output: 1121
🏁 Script executed:
rg "func (kafka|DataType_value)" internal/app/options/connectorflags.go -A 5Repository: adiom-data/dsync
Length of output: 601
🏁 Script executed:
fd "adiomv1" --type f | head -20Repository: adiom-data/dsync
Length of output: 42
🏁 Script executed:
rg "adiomv1" internal/app/options/connectorflags.go | head -5Repository: adiom-data/dsync
Length of output: 430
🏁 Script executed:
rg "import" internal/app/options/connectorflags.go | grep -E "adiom|connect"Repository: adiom-data/dsync
Length of output: 42
🏁 Script executed:
find . -name "*.pb.go" -o -name "*adiom*" | head -20Repository: adiom-data/dsync
Length of output: 896
🏁 Script executed:
rg "type DataType" --type go | head -10Repository: adiom-data/dsync
Length of output: 107
🏁 Script executed:
sed -n '1,100p' gen/adiom/v1/messages.pb.go | head -50Repository: adiom-data/dsync
Length of output: 1326
🏁 Script executed:
rg "type DataType" gen/adiom/v1/messages.pb.go -A 30Repository: adiom-data/dsync
Length of output: 774
🏁 Script executed:
rg "DataType_value\s*=" gen/adiom/v1/messages.pb.go -A 15Repository: adiom-data/dsync
Length of output: 381
Add validation for --data-type flag before enum conversion.
When an unknown data type string is passed to kafkaSrcFromFlags (line 594) and kafkaDstFromFlags (line 610), the map lookup adiomv1.DataType_value[f.DataType] silently returns 0 (DATA_TYPE_UNKNOWN) instead of failing. This masks typos and changes connector behavior without error. Add a validation helper function to check the string exists in the enum map before conversion.
Suggested fix
+func parseKafkaDataType(v string) (adiomv1.DataType, error) {
+ raw, ok := adiomv1.DataType_value[v]
+ if !ok {
+ return 0, fmt.Errorf("invalid data-type %q", v)
+ }
+ return adiomv1.DataType(raw), nil
+}
+
func kafkaSrcFromFlags(f *connectorsv1.KafkaSrcFlags) (adiomv1connect.ConnectorServiceHandler, error) {
- dataType := adiomv1.DataType(adiomv1.DataType_value[f.DataType])
+ dataType, err := parseKafkaDataType(f.DataType)
+ if err != nil {
+ return nil, err
+ }
tm := map[string][]string{}
for _, topic := range f.Topics {
tm[topic] = nil
@@
return kafka.NewKafkaConn(f.Brokers, tm, kafka.DsyncMessageToUpdate, kafka.DsyncMessageToNamespace, f.SaslUser, f.SaslPassword, f.Offset, dataType), nil
}
func kafkaDstFromFlags(f *connectorsv1.KafkaDstFlags) (adiomv1connect.ConnectorServiceHandler, error) {
- dataType := adiomv1.DataType(adiomv1.DataType_value[f.DataType])
+ dataType, err := parseKafkaDataType(f.DataType)
+ if err != nil {
+ return nil, err
+ }
tm := map[string]string{}
for _, m := range f.NamespaceTopic {
ns, topic, ok := strings.Cut(m, ":")📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| func kafkaSrcFromFlags(f *connectorsv1.KafkaSrcFlags) (adiomv1connect.ConnectorServiceHandler, error) { | |
| dataType := adiomv1.DataType(adiomv1.DataType_value[f.DataType]) | |
| tm := map[string][]string{} | |
| for _, topic := range topics { | |
| for _, topic := range f.Topics { | |
| tm[topic] = nil | |
| } | |
| for _, m := range topicMappings { | |
| for _, m := range f.TopicMappings { | |
| topic, ns, ok := strings.Cut(m, ":") | |
| if !ok { | |
| return nil, fmt.Errorf("invalid topic mapping %v", m) | |
| } | |
| tm[topic] = append(tm[topic], ns) | |
| } | |
| return kafka.NewKafkaConn(brokers, tm, kafka.DsyncMessageToUpdate, kafka.DsyncMessageToNamespace, user, password, kafkaOffset, dataType), nil | |
| return kafka.NewKafkaConn(f.Brokers, tm, kafka.DsyncMessageToUpdate, kafka.DsyncMessageToNamespace, f.SaslUser, f.SaslPassword, f.Offset, dataType), nil | |
| func parseKafkaDataType(v string) (adiomv1.DataType, error) { | |
| raw, ok := adiomv1.DataType_value[v] | |
| if !ok { | |
| return 0, fmt.Errorf("invalid data-type %q", v) | |
| } | |
| return adiomv1.DataType(raw), nil | |
| } | |
| func kafkaSrcFromFlags(f *connectorsv1.KafkaSrcFlags) (adiomv1connect.ConnectorServiceHandler, error) { | |
| dataType, err := parseKafkaDataType(f.DataType) | |
| if err != nil { | |
| return nil, err | |
| } | |
| tm := map[string][]string{} | |
| for _, topic := range f.Topics { | |
| tm[topic] = nil | |
| } | |
| for _, m := range f.TopicMappings { | |
| topic, ns, ok := strings.Cut(m, ":") | |
| if !ok { | |
| return nil, fmt.Errorf("invalid topic mapping %v", m) | |
| } | |
| tm[topic] = append(tm[topic], ns) | |
| } | |
| return kafka.NewKafkaConn(f.Brokers, tm, kafka.DsyncMessageToUpdate, kafka.DsyncMessageToNamespace, f.SaslUser, f.SaslPassword, f.Offset, dataType), nil | |
| } |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@internal/app/options/connectorflags.go` around lines 593 - 606, Validate the
--data-type string against the adiomv1.DataType_value map before converting in
kafkaSrcFromFlags and kafkaDstFromFlags: add a small helper (e.g.,
validateDataTypeString) that looks up f.DataType in adiomv1.DataType_value,
returns an error if missing, and only then performs the conversion to
adiomv1.DataType; update kafkaSrcFromFlags and kafkaDstFromFlags to call this
helper and return the error when the data type is invalid so typos or unknown
values fail fast instead of defaulting to DATA_TYPE_UNKNOWN.
Comment on lines
+652
to
+663
| func fileSettingsFromFlags(f *connectorsv1.FileFlags, uri string) (fileconnector.ConnectorSettings, error) { | ||
| if f.Delimiter != "" && len(f.Delimiter) != 1 { | ||
| return fileconnector.ConnectorSettings{}, fileconnector.ErrInvalidDelimiter | ||
| } | ||
| s := fileconnector.ConnectorSettings{ | ||
| Uri: uri, | ||
| Format: f.Format, | ||
| BatchSize: int(f.BatchSize), | ||
| } | ||
| if len(f.Delimiter) == 1 { | ||
| s.Delimiter = rune(f.Delimiter[0]) | ||
| } |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
git ls-files | grep -E "connectorflags\.go"Repository: adiom-data/dsync
Length of output: 98
🏁 Script executed:
cat -n internal/app/options/connectorflags.go | sed -n '650,665p'Repository: adiom-data/dsync
Length of output: 612
🏁 Script executed:
rg "fileSettingsFromFlags" --type go -A 5 -B 5Repository: adiom-data/dsync
Length of output: 1755
🏁 Script executed:
rg -t go "Delimiter" internal/app/options/ | head -20Repository: adiom-data/dsync
Length of output: 407
🏁 Script executed:
rg -t go "fileSettingsFromFlags|FileFlags" --type-list | head -5Repository: adiom-data/dsync
Length of output: 449
🏁 Script executed:
find . -type f -name "*test.go" | xargs rg -l "fileSettingsFromFlags|FileFlags" 2>/dev/nullRepository: adiom-data/dsync
Length of output: 44
🏁 Script executed:
rg -t go "type ConnectorSettings" -A 20Repository: adiom-data/dsync
Length of output: 7448
🏁 Script executed:
rg -t go "type FileFlags" -A 15Repository: adiom-data/dsync
Length of output: 1468
🏁 Script executed:
fd -t go ".*test\.go" | xargs rg -l "delimiter|Delimiter" 2>/dev/null | head -10Repository: adiom-data/dsync
Length of output: 509
🏁 Script executed:
rg -t go "csv\.|csv\.Writer" connectors/file/ -B 2 -A 5Repository: adiom-data/dsync
Length of output: 1299
🏁 Script executed:
rg -t go "Delimiter" connectors/file/connector.go -B 3 -A 3Repository: adiom-data/dsync
Length of output: 1184
🏁 Script executed:
cat connectors/file/connector_test.go | head -100Repository: adiom-data/dsync
Length of output: 2687
🏁 Script executed:
rg "TestFileSettings|fileSettingsFromFlags" -A 20Repository: adiom-data/dsync
Length of output: 3052
🏁 Script executed:
rg "Delimiter" connectors/file/connector_test.go -B 3 -A 3Repository: adiom-data/dsync
Length of output: 626
Use utf8.RuneCountInString() to properly validate single-character delimiters.
Lines 653 and 661 use byte-length checks, which reject or mangle multi-byte UTF-8 characters even though they represent a single character. A non-ASCII delimiter like '€' would fail the validation check, contradicting the error message stating "must be a single character."
Suggested fix
+import "unicode/utf8"
+
func fileSettingsFromFlags(f *connectorsv1.FileFlags, uri string) (fileconnector.ConnectorSettings, error) {
- if f.Delimiter != "" && len(f.Delimiter) != 1 {
+ if f.Delimiter != "" && utf8.RuneCountInString(f.Delimiter) != 1 {
return fileconnector.ConnectorSettings{}, fileconnector.ErrInvalidDelimiter
}
s := fileconnector.ConnectorSettings{
Uri: uri,
Format: f.Format,
BatchSize: int(f.BatchSize),
}
- if len(f.Delimiter) == 1 {
- s.Delimiter = rune(f.Delimiter[0])
+ if f.Delimiter != "" {
+ s.Delimiter = []rune(f.Delimiter)[0]
}
return s, nil
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| func fileSettingsFromFlags(f *connectorsv1.FileFlags, uri string) (fileconnector.ConnectorSettings, error) { | |
| if f.Delimiter != "" && len(f.Delimiter) != 1 { | |
| return fileconnector.ConnectorSettings{}, fileconnector.ErrInvalidDelimiter | |
| } | |
| s := fileconnector.ConnectorSettings{ | |
| Uri: uri, | |
| Format: f.Format, | |
| BatchSize: int(f.BatchSize), | |
| } | |
| if len(f.Delimiter) == 1 { | |
| s.Delimiter = rune(f.Delimiter[0]) | |
| } | |
| import "unicode/utf8" | |
| func fileSettingsFromFlags(f *connectorsv1.FileFlags, uri string) (fileconnector.ConnectorSettings, error) { | |
| if f.Delimiter != "" && utf8.RuneCountInString(f.Delimiter) != 1 { | |
| return fileconnector.ConnectorSettings{}, fileconnector.ErrInvalidDelimiter | |
| } | |
| s := fileconnector.ConnectorSettings{ | |
| Uri: uri, | |
| Format: f.Format, | |
| BatchSize: int(f.BatchSize), | |
| } | |
| if f.Delimiter != "" { | |
| s.Delimiter = []rune(f.Delimiter)[0] | |
| } | |
| return s, nil | |
| } |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@internal/app/options/connectorflags.go` around lines 652 - 663,
fileSettingsFromFlags currently checks delimiter length using byte-length which
rejects valid single UTF-8 characters; change the validation to use
utf8.RuneCountInString(f.Delimiter) != 1 and when assigning s.Delimiter convert
the string to its first rune (e.g., []rune(f.Delimiter)[0]) so multi-byte single
characters (like '€') pass and are stored correctly; keep the same error return
(fileconnector.ErrInvalidDelimiter) and preserve existing fields (Uri, Format,
BatchSize).
Comment on lines
+733
to
+736
| for _, p := range f.Payload { | ||
| k, v, _ := strings.Cut(p, ":") | ||
| m[k] = v | ||
| } |
There was a problem hiding this comment.
Reject malformed --payload entries.
Line 734 ignores whether : was present, so --payload foo silently becomes {"foo":""} instead of surfacing a bad argument.
Suggested fix
for _, p := range f.Payload {
- k, v, _ := strings.Cut(p, ":")
+ k, v, ok := strings.Cut(p, ":")
+ if !ok {
+ return nil, fmt.Errorf("invalid payload %q: expected key:value", p)
+ }
m[k] = v
}🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@internal/app/options/connectorflags.go` around lines 733 - 736, The loop over
f.Payload (in connectorflags.go) currently uses strings.Cut(p, ":") but ignores
the boolean result, causing entries like "--payload foo" to become key "foo"
with empty value; update the loop in the function handling f.Payload to validate
the Cut result (the third return value, often named ok) and reject or return an
error/log when ok is false (i.e., no ':' present), rather than inserting an
empty value into m; reference the variables f.Payload, p, strings.Cut, and the
map m when making the change so malformed payload entries are surfaced to the
caller.
Comment on lines
+23
to
+25
| string id = 3 [(commandargs.v1.flag) = { | ||
| usage: "A fixed id for the connector" | ||
| }]; |
There was a problem hiding this comment.
id flag help text is misleading.
Line 24 says this is “a fixed id,” but the field is currently a regular user-provided string flag. Please either enforce a fixed value/default or reword the usage to describe it as configurable.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@proto/connectorcommands/adiom/commands/connectors/v1/dynamodb.proto` around
lines 23 - 25, The help text for the proto field declaration string id = 3
[(commandargs.v1.flag) = { usage: "A fixed id for the connector" }]; is
misleading because the field is currently a user-supplied string; either enforce
a fixed value or update the usage text—so either (A) enforce a fixed id by
removing the flag annotation and hardcoding the connector id where the proto is
consumed (or set the server-side default and validate incoming values against
it), or (B) simply change the usage string to something accurate like "A
user-provided/unique id for the connector" (update the usage in the
(commandargs.v1.flag) annotation for the id field).
Comment on lines
+58
to
+61
| int32 max_updates_per_tick = 11 [(commandargs.v1.flag) = { | ||
| name: "max-updates-per-tick" | ||
| usage: "Number of docs per update" | ||
| default: "300" |
There was a problem hiding this comment.
Fix max-updates-per-tick help text mismatch.
Line 60 describes “per update,” but this flag is clearly “per tick.” This can mislead users tuning throughput.
Suggested text fix
int32 max_updates_per_tick = 11 [(commandargs.v1.flag) = {
name: "max-updates-per-tick"
- usage: "Number of docs per update"
+ usage: "Maximum number of updates produced per tick"
default: "300"
}];📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| int32 max_updates_per_tick = 11 [(commandargs.v1.flag) = { | |
| name: "max-updates-per-tick" | |
| usage: "Number of docs per update" | |
| default: "300" | |
| int32 max_updates_per_tick = 11 [(commandargs.v1.flag) = { | |
| name: "max-updates-per-tick" | |
| usage: "Maximum number of updates produced per tick" | |
| default: "300" |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@proto/connectorcommands/adiom/commands/connectors/v1/fakesource.proto` around
lines 58 - 61, The flag help text for the protobuf field max_updates_per_tick
(flag name "max-updates-per-tick") is misleading—change its usage string from
"Number of docs per update" to indicate "per tick" instead (e.g., "Number of
docs per tick") so the flag description matches its semantics; update the usage
value in the options for the int32 max_updates_per_tick field accordingly.
There was a problem hiding this comment.
♻️ Duplicate comments (3)
internal/app/options/connectorflags.go (3)
652-663:⚠️ Potential issue | 🟡 MinorUse rune-based delimiter validation and assignment.
Line 653 and Line 661 use byte length/indexing, which rejects valid single UTF-8 delimiters and can assign the wrong character.
Suggested fix
import ( ... + "unicode/utf8" ) func fileSettingsFromFlags(f *connectorsv1.FileFlags, uri string) (fileconnector.ConnectorSettings, error) { - if f.Delimiter != "" && len(f.Delimiter) != 1 { + if f.Delimiter != "" && utf8.RuneCountInString(f.Delimiter) != 1 { return fileconnector.ConnectorSettings{}, fileconnector.ErrInvalidDelimiter } s := fileconnector.ConnectorSettings{ Uri: uri, Format: f.Format, BatchSize: int(f.BatchSize), } - if len(f.Delimiter) == 1 { - s.Delimiter = rune(f.Delimiter[0]) + if f.Delimiter != "" { + s.Delimiter = []rune(f.Delimiter)[0] } return s, nil }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@internal/app/options/connectorflags.go` around lines 652 - 663, In fileSettingsFromFlags replace the byte-based delimiter checks and assignment with rune-aware logic: use utf8.RuneCountInString (or convert to []rune) to ensure the delimiter string contains exactly one Unicode rune, returning fileconnector.ErrInvalidDelimiter when not; when valid, assign s.Delimiter from the single rune (e.g., runeSlice[0]) instead of using f.Delimiter[0]; update the validation at the place currently checking len(f.Delimiter) != 1 and the assignment where s.Delimiter is set.
733-736:⚠️ Potential issue | 🟡 MinorReject malformed
--payloadentries without:.Line 734 ignores
strings.Cutsuccess, so malformed payloads are silently accepted as empty-valued keys.Suggested fix
for _, p := range f.Payload { - k, v, _ := strings.Cut(p, ":") + k, v, ok := strings.Cut(p, ":") + if !ok { + return nil, fmt.Errorf("invalid payload %q: expected key:value", p) + } m[k] = v }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@internal/app/options/connectorflags.go` around lines 733 - 736, The loop over f.Payload currently ignores the success return from strings.Cut, allowing malformed entries like "foo" to become empty-valued keys; update the parsing in the for _, p := range f.Payload loop to use k, v, ok := strings.Cut(p, ":") and if ok is false reject the input (e.g., return a clear error such as fmt.Errorf("invalid --payload entry %q: must contain ':'", p)) instead of assigning m[k]=v, otherwise assign m[k]=v as before; reference the f.Payload loop, strings.Cut call, and the map m when making this change.
593-595:⚠️ Potential issue | 🟠 MajorValidate
--data-typebefore enum conversion.Line 594 and Line 610 currently use direct map indexing; unknown values silently become
DATA_TYPE_UNKNOWN(0). This should fail fast.Suggested fix
+func parseKafkaDataType(v string) (adiomv1.DataType, error) { + raw, ok := adiomv1.DataType_value[v] + if !ok { + return 0, fmt.Errorf("invalid data-type %q", v) + } + return adiomv1.DataType(raw), nil +} + func kafkaSrcFromFlags(f *connectorsv1.KafkaSrcFlags) (adiomv1connect.ConnectorServiceHandler, error) { - dataType := adiomv1.DataType(adiomv1.DataType_value[f.DataType]) + dataType, err := parseKafkaDataType(f.DataType) + if err != nil { + return nil, err + } tm := map[string][]string{} ... } func kafkaDstFromFlags(f *connectorsv1.KafkaDstFlags) (adiomv1connect.ConnectorServiceHandler, error) { - dataType := adiomv1.DataType(adiomv1.DataType_value[f.DataType]) + dataType, err := parseKafkaDataType(f.DataType) + if err != nil { + return nil, err + } tm := map[string]string{} ... }Also applies to: 609-611
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@internal/app/options/connectorflags.go` around lines 593 - 595, The code in kafkaSrcFromFlags uses direct map indexing into adiomv1.DataType_value (via f.DataType) which silently yields DATA_TYPE_UNKNOWN for invalid values; change kafkaSrcFromFlags to first check whether f.DataType exists in adiomv1.DataType_value (e.g., via a lookup with ok := adiomv1.DataType_value[f.DataType]) and if not return a clear error indicating the invalid --data-type value, then only perform the conversion to adiomv1.DataType when the lookup succeeds; apply the same validation pattern to the other occurrence(s) around the DataType conversion (the similar block at lines ~609-611) so invalid enums fail fast instead of becoming DATA_TYPE_UNKNOWN.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Duplicate comments:
In `@internal/app/options/connectorflags.go`:
- Around line 652-663: In fileSettingsFromFlags replace the byte-based delimiter
checks and assignment with rune-aware logic: use utf8.RuneCountInString (or
convert to []rune) to ensure the delimiter string contains exactly one Unicode
rune, returning fileconnector.ErrInvalidDelimiter when not; when valid, assign
s.Delimiter from the single rune (e.g., runeSlice[0]) instead of using
f.Delimiter[0]; update the validation at the place currently checking
len(f.Delimiter) != 1 and the assignment where s.Delimiter is set.
- Around line 733-736: The loop over f.Payload currently ignores the success
return from strings.Cut, allowing malformed entries like "foo" to become
empty-valued keys; update the parsing in the for _, p := range f.Payload loop to
use k, v, ok := strings.Cut(p, ":") and if ok is false reject the input (e.g.,
return a clear error such as fmt.Errorf("invalid --payload entry %q: must
contain ':'", p)) instead of assigning m[k]=v, otherwise assign m[k]=v as
before; reference the f.Payload loop, strings.Cut call, and the map m when
making this change.
- Around line 593-595: The code in kafkaSrcFromFlags uses direct map indexing
into adiomv1.DataType_value (via f.DataType) which silently yields
DATA_TYPE_UNKNOWN for invalid values; change kafkaSrcFromFlags to first check
whether f.DataType exists in adiomv1.DataType_value (e.g., via a lookup with ok
:= adiomv1.DataType_value[f.DataType]) and if not return a clear error
indicating the invalid --data-type value, then only perform the conversion to
adiomv1.DataType when the lookup succeeds; apply the same validation pattern to
the other occurrence(s) around the DataType conversion (the similar block at
lines ~609-611) so invalid enums fail fast instead of becoming
DATA_TYPE_UNKNOWN.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 3673623b-ff3b-44a3-b55e-6b9200278aff
⛔ Files ignored due to path filters (31)
buf.lockis excluded by!**/*.lockgen/adiom/commands/connectors/v1/cosmos.pb.gois excluded by!**/*.pb.go,!**/gen/**gen/adiom/commands/connectors/v1/cosmos_commandargs.gois excluded by!**/gen/**gen/adiom/commands/connectors/v1/devnull.pb.gois excluded by!**/*.pb.go,!**/gen/**gen/adiom/commands/connectors/v1/devnull_commandargs.gois excluded by!**/gen/**gen/adiom/commands/connectors/v1/devrandom.pb.gois excluded by!**/*.pb.go,!**/gen/**gen/adiom/commands/connectors/v1/devrandom_commandargs.gois excluded by!**/gen/**gen/adiom/commands/connectors/v1/dynamodb.pb.gois excluded by!**/*.pb.go,!**/gen/**gen/adiom/commands/connectors/v1/dynamodb_commandargs.gois excluded by!**/gen/**gen/adiom/commands/connectors/v1/fakesource.pb.gois excluded by!**/*.pb.go,!**/gen/**gen/adiom/commands/connectors/v1/fakesource_commandargs.gois excluded by!**/gen/**gen/adiom/commands/connectors/v1/file.pb.gois excluded by!**/*.pb.go,!**/gen/**gen/adiom/commands/connectors/v1/file_commandargs.gois excluded by!**/gen/**gen/adiom/commands/connectors/v1/grpc.pb.gois excluded by!**/*.pb.go,!**/gen/**gen/adiom/commands/connectors/v1/grpc_commandargs.gois excluded by!**/gen/**gen/adiom/commands/connectors/v1/kafka.pb.gois excluded by!**/*.pb.go,!**/gen/**gen/adiom/commands/connectors/v1/kafka_commandargs.gois excluded by!**/gen/**gen/adiom/commands/connectors/v1/mongo.pb.gois excluded by!**/*.pb.go,!**/gen/**gen/adiom/commands/connectors/v1/mongo_commandargs.gois excluded by!**/gen/**gen/adiom/commands/connectors/v1/postgres.pb.gois excluded by!**/*.pb.go,!**/gen/**gen/adiom/commands/connectors/v1/postgres_commandargs.gois excluded by!**/gen/**gen/adiom/commands/connectors/v1/s3.pb.gois excluded by!**/*.pb.go,!**/gen/**gen/adiom/commands/connectors/v1/s3_commandargs.gois excluded by!**/gen/**gen/adiom/commands/connectors/v1/s3vectors.pb.gois excluded by!**/*.pb.go,!**/gen/**gen/adiom/commands/connectors/v1/s3vectors_commandargs.gois excluded by!**/gen/**gen/adiom/commands/connectors/v1/sqlbatch.pb.gois excluded by!**/*.pb.go,!**/gen/**gen/adiom/commands/connectors/v1/sqlbatch_commandargs.gois excluded by!**/gen/**gen/adiom/v1/adiom.pb.gois excluded by!**/*.pb.go,!**/gen/**gen/adiom/v1/messages.pb.gois excluded by!**/*.pb.go,!**/gen/**gen/adiom/v1/vector.pb.gois excluded by!**/*.pb.go,!**/gen/**go.sumis excluded by!**/*.sum
📒 Files selected for processing (26)
.github/workflows/test.yml.gitignorebuf.gen.commands.yamlbuf.gen.yamlbuf.yamlgo.modinternal/app/options/connectorflags.gointernal/app/options/flags.gointernal/app/options/grpc.goproto/connectorcommands/adiom/commands/connectors/v1/cosmos.protoproto/connectorcommands/adiom/commands/connectors/v1/devnull.protoproto/connectorcommands/adiom/commands/connectors/v1/devrandom.protoproto/connectorcommands/adiom/commands/connectors/v1/dynamodb.protoproto/connectorcommands/adiom/commands/connectors/v1/fakesource.protoproto/connectorcommands/adiom/commands/connectors/v1/file.protoproto/connectorcommands/adiom/commands/connectors/v1/grpc.protoproto/connectorcommands/adiom/commands/connectors/v1/kafka.protoproto/connectorcommands/adiom/commands/connectors/v1/mongo.protoproto/connectorcommands/adiom/commands/connectors/v1/postgres.protoproto/connectorcommands/adiom/commands/connectors/v1/s3.protoproto/connectorcommands/adiom/commands/connectors/v1/s3vectors.protoproto/connectorcommands/adiom/commands/connectors/v1/sqlbatch.protoproto/dsync/adiom/v1/adiom.protoproto/dsync/adiom/v1/messages.protoproto/dsync/adiom/v1/vector.protoscripts/devsetup.sh
✅ Files skipped from review due to trivial changes (19)
- .gitignore
- scripts/devsetup.sh
- buf.gen.commands.yaml
- proto/connectorcommands/adiom/commands/connectors/v1/devrandom.proto
- internal/app/options/grpc.go
- buf.gen.yaml
- proto/connectorcommands/adiom/commands/connectors/v1/s3vectors.proto
- proto/connectorcommands/adiom/commands/connectors/v1/sqlbatch.proto
- proto/connectorcommands/adiom/commands/connectors/v1/file.proto
- internal/app/options/flags.go
- proto/connectorcommands/adiom/commands/connectors/v1/dynamodb.proto
- proto/connectorcommands/adiom/commands/connectors/v1/s3.proto
- proto/connectorcommands/adiom/commands/connectors/v1/postgres.proto
- proto/connectorcommands/adiom/commands/connectors/v1/cosmos.proto
- .github/workflows/test.yml
- proto/connectorcommands/adiom/commands/connectors/v1/mongo.proto
- proto/connectorcommands/adiom/commands/connectors/v1/kafka.proto
- proto/connectorcommands/adiom/commands/connectors/v1/devnull.proto
- proto/connectorcommands/adiom/commands/connectors/v1/fakesource.proto
🚧 Files skipped from review as they are similar to previous changes (3)
- go.mod
- buf.yaml
- proto/connectorcommands/adiom/commands/connectors/v1/grpc.proto
This file contains hidden or 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
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary by CodeRabbit
New Features
Infrastructure Updates
Developer Experience