Provide clear error message for reserved annotation keys

As described in https://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-messaging-v1.0-os.html#type-annotations
> The annotations type is a map where the keys are restricted to be of type symbol or of type ulong.
> All ulong keys, and all symbolic keys except those beginning with "x-" are reserved.

Prior to this commit, if an AMQP client used a reserved annotation key,
the entire AMQP connection terminated with a function_clause error
message that might be difficult to understand for client libs:
```
<<"Session error: function_clause\n[{amqp10_framing,'-decode_annotations/1-fun-0-',\n                 [{{symbol,<<\"aa\">>},{utf8,<<\"bbb\">>}}],\n                 [{file,\"amqp10_framing.erl\"},{line,158}]},\n {lists,map,2,[{file,\"lists.erl\"},{line,1559}]},\n {amqp10_framing,decode,1,[{file,\"amqp10_framing.erl\"},{line,127}]},\n {lists,map_1,2,[{file,\"lists.erl\"},{line,1564}]},\n {lists,map,2,[{file,\"lists.erl\"},{line,1559}]},\n {mc_amqp,init,1,[{file,\"mc_amqp.erl\"},{line,102}]},\n {mc,init,4,[{file,\"mc.erl\"},{line,150}]},\n {rabbit_amqp_session,incoming_link_transfer,4,\n                      [{file,\"rabbit_amqp_session.erl\"},{line,2341}]}]">>
```

This commit ends only the session and provides a clearer error message.

(cherry picked from commit 358ff79)

Author: ansd

deps/amqp10_common/src/amqp10_framing.erl

@@ -153,10 +153,12 @@ decode_map(Fields) ->
 %% or of type ulong. All ulong keys, and all symbolic keys except those beginning
 %% with "x-" are reserved." [3.2.10]
 %% Since we already parse annotations here and neither the client nor server uses
-%% reserved keys, we perform strict validation and crash if any reserved keys are used.
+%% reserved keys, we perform strict validation and throw if any reserved keys are used.
 decode_annotations(Fields) ->
     lists:map(fun({{symbol, <<"x-", _/binary>>} = K, V}) ->
-                      {K, decode(V)}
+                      {K, decode(V)};
+                 ({ReservedKey, _V}) ->
+                      throw({reserved_annotation_key, ReservedKey})
               end, Fields).
 
 -spec encode_described(list | map | binary | annotations | '*',

deps/rabbit/src/rabbit_amqp_session.erl

@@ -508,12 +508,16 @@ handle_cast({frame_body, FrameBody},
             noreply(State);
         {stop, _, _} = Stop ->
             Stop
-    catch exit:#'v1_0.error'{} = Error ->
-              log_error_and_close_session(Error, State0);
-          exit:normal ->
+    catch exit:normal ->
               {stop, normal, State0};
+          exit:#'v1_0.error'{} = Error ->
+              log_error_and_close_session(Error, State0);
           _:Reason:Stacktrace ->
-              {stop, {Reason, Stacktrace}, State0}
+              Description = unicode:characters_to_binary(
+                              lists:flatten(io_lib:format("~tp~n~tp", [Reason, Stacktrace]))),
+              Err = #'v1_0.error'{condition = ?V_1_0_AMQP_ERROR_INTERNAL_ERROR,
+                                  description = {utf8, Description}},
+              log_error_and_close_session(Err, State0)
     end;
 handle_cast({queue_event, _, _} = QEvent, State0) ->
     try handle_queue_event(QEvent, State0) of

deps/rabbit/test/amqp_client_SUITE.erl

@@ -144,7 +144,8 @@ groups() ->
        tcp_back_pressure_rabbitmq_internal_flow_classic_queue,
        tcp_back_pressure_rabbitmq_internal_flow_quorum_queue,
        session_max_per_connection,
-       link_max_per_session
+       link_max_per_session,
+       reserved_annotation
       ]},
 
      {cluster_size_3, [shuffle],
@@ -5894,6 +5895,32 @@ link_max_per_session(Config) ->
     flush(test_succeeded),
     ok = rpc(Config, application, set_env, [App, Par, Default]).
 
+reserved_annotation(Config) ->
+    OpnConf = connection_config(Config),
+    {ok, Connection} = amqp10_client:open_connection(OpnConf),
+    {ok, Session} = amqp10_client:begin_session(Connection),
+    TargetAddr = rabbitmq_amqp_address:exchange(<<"amq.fanout">>),
+    {ok, Sender} = amqp10_client:attach_sender_link_sync(
+                     Session, <<"sender">>, TargetAddr, settled),
+    ok = wait_for_credit(Sender),
+
+    Msg = amqp10_msg:set_message_annotations(
+            #{<<"reserved-key">> => 1},
+            amqp10_msg:new(<<"tag">>, <<"payload">>, true)),
+    ok = amqp10_client:send_msg(Sender, Msg),
+    receive
+        {amqp10_event,
+         {session, Session,
+          {ended,
+           #'v1_0.error'{description = {utf8, Description}}}}} ->
+            ?assertMatch(
+               <<"{reserved_annotation_key,{symbol,<<\"reserved-key\">>}}", _/binary>>,
+               Description)
+    after 5000 -> flush(missing_ended),
+                  ct:fail({missing_event, ?LINE})
+    end,
+    ok = close_connection_sync(Connection).
+
 %% internal
 %%