Merge pull request #3565 from rabbitmq/stream-queue-handle-unsupported-header-values

Stream queue: handle unsupported header value types

Author: michaelklishin

deps/rabbit/src/rabbit_msg_record.erl

@@ -195,7 +195,7 @@ from_amqp091(#'P_basic'{message_id = MsgId,
                                                <- case Headers of
                                                       undefined -> [];
                                                       _ -> Headers
-                                                  end],
+                                                  end, not unsupported_header_value_type(T)],
     %% properties that do not map directly to AMQP 1.0 properties are stored
     %% in application properties
     APC = map_add(utf8, <<"x-basic-type">>, utf8, Type,
@@ -395,6 +395,13 @@ message_id({utf8, S}, HKey, H0) ->
 message_id(MsgId, _, H) ->
     {H, unwrap(MsgId)}.
 
+ unsupported_header_value_type(array) ->
+     true;
+ unsupported_header_value_type(table) ->
+     true;
+ unsupported_header_value_type(_) ->
+     false.
+
 -ifdef(TEST).
 -include_lib("eunit/include/eunit.hrl").
 -endif.

deps/rabbit/test/rabbit_msg_record_SUITE.erl

@@ -24,6 +24,7 @@ all() ->
 all_tests() ->
     [
      ampq091_roundtrip,
+     unsupported_091_header_is_dropped,
      message_id_ulong,
      message_id_uuid,
      message_id_binary,
@@ -90,6 +91,22 @@ ampq091_roundtrip(_Config) ->
     test_amqp091_roundtrip(#'P_basic'{}, Payload),
     ok.
 
+unsupported_091_header_is_dropped(_Config) ->
+    Props = #'P_basic'{
+                       headers = [
+                                  {<<"x-received-from">>, array, []}
+                                 ]
+                      },
+    MsgRecord0 = rabbit_msg_record:from_amqp091(Props, <<"payload">>),
+    MsgRecord = rabbit_msg_record:init(
+                  iolist_to_binary(rabbit_msg_record:to_iodata(MsgRecord0))),
+    % meck:unload(),
+    {PropsOut, <<"payload">>} = rabbit_msg_record:to_amqp091(MsgRecord),
+
+    ?assertMatch(#'P_basic'{headers = undefined}, PropsOut),
+
+    ok.
+
 message_id_ulong(_Config) ->
     Num = 9876789,
     ULong = erlang:integer_to_binary(Num),