Merge pull request #12805 from thalesmg/port-fix-12755-r56-20240328

fix(kafka-based bridges): avoid trying to get raw config for replayq dir

apps/emqx_bridge_kafka/src/emqx_bridge_kafka.app.src

@@ -1,7 +1,7 @@
 %% -*- mode: erlang -*-
 {application, emqx_bridge_kafka, [
     {description, "EMQX Enterprise Kafka Bridge"},
-    {vsn, "0.2.2"},
+    {vsn, "0.2.3"},
     {registered, [emqx_bridge_kafka_consumer_sup]},
     {applications, [
         kernel,

apps/emqx_bridge_kafka/src/emqx_bridge_kafka_impl_producer.erl

@@ -668,9 +668,8 @@ partitioner(random) -> random;
 partitioner(key_dispatch) -> first_key_dispatch.
 
 replayq_dir(BridgeType, BridgeName) ->
-    RawConf = emqx_conf:get_raw([actions, BridgeType, BridgeName]),
     DirName = iolist_to_binary([
-        emqx_bridge_lib:downgrade_type(BridgeType, RawConf),
+        maybe_v1_type_name(BridgeType),
         ":",
         BridgeName,
         ":",
@@ -678,6 +677,14 @@ replayq_dir(BridgeType, BridgeName) ->
     ]),
     filename:join([emqx:data_dir(), "kafka", DirName]).
 
+%% To avoid losing queued data on disk, we must use the same directory as the old v1
+%% bridges, if any.  Among the Kafka-based bridges that exist since v1, only Kafka changed
+%% its type name.  Other bridges are either unchanged, or v2-only, and should use their v2
+%% type names.
+maybe_v1_type_name(Type) when is_atom(Type) -> maybe_v1_type_name(atom_to_binary(Type));
+maybe_v1_type_name(<<"kafka_producer">>) -> <<"kafka">>;
+maybe_v1_type_name(Type) -> Type.
+
 with_log_at_error(Fun, Log) ->
     try
         Fun()