fix(kinesis): set validators for static constraints

Fixes https://emqx.atlassian.net/browse/EMQX-10833

apps/emqx_bridge_kinesis/src/emqx_bridge_kinesis.erl

@@ -30,8 +30,27 @@ roots() ->
 
 fields("config_producer") ->
     emqx_bridge_schema:common_bridge_fields() ++
-        emqx_resource_schema:fields("resource_opts") ++
-        fields(connector_config) ++ fields(producer);
+        fields("resource_opts") ++
+        fields(connector_config) ++
+        fields(producer);
+fields("resource_opts") ->
+    [
+        {resource_opts,
+            mk(
+                ref(?MODULE, "creation_opts"),
+                #{
+                    required => false,
+                    default => #{},
+                    desc => ?DESC(emqx_resource_schema, "creation_opts")
+                }
+            )}
+    ];
+fields("creation_opts") ->
+    emqx_resource_schema:create_opts([
+        {batch_size, #{
+            validator => emqx_resource_validator:max(int, 500)
+        }}
+    ]);
 fields(connector_config) ->
     [
         {aws_access_key_id,
@@ -121,6 +140,8 @@ fields("put_producer") ->
 
 desc("config_producer") ->
     ?DESC("desc_config");
+desc("creation_opts") ->
+    ?DESC(emqx_resource_schema, "creation_opts");
 desc(_) ->
     undefined.
 
@@ -161,6 +182,8 @@ mk(Type, Meta) -> hoconsc:mk(Type, Meta).
 
 enum(OfSymbols) -> hoconsc:enum(OfSymbols).
 
+ref(Module, Name) -> hoconsc:ref(Module, Name).
+
 type_field_producer() ->
     {type, mk(enum([kinesis_producer]), #{required => true, desc => ?DESC("desc_type")})}.
 

apps/emqx_bridge_kinesis/test/emqx_bridge_kinesis_impl_producer_SUITE.erl

@@ -902,3 +902,17 @@ t_empty_payload_template(Config) ->
         emqx_utils_json:decode(Data, [return_maps])
     ),
     ok.
+
+t_validate_static_constraints(Config) ->
+    % From <https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecords.html>:
+    % "Each PutRecords request can support up to 500 records.
+    %  Each record in the request can be as large as 1 MiB,
+    %  up to a limit of 5 MiB for the entire request, including partition keys."
+    %
+    % Message size and request size shall be controlled by user, so there is no validators
+    % for them - if exceeded, it will fail like on `t_publish_big_msg` test.
+    ?assertThrow(
+        {emqx_bridge_schema, [#{kind := validation_error, value := 501}]},
+        generate_config([{batch_size, 501} | Config])
+    ),
+    ok.

changes/ee/fix-11494.en.md

@@ -0,0 +1 @@
+Added schema validator to reflect Amazon Kinesis' static constraint: batch request can support up to 500 records (max batch size);