From 20b46091f928c4a6e0f364390624f343c0cb8c44 Mon Sep 17 00:00:00 2001
From: Paulo Zulato <paulozulato@gmail.com>
Date: Mon, 21 Aug 2023 21:51:46 -0300
Subject: [PATCH] fix(kinesis): set validators for static constraints

Fixes https://emqx.atlassian.net/browse/EMQX-10833
---
 .../src/emqx_bridge_kinesis.erl               | 27 +++++++++++++++++--
 ...mqx_bridge_kinesis_impl_producer_SUITE.erl | 14 ++++++++++
 changes/ee/fix-11494.en.md                    |  1 +
 3 files changed, 40 insertions(+), 2 deletions(-)
 create mode 100644 changes/ee/fix-11494.en.md

diff --git a/apps/emqx_bridge_kinesis/src/emqx_bridge_kinesis.erl b/apps/emqx_bridge_kinesis/src/emqx_bridge_kinesis.erl
index 92faa2279..d98e7ab11 100644
--- a/apps/emqx_bridge_kinesis/src/emqx_bridge_kinesis.erl
+++ b/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")})}.
 
diff --git a/apps/emqx_bridge_kinesis/test/emqx_bridge_kinesis_impl_producer_SUITE.erl b/apps/emqx_bridge_kinesis/test/emqx_bridge_kinesis_impl_producer_SUITE.erl
index 69b4a2d86..ea926fc33 100644
--- a/apps/emqx_bridge_kinesis/test/emqx_bridge_kinesis_impl_producer_SUITE.erl
+++ b/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.
diff --git a/changes/ee/fix-11494.en.md b/changes/ee/fix-11494.en.md
new file mode 100644
index 000000000..5ff887562
--- /dev/null
+++ b/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);