Support per-table schema registry with authentication

Add an extra option for tables with the Kafka engine to pass the URL of a
schema registry. Also add support for schema registry basic authentication
via username:password in the URL.

Changes:
- Add kafka_format_avro_schema_registry_url setting to KafkaSettings
- Store format_avro_schema_registry_url in StorageKafka class
- Propagate schema registry URL to format settings via
  createSettingsAdjustments() (works for both StorageKafka and StorageKafka2)
- Add basic authentication parsing in AvroRowInputFormat::SchemaRegistry
  to extract username:password from URL and authenticate HTTP requests

This allows per-table configuration of Avro Schema Registry URLs with
authentication, enabling different Kafka tables to use different schema
registries or credentials.

Usage example:
  CREATE TABLE kafka_table (...) ENGINE = Kafka(...) SETTINGS
    kafka_format_avro_schema_registry_url = 'http://user:pass@registry:8081';

Co-authored-by: Kevin Michel <kevin.michel@aiven.io>

Author: tilman-aiven

src/Processors/Formats/Impl/AvroRowInputFormat.cpp

@@ -1083,6 +1083,21 @@ class AvroConfluentRowInputFormat::SchemaRegistry
                     .withReceiveTimeout(1);
 
                 Poco::Net::HTTPRequest request(Poco::Net::HTTPRequest::HTTP_GET, url.getPathAndQuery(), Poco::Net::HTTPRequest::HTTP_1_1);
+                const auto & user_info = base_url.getUserInfo();
+                if (!user_info.empty())
+                {
+                    std::size_t n = user_info.find(':');
+                    if (n != std::string::npos)
+                    {
+                        Poco::Net::HTTPBasicCredentials credentials;
+                        credentials.setUsername(user_info.substr(0, n));
+                        credentials.setPassword(user_info.substr(n + 1));
+                        if (!credentials.getUsername().empty())
+                        {
+                            credentials.authenticate(request);
+                        }
+                    }
+                }
                 if (url.getPort())
                     request.setHost(url.getHost(), url.getPort());
                 else

src/Storages/Kafka/KafkaSettings.cpp

@@ -24,6 +24,7 @@ namespace ErrorCodes
     DECLARE(String, kafka_group_name, "", "Client group id string. All Kafka consumers sharing the same group.id belong to the same group.", 0) \
     /* those are mapped to format factory settings */ \
     DECLARE(String, kafka_format, "", "The message format for Kafka engine.", 0) \
+    DECLARE(String, kafka_format_avro_schema_registry_url, "", "For AvroConfluent format: Kafka Schema Registry URL.", 0) \
     DECLARE(String, kafka_schema, "", "Schema identifier (used by schema-based formats) for Kafka engine", 0) \
     DECLARE(UInt64, kafka_num_consumers, 1, "The number of consumers per table for Kafka engine.", 0) \
     /* default is = max_insert_block_size / kafka_num_consumers  */ \

src/Storages/Kafka/StorageKafka.cpp

@@ -98,6 +98,7 @@ namespace KafkaSetting
     extern const KafkaSettingsString kafka_ssl_key_location;
     extern const KafkaSettingsBool kafka_thread_per_consumer;
     extern const KafkaSettingsString kafka_topic_list;
+    extern const KafkaSettingsString kafka_format_avro_schema_registry_url;
 }
 
 namespace ErrorCodes
@@ -188,6 +189,7 @@ StorageKafka::StorageKafka(
               ? StorageKafkaUtils::getDefaultClientId(table_id_)
               : getContext()->getMacros()->expand((*kafka_settings)[KafkaSetting::kafka_client_id].value, macros_info))
     , format_name(getContext()->getMacros()->expand((*kafka_settings)[KafkaSetting::kafka_format].value))
+    , format_avro_schema_registry_url((*kafka_settings)[KafkaSetting::kafka_format_avro_schema_registry_url].value)
     , max_rows_per_message((*kafka_settings)[KafkaSetting::kafka_max_rows_per_message].value)
     , schema_name(getContext()->getMacros()->expand((*kafka_settings)[KafkaSetting::kafka_schema].value, macros_info))
     , num_consumers((*kafka_settings)[KafkaSetting::kafka_num_consumers].value)

src/Storages/Kafka/StorageKafka.h

@@ -107,6 +107,7 @@ class StorageKafka final : public IStorage, WithContext
     const String group;
     const String client_id;
     const String format_name;
+    const String format_avro_schema_registry_url;
     const size_t max_rows_per_message;
     const String schema_name;
     const size_t num_consumers; /// total number of consumers

src/Storages/Kafka/StorageKafkaUtils.cpp

@@ -82,6 +82,7 @@ namespace KafkaSetting
     extern const KafkaSettingsUInt64 kafka_skip_broken_messages;
     extern const KafkaSettingsBool kafka_thread_per_consumer;
     extern const KafkaSettingsString kafka_topic_list;
+    extern const KafkaSettingsString kafka_format_avro_schema_registry_url;
 }
 
 using namespace std::chrono_literals;
@@ -520,6 +521,11 @@ SettingsChanges createSettingsAdjustments(KafkaSettings & kafka_settings, const
     result.setSetting("input_format_csv_detect_header", false);
     result.setSetting("input_format_tsv_detect_header", false);
     result.setSetting("input_format_custom_detect_header", false);
+    const String & format_avro_schema_registry_url = kafka_settings[KafkaSetting::kafka_format_avro_schema_registry_url].value;
+    if (!format_avro_schema_registry_url.empty())
+    {
+        result.emplace_back("format_avro_schema_registry_url", format_avro_schema_registry_url);
+    }
 
     return result;
 }