fix(channel): send DISCONNECT packet if connection has been kicked

fix #7241

apps/emqx/src/emqx_channel.erl

@@ -1143,7 +1143,14 @@ return_sub_unsub_ack(Packet, Channel) ->
     | {shutdown, Reason :: term(), Reply :: term(), emqx_types:packet(), channel()}.
 handle_call(kick, Channel) ->
     Channel1 = ensure_disconnected(kicked, Channel),
-    disconnect_and_shutdown(kicked, ok, Channel1);
+    case Channel1 of
+        ?IS_MQTT_V5 ->
+            shutdown(kicked, ok,
+                     ?DISCONNECT_PACKET(?RC_ADMINISTRATIVE_ACTION), Channel1);
+        _ ->
+            shutdown(kicked, ok, Channel1)
+    end;
+
 handle_call(discard, Channel) ->
     disconnect_and_shutdown(discarded, ok, Channel);
 %% Session Takeover

apps/emqx/test/emqx_channel_SUITE.erl

@@ -907,7 +907,12 @@ t_handle_out_unexpected(_) ->
 %%--------------------------------------------------------------------
 
 t_handle_call_kick(_) ->
-    {shutdown, kicked, ok, _Chan} = emqx_channel:handle_call(kick, channel()).
+    Channelv5 = channel(),
+    Channelv4 = v4(Channelv5),
+    {shutdown, kicked, ok, _} = emqx_channel:handle_call(kick, Channelv4),
+    {shutdown, kicked, ok,
+     ?DISCONNECT_PACKET(?RC_ADMINISTRATIVE_ACTION),
+     _} = emqx_channel:handle_call(kick, Channelv5).
 
 t_handle_call_discard(_) ->
     Packet = ?DISCONNECT_PACKET(?RC_SESSION_TAKEN_OVER),
@@ -1243,3 +1248,11 @@ quota() ->
     emqx_limiter_container:get_limiter_by_names([message_routing], limiter_cfg()).
 
 limiter_cfg() -> #{message_routing => default}.
+
+v4(Channel) ->
+    ConnInfo = emqx_channel:info(conninfo, Channel),
+    emqx_channel:set_field(
+      conninfo,
+      maps:put(proto_ver, ?MQTT_PROTO_V4, ConnInfo),
+      Channel
+     ).