Merge pull request #9581 from zmstone/1219-fix-mqtt-bridge-config

1219 fix mqtt bridge config

apps/emqx_bridge/src/schema/emqx_bridge_mqtt_config.erl

@@ -86,7 +86,9 @@ default_resource_opts() ->
         <<"health_check_interval">> => <<"15s">>,
         <<"max_queue_bytes">> => <<"1GB">>,
         <<"query_mode">> => <<"sync">>,
-        <<"worker_pool_size">> => 16
+        %% there is only one underlying MQTT connection
+        %% doesn't make a lot of sense to have a large pool
+        <<"worker_pool_size">> => 4
     }.
 
 egress(Config) ->

apps/emqx_connector/src/mqtt/emqx_connector_mqtt_schema.erl

@@ -255,7 +255,8 @@ fields("egress_remote") ->
             mk(
                 qos(),
                 #{
-                    required => true,
+                    required => false,
+                    default => 1,
                     desc => ?DESC("egress_remote_qos")
                 }
             )},
@@ -263,7 +264,8 @@ fields("egress_remote") ->
             mk(
                 hoconsc:union([boolean(), binary()]),
                 #{
-                    required => true,
+                    required => false,
+                    default => false,
                     desc => ?DESC("retain")
                 }
             )},

apps/emqx_resource/i18n/emqx_resource_schema_i18n.conf

@@ -24,12 +24,13 @@ emqx_resource_schema {
 
   worker_pool_size {
     desc {
-      en: """Resource worker pool size."""
-      zh: """资源连接池大小。"""
+      en: """The number of buffer workers. Only applicable for egress type bridges.
+For bridges only have ingress direction data flow, it can be set to 0 otherwise must be greater than 0."""
+      zh: """缓存队列 worker 数量。仅对 egress 类型的桥接有意义。当桥接仅有 ingress 方向时，可设置为 0，否则必须大于 0）。"""
     }
     label {
-      en: """Worker Pool Size"""
-      zh: """资源连接池大小"""
+      en: """Buffer Pool Size"""
+      zh: """缓存池大小"""
     }
   }
 
@@ -101,12 +102,14 @@ emqx_resource_schema {
 
   enable_queue {
     desc {
-      en: """Queue mode enabled."""
-      zh: """启用队列模式。"""
+      en: """Enable disk buffer queue (only applicable for egress bridges).
+When Enabled, messages will be buffered on disk when the bridge connection is down.
+When disabled the messages are buffered in RAM only."""
+      zh: """启用磁盘缓存队列（仅对 egress 方向桥接有用）。"""
     }
     label {
-      en: """Enable queue"""
-      zh: """启用队列模式"""
+      en: """Enable disk buffer queue"""
+      zh: """启用磁盘缓存队列"""
     }
   }
 
@@ -145,12 +148,12 @@ emqx_resource_schema {
 
   max_queue_bytes {
     desc {
-      en: """Maximum queue storage."""
-      zh: """消息队列的最大长度。"""
+      en: """Maximum number of bytes to buffer for each buffer worker."""
+      zh: """每个缓存 worker 允许使用的最大字节数。"""
     }
     label {
-      en: """Queue max bytes"""
-      zh: """队列最大长度"""
+      en: """Max buffer queue size"""
+      zh: """缓存队列最大长度"""
     }
   }
 

apps/emqx_resource/include/emqx_resource.hrl

@@ -43,15 +43,15 @@
 }.
 -type resource_group() :: binary().
 -type creation_opts() :: #{
-    %%======================================= Deprecated Opts:
+    %%======================================= Deprecated Opts BEGIN
     %% use health_check_interval instead
     health_check_timeout => integer(),
     %% use start_timeout instead
     wait_for_resource_ready => integer(),
     %% use auto_restart_interval instead
     auto_retry_interval => integer(),
-    %%======================================= Deprecated Opts End
-    worker_pool_size => pos_integer(),
+    %%======================================= Deprecated Opts END
+    worker_pool_size => non_neg_integer(),
     %% use `integer()` compatibility to release 5.0.0 bpapi
     health_check_interval => integer(),
     %% We can choose to block the return of emqx_resource:start until

apps/emqx_resource/src/emqx_resource_worker.erl

@@ -117,13 +117,16 @@ init({Id, Index, Opts}) ->
     true = gproc_pool:connect_worker(Id, {Id, Index}),
     Name = name(Id, Index),
     BatchSize = maps:get(batch_size, Opts, ?DEFAULT_BATCH_SIZE),
+    SegBytes0 = maps:get(queue_seg_bytes, Opts, ?DEFAULT_QUEUE_SEG_SIZE),
+    TotalBytes = maps:get(max_queue_bytes, Opts, ?DEFAULT_QUEUE_SIZE),
+    SegBytes = min(SegBytes0, TotalBytes),
     Queue =
         case maps:get(enable_queue, Opts, false) of
             true ->
                 replayq:open(#{
                     dir => disk_queue_dir(Id, Index),
-                    seg_bytes => maps:get(queue_seg_bytes, Opts, ?DEFAULT_QUEUE_SEG_SIZE),
-                    max_total_bytes => maps:get(max_queue_bytes, Opts, ?DEFAULT_QUEUE_SIZE),
+                    seg_bytes => SegBytes,
+                    max_total_bytes => TotalBytes,
                     sizer => fun ?MODULE:estimate_size/1,
                     marshaller => fun ?MODULE:queue_item_marshaller/1
                 });

apps/emqx_resource/src/schema/emqx_resource_schema.erl

@@ -56,7 +56,7 @@ fields("creation_opts") ->
         {max_queue_bytes, fun max_queue_bytes/1}
     ].
 
-worker_pool_size(type) -> pos_integer();
+worker_pool_size(type) -> non_neg_integer();
 worker_pool_size(desc) -> ?DESC("worker_pool_size");
 worker_pool_size(default) -> ?WORKER_POOL_SIZE;
 worker_pool_size(required) -> false;