From e837791f94054763038151ac0256aa410814835a Mon Sep 17 00:00:00 2001 From: Thales Macedo Garitezi Date: Thu, 21 Mar 2024 11:49:00 -0300 Subject: [PATCH] fix(kafka-based bridges): avoid trying to get raw config for replayq dir Fixes https://emqx.atlassian.net/browse/EMQX-12049 --- .../src/emqx_bridge_kafka_impl_producer.erl | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/apps/emqx_bridge_kafka/src/emqx_bridge_kafka_impl_producer.erl b/apps/emqx_bridge_kafka/src/emqx_bridge_kafka_impl_producer.erl index 20241fdcd..6bb1690ff 100644 --- a/apps/emqx_bridge_kafka/src/emqx_bridge_kafka_impl_producer.erl +++ b/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()