Add kafka connection properties to profile (#791)

crates/arroyo-connectors/src/confluent/mod.rs

@@ -84,6 +84,7 @@ impl From<ConfluentProfile> for KafkaConfig {
                 password: c.secret,
             },
             schema_registry_enum: Some(c.schema_registry.into()),
+            connection_properties: HashMap::new(),
         }
     }
 }

crates/arroyo-connectors/src/kafka/mod.rs

@@ -105,6 +105,7 @@ impl KafkaConnector {
             authentication: auth,
             bootstrap_servers: BootstrapServers(pull_opt("bootstrap_servers", options)?),
             schema_registry_enum: schema_registry,
+            connection_properties: HashMap::new(),
         })
     }
 
@@ -926,13 +927,21 @@ pub fn client_configs(
         }
     };
 
+    for (k, v) in connection.connection_properties.iter() {
+        client_configs.insert(k.to_string(), v.to_string());
+    }
+
     if let Some(table) = table {
-        client_configs.extend(
-            table
-                .client_configs
-                .iter()
-                .map(|(k, v)| (k.to_string(), v.to_string())),
-        );
+        for (k, v) in table.client_configs.iter() {
+            if connection.connection_properties.contains_key(k) {
+                warn!(
+                    "rdkafka config key {:?} defined in both connection and table config",
+                    k
+                );
+            }
+
+            client_configs.insert(k.to_string(), v.to_string());
+        }
     }
 
     Ok(client_configs)

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

@@ -106,6 +106,18 @@
                     "sensitive": ["apiSecret"]
                 }
             ]
+        },
+        "connectionProperties": {
+            "type": "object",
+            "title": "Connection Properties",
+            "description": "Key-value pairs of rdkafka configuration options",
+            "items": {
+                "type": "string",
+                "title": "property"
+            },
+            "additionalProperties": {
+                "type": "string"
+            }
         }
     },
     "required": ["bootstrapServers", "authentication"]