Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

add kafka connection properties to profile #791

Merged
merged 4 commits into from
Dec 3, 2024
Merged

add kafka connection properties to profile #791

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
c9b7383
add kafka connection properties to profile
emef Nov 22, 2024
f7568dd
PR feedback
emef Dec 2, 2024
9752a03
format
emef Dec 2, 2024
2f50d32
Merge remote-tracking branch 'upstream/master' into matt/kafka-conn-p…
emef Dec 3, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions crates/arroyo-connectors/src/confluent/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,6 +84,7 @@ impl From<ConfluentProfile> for KafkaConfig {
password: c.secret,
},
schema_registry_enum: Some(c.schema_registry.into()),
connection_properties: vec![],
}
}
}
Expand Down
9 changes: 9 additions & 0 deletions crates/arroyo-connectors/src/kafka/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -105,6 +105,7 @@ impl KafkaConnector {
authentication: auth,
bootstrap_servers: BootstrapServers(pull_opt("bootstrap_servers", options)?),
schema_registry_enum: schema_registry,
connection_properties: vec![],
})
}

Expand Down Expand Up @@ -926,6 +927,14 @@ pub fn client_configs(
}
};

for prop in connection.connection_properties.iter() {
if let Some((k, v)) = prop.split_once('=') {
client_configs.insert(k.to_string(), v.to_string());
} else {
bail!("invalid connection property: {}", prop);
}
}

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

With this change we'll be able to set client configs from both the connection and the table. I think it would be good to warn when iterating through the table configs (on 939) if both define the same property.

if let Some(table) = table {
client_configs.extend(
table
Expand Down
10 changes: 10 additions & 0 deletions crates/arroyo-connectors/src/kafka/profile.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -106,6 +106,16 @@
"sensitive": ["apiSecret"]
}
]
},
"connectionProperties": {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This would be a bit better as as map instead of a vec so that users don't need to know the k/v syntax. This is how we do property overrides on the table:

arroyo/crates/arroyo-connectors/src/kafka/table.json

Lines 83 to 90 in a9ac8f7

"client_configs": {
"type": "object",
"title": "Client Configs",
"description": "Additional Kafka configs to pass to the underlying Kafka consumer or producer. See [Kafka Consumer Configs](https://kafka.apache.org/documentation/#consumerconfigs) and [Kafka Producer Configs](https://kafka.apache.org/documentation/#producerconfigs) for more details.",
"additionalProperties": {
"type": "string"
}
},

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

👍 agreed map is better

"type": "array",
"title": "Connection Properties",
"description": "Key-value pairs of rdkafka configuration options",
"examples": ["client.id=arroyo"],
"items": {
"type": "string",
"title": "property"
}
}
},
"required": ["bootstrapServers", "authentication"]
Expand Down
Loading