fix(s3-aggeg): apply CSV column order setting consistently

Otherwise, columns that are part of column order could appear and
disappear from consecutive uploads, depending on if they are part
of the very first buffered event or not.

apps/emqx_bridge_s3/src/emqx_bridge_s3_aggreg_csv.erl

@@ -58,9 +58,8 @@ close(#csv{}) ->
 
 mk_columns(Record, #csv{order = ColumnOrder}) ->
     Columns = lists:sort(maps:keys(Record)),
-    OrderedFirst = [CO || CO <- ColumnOrder, lists:member(CO, Columns)],
     Unoredered = Columns -- ColumnOrder,
-    OrderedFirst ++ Unoredered.
+    ColumnOrder ++ Unoredered.
 
 -spec emit_header([column()], container()) -> iodata().
 emit_header([C], #csv{delimiter = Delim}) ->

apps/emqx_bridge_s3/test/emqx_bridge_s3_aggreg_upload_SUITE.erl

@@ -26,7 +26,8 @@
     <<"publish_received_at">>,
     <<"clientid">>,
     <<"topic">>,
-    <<"payload">>
+    <<"payload">>,
+    <<"empty">>
     | T
 ]).
 
@@ -185,9 +186,9 @@ t_aggreg_upload(Config) ->
     ?assertMatch(
         {ok, [
             ?CONF_COLUMN_ORDER(_),
-            [TS, <<"C1">>, T1, P1 | _],
-            [TS, <<"C2">>, T2, P2 | _],
-            [TS, <<"C3">>, T3, P3 | _]
+            [TS, <<"C1">>, T1, P1, <<>> | _],
+            [TS, <<"C2">>, T2, P2, <<>> | _],
+            [TS, <<"C3">>, T3, P3, <<>> | _]
         ]},
         erl_csv:decode(Content)
     ).