fix(kafka): Fix server name indication config for Kafka

Also impoved azur event hub's doc for SNI

apps/emqx_bridge_azure_event_hub/src/emqx_bridge_azure_event_hub.erl

@@ -98,9 +98,9 @@ fields(auth_username_password) ->
         auth_overrides()
     ),
     override_documentations(Fields);
-fields("ssl_client_opts") ->
+fields(ssl_client_opts) ->
     Fields = override(
-        emqx_schema:fields("ssl_client_opts"),
+        emqx_bridge_kafka:ssl_client_opts_fields(),
         ssl_overrides()
     ),
     override_documentations(Fields);
@@ -138,8 +138,6 @@ desc("config_connector") ->
     ?DESC("desc_config");
 desc("config_producer") ->
     ?DESC("desc_config");
-desc("ssl_client_opts") ->
-    emqx_schema:desc("ssl_client_opts");
 desc("get_" ++ Type) when Type == "producer"; Type == "connector"; Type == "bridge_v2" ->
     ["Configuration for Azure Event Hub using `GET` method."];
 desc("put_" ++ Type) when Type == "producer"; Type == "connector"; Type == "bridge_v2" ->
@@ -155,7 +153,8 @@ struct_names() ->
         auth_username_password,
         kafka_message,
         producer_kafka_opts,
-        bridge_v2
+        bridge_v2,
+        ssl_client_opts
     ].
 
 bridge_v2_examples(Method) ->
@@ -319,7 +318,7 @@ connector_overrides() ->
                     )
                 }
             ),
-        ssl => mk(ref("ssl_client_opts"), #{default => #{<<"enable">> => true}}),
+        ssl => mk(ref(ssl_client_opts), #{default => #{<<"enable">> => true}}),
         type => mk(
             ?AEH_CONNECTOR_TYPE,
             #{
@@ -355,7 +354,7 @@ producer_overrides() ->
                 required => true,
                 validator => fun emqx_bridge_kafka:producer_strategy_key_validator/1
             }),
-        ssl => mk(ref("ssl_client_opts"), #{default => #{<<"enable">> => true}}),
+        ssl => mk(ref(ssl_client_opts), #{default => #{<<"enable">> => true}}),
         type => mk(azure_event_hub_producer, #{required => true})
     }.
 
@@ -366,7 +365,7 @@ bridge_v2_overrides() ->
                 required => true,
                 validator => fun emqx_bridge_kafka:producer_strategy_key_validator/1
             }),
-        ssl => mk(ref("ssl_client_opts"), #{default => #{<<"enable">> => true}}),
+        ssl => mk(ref(ssl_client_opts), #{default => #{<<"enable">> => true}}),
         type => mk(
             ?AEH_CONNECTOR_TYPE,
             #{
@@ -391,19 +390,11 @@ auth_overrides() ->
             })
     }.
 
+%% Kafka has SSL disabled by default
+%% Azure must use SSL
 ssl_overrides() ->
     #{
-        %% FIXME: change this once the config option is defined
-        %% "cacerts" => mk(boolean(), #{default => true}),
-        "enable" => mk(true, #{default => true}),
-        "server_name_indication" =>
-            mk(
-                hoconsc:union([disable, auto, string()]),
-                #{
-                    example => auto,
-                    default => <<"auto">>
-                }
-            )
+        "enable" => mk(true, #{default => true})
     }.
 
 kafka_producer_overrides() ->

apps/emqx_bridge_kafka/src/emqx_bridge_kafka.erl

@@ -27,7 +27,8 @@
     roots/0,
     fields/1,
     desc/1,
-    host_opts/0
+    host_opts/0,
+    ssl_client_opts_fields/0
 ]).
 
 -export([kafka_producer_converter/2, producer_strategy_key_validator/1]).
@@ -195,6 +196,40 @@ values(consumer) ->
 %% -------------------------------------------------------------------------------------------------
 %% Hocon Schema Definitions
 
+%% In addition to the common ssl client options defined in emqx_schema module
+%% Kafka supports a special value 'auto' in order to support different bootstrap endpoints
+%% as well as partition leaders.
+%% A static SNI is quite unusual for Kafka, but it's kept anyway.
+ssl_overrides() ->
+    #{
+        "server_name_indication" =>
+            mk(
+                hoconsc:union([auto, disable, string()]),
+                #{
+                    example => auto,
+                    default => <<"auto">>,
+                    importance => ?IMPORTANCE_LOW,
+                    desc => ?DESC("server_name_indication")
+                }
+            )
+    }.
+
+override(Fields, Overrides) ->
+    lists:map(
+        fun({Name, Sc}) ->
+            case maps:find(Name, Overrides) of
+                {ok, Override} ->
+                    {Name, hocon_schema:override(Sc, Override)};
+                error ->
+                    {Name, Sc}
+            end
+        end,
+        Fields
+    ).
+
+ssl_client_opts_fields() ->
+    override(emqx_schema:client_ssl_opts_schema(#{}), ssl_overrides()).
+
 host_opts() ->
     #{default_port => 9092}.
 
@@ -266,8 +301,11 @@ fields("config") ->
             mk(hoconsc:union([none, ref(auth_username_password), ref(auth_gssapi_kerberos)]), #{
                 default => none, desc => ?DESC("authentication")
             })},
-        {socket_opts, mk(ref(socket_opts), #{required => false, desc => ?DESC(socket_opts)})}
-    ] ++ emqx_connector_schema_lib:ssl_fields();
+        {socket_opts, mk(ref(socket_opts), #{required => false, desc => ?DESC(socket_opts)})},
+        {ssl, mk(ref(ssl_client_opts), #{})}
+    ];
+fields(ssl_client_opts) ->
+    ssl_client_opts_fields();
 fields(auth_username_password) ->
     [
         {mechanism,
@@ -571,7 +609,8 @@ struct_names() ->
         consumer_opts,
         consumer_kafka_opts,
         consumer_topic_mapping,
-        producer_kafka_ext_headers
+        producer_kafka_ext_headers,
+        ssl_client_opts
     ].
 
 %% -------------------------------------------------------------------------------------------------

rel/i18n/emqx_bridge_azure_event_hub.hocon

@@ -325,4 +325,18 @@ desc_config.desc:
 desc_config.label:
 """Azure Event Hub Bridge Configuration"""
 
+ssl_client_opts.desc:
+"""TLS/SSL options for Azure Event Hub client."""
+ssl_client_opts.label:
+"""TLS/SSL options"""
+
+server_name_indication.desc:
+"""Server Name Indication (SNI) setting for TLS handshake.<br/>
+- <code>auto</code>: The client will use <code>"servicebus.windows.net"</code> as SNI.<br/>
+- <code>disable</code>: If you wish to prevent the client from sending the SNI.<br/>
+- Other string values it will be sent as-is."""
+
+server_name_indication.label:
+"""SNI"""
+
 }

rel/i18n/emqx_bridge_kafka.hocon

@@ -436,4 +436,19 @@ kafka_producer_action.desc:
 kafka_producer_action.label:
 """Kafka Producer Action"""
 
+ssl_client_opts.desc:
+"""TLS/SSL options for Kafka client."""
+ssl_client_opts.label:
+"""TLS/SSL options"""
+
+server_name_indication.desc:
+"""Server Name Indication (SNI) setting for TLS handshake.<br/>
+- <code>auto</code>: Allow the client to automatically determine the appropriate SNI.<br/>
+- <code>disable</code>: If you wish to prevent the client from sending the SNI.<br/>
+- Other string values will be sent as-is."""
+
+server_name_indication.label:
+"""SNI"""
+
+
 }

rel/i18n/emqx_schema.hocon

@@ -326,7 +326,7 @@ which accepts the connection and performs TLS handshake may differ from the
 host the TLS client initially connects to, e.g. when connecting to an IP address
 or when the host has multiple resolvable DNS records <br/>
 If not specified, it will default to the host name string which is used
-to establish the connection, unless it is IP addressed used.<br/>
+to establish the connection, unless it is IP address used.<br/>
 The host name is then also used in the host name verification of the peer
 certificate.<br/> The special value 'disable' prevents the Server Name
 Indication extension from being sent and disables the hostname