From 6023012f8b4ee39e2491c2423c8f260063c92628 Mon Sep 17 00:00:00 2001 From: Thales Macedo Garitezi Date: Mon, 24 Jun 2024 10:02:26 -0300 Subject: [PATCH] fix(kafka and derivatives): add `alias` config to avoid clashes with same topic Fixes https://emqx.atlassian.net/browse/EMQX-12592 --- .../test/emqx_bridge_v2_testlib.erl | 39 +++++++- apps/emqx_bridge_azure_event_hub/rebar.config | 2 +- .../emqx_bridge_azure_event_hub_v2_SUITE.erl | 45 ++++++--- apps/emqx_bridge_confluent/rebar.config | 2 +- .../emqx_bridge_confluent_producer_SUITE.erl | 24 ++++- apps/emqx_bridge_kafka/rebar.config | 2 +- .../src/emqx_bridge_kafka_impl_producer.erl | 11 +++ .../emqx_bridge_v2_kafka_producer_SUITE.erl | 93 ++++++++++++++++++- changes/ee/breaking-13327.en.md | 3 + changes/ee/fix-13327.en.md | 1 + mix.exs | 2 +- 11 files changed, 202 insertions(+), 22 deletions(-) create mode 100644 changes/ee/breaking-13327.en.md create mode 100644 changes/ee/fix-13327.en.md diff --git a/apps/emqx_bridge/test/emqx_bridge_v2_testlib.erl b/apps/emqx_bridge/test/emqx_bridge_v2_testlib.erl index 8cf8730b0..41c184af9 100644 --- a/apps/emqx_bridge/test/emqx_bridge_v2_testlib.erl +++ b/apps/emqx_bridge/test/emqx_bridge_v2_testlib.erl @@ -288,6 +288,14 @@ request(Method, Path, Params) -> Error end. +simplify_result(Res) -> + case Res of + {error, {{_, Status, _}, _, Body}} -> + {Status, Body}; + {ok, {{_, Status, _}, _, Body}} -> + {Status, Body} + end. + list_bridges_api() -> Params = [], Path = emqx_mgmt_api_test_util:api_path(["actions"]), @@ -321,7 +329,7 @@ get_bridge_api(BridgeKind, BridgeType, BridgeName) -> Path = emqx_mgmt_api_test_util:api_path([Root, BridgeId]), ct:pal("get bridge ~p (via http)", [{BridgeKind, BridgeType, BridgeName}]), Res = request(get, Path, Params), - ct:pal("get bridge ~p result: ~p", [{BridgeKind, BridgeType, BridgeName}, Res]), + ct:pal("get bridge ~p result:\n ~p", [{BridgeKind, BridgeType, BridgeName}, Res]), Res. create_bridge_api(Config) -> @@ -349,6 +357,26 @@ create_kind_api(Config, Overrides) -> ct:pal("bridge create (~s, http) result:\n ~p", [Kind, Res]), Res. +enable_kind_api(Kind, ConnectorType, ConnectorName) -> + do_enable_disable_kind_api(Kind, ConnectorType, ConnectorName, enable). + +disable_kind_api(Kind, ConnectorType, ConnectorName) -> + do_enable_disable_kind_api(Kind, ConnectorType, ConnectorName, disable). + +do_enable_disable_kind_api(Kind, Type, Name, Op) -> + BridgeId = emqx_bridge_resource:bridge_id(Type, Name), + RootBin = api_path_root(Kind), + {OpPath, OpStr} = + case Op of + enable -> {"true", "enable"}; + disable -> {"false", "disable"} + end, + Path = emqx_mgmt_api_test_util:api_path([RootBin, BridgeId, "enable", OpPath]), + ct:pal(OpStr ++ " ~s ~s (http)", [Kind, BridgeId]), + Res = request(put, Path, []), + ct:pal(OpStr ++ " ~s ~s (http) result:\n ~p", [Kind, BridgeId, Res]), + simplify_result(Res). + create_connector_api(Config) -> create_connector_api(Config, _Overrides = #{}). @@ -453,6 +481,15 @@ update_bridge_api(Config, Overrides) -> ct:pal("update bridge (~s, http) result:\n ~p", [Kind, Res]), Res. +delete_kind_api(Kind, Type, Name) -> + BridgeId = emqx_bridge_resource:bridge_id(Type, Name), + PathRoot = api_path_root(Kind), + Path = emqx_mgmt_api_test_util:api_path([PathRoot, BridgeId]), + ct:pal("deleting bridge (~s, http)", [Kind]), + Res = request(delete, Path, _Params = []), + ct:pal("delete bridge (~s, http) result:\n ~p", [Kind, Res]), + simplify_result(Res). + op_bridge_api(Op, BridgeType, BridgeName) -> op_bridge_api(_Kind = action, Op, BridgeType, BridgeName). diff --git a/apps/emqx_bridge_azure_event_hub/rebar.config b/apps/emqx_bridge_azure_event_hub/rebar.config index a0cc8def3..63f944a5c 100644 --- a/apps/emqx_bridge_azure_event_hub/rebar.config +++ b/apps/emqx_bridge_azure_event_hub/rebar.config @@ -2,7 +2,7 @@ {erl_opts, [debug_info]}. {deps, [ - {wolff, {git, "https://github.com/kafka4beam/wolff.git", {tag, "1.10.5"}}}, + {wolff, {git, "https://github.com/kafka4beam/wolff.git", {tag, "1.10.6"}}}, {kafka_protocol, {git, "https://github.com/kafka4beam/kafka_protocol.git", {tag, "4.1.5"}}}, {brod_gssapi, {git, "https://github.com/kafka4beam/brod_gssapi.git", {tag, "v0.1.1"}}}, {brod, {git, "https://github.com/kafka4beam/brod.git", {tag, "3.18.0"}}}, diff --git a/apps/emqx_bridge_azure_event_hub/test/emqx_bridge_azure_event_hub_v2_SUITE.erl b/apps/emqx_bridge_azure_event_hub/test/emqx_bridge_azure_event_hub_v2_SUITE.erl index bc09a5509..661b8819c 100644 --- a/apps/emqx_bridge_azure_event_hub/test/emqx_bridge_azure_event_hub_v2_SUITE.erl +++ b/apps/emqx_bridge_azure_event_hub/test/emqx_bridge_azure_event_hub_v2_SUITE.erl @@ -40,6 +40,8 @@ init_per_suite(Config) -> emqx, emqx_management, emqx_resource, + %% Just for test helpers + brod, emqx_bridge_azure_event_hub, emqx_bridge, emqx_rule_engine, @@ -93,6 +95,9 @@ common_init_per_testcase(TestCase, Config) -> {connector_type, ?CONNECTOR_TYPE}, {connector_name, Name}, {connector_config, ConnectorConfig}, + {action_type, ?BRIDGE_TYPE}, + {action_name, Name}, + {action_config, BridgeConfig}, {bridge_type, ?BRIDGE_TYPE}, {bridge_name, Name}, {bridge_config, BridgeConfig} @@ -100,18 +105,13 @@ common_init_per_testcase(TestCase, Config) -> ]. end_per_testcase(_Testcase, Config) -> - case proplists:get_bool(skip_does_not_apply, Config) of - true -> - ok; - false -> - ProxyHost = ?config(proxy_host, Config), - ProxyPort = ?config(proxy_port, Config), - emqx_common_test_helpers:reset_proxy(ProxyHost, ProxyPort), - emqx_bridge_v2_testlib:delete_all_bridges_and_connectors(), - emqx_common_test_helpers:call_janitor(60_000), - ok = snabbkaffe:stop(), - ok - end. + ProxyHost = ?config(proxy_host, Config), + ProxyPort = ?config(proxy_port, Config), + emqx_common_test_helpers:reset_proxy(ProxyHost, ProxyPort), + emqx_bridge_v2_testlib:delete_all_bridges_and_connectors(), + emqx_common_test_helpers:call_janitor(60_000), + ok = snabbkaffe:stop(), + ok. %%------------------------------------------------------------------------------ %% Helper fns @@ -172,7 +172,7 @@ bridge_config(Name, ConnectorId, KafkaTopic) -> #{ <<"enable">> => true, <<"connector">> => ConnectorId, - <<"kafka">> => + <<"parameters">> => #{ <<"buffer">> => #{ @@ -322,7 +322,7 @@ t_same_name_azure_kafka_bridges(Config) -> ), %% then create a Kafka bridge with same name and delete it after creation - ConfigKafka0 = lists:keyreplace(bridge_type, 1, Config, {bridge_type, ?KAFKA_BRIDGE_TYPE}), + ConfigKafka0 = lists:keyreplace(action_type, 1, Config, {action_type, ?KAFKA_BRIDGE_TYPE}), ConfigKafka = lists:keyreplace( connector_type, 1, ConfigKafka0, {connector_type, ?KAFKA_BRIDGE_TYPE} ), @@ -374,3 +374,20 @@ t_http_api_get(Config) -> emqx_bridge_testlib:list_bridges_api() ), ok. + +t_multiple_actions_sharing_topic(Config) -> + ActionConfig0 = ?config(action_config, Config), + ActionConfig = + emqx_utils_maps:deep_merge( + ActionConfig0, + #{<<"parameters">> => #{<<"query_mode">> => <<"sync">>}} + ), + ok = emqx_bridge_v2_kafka_producer_SUITE:t_multiple_actions_sharing_topic( + [ + {type, ?BRIDGE_TYPE_BIN}, + {connector_name, ?config(connector_name, Config)}, + {connector_config, ?config(connector_config, Config)}, + {action_config, ActionConfig} + ] + ), + ok. diff --git a/apps/emqx_bridge_confluent/rebar.config b/apps/emqx_bridge_confluent/rebar.config index f518c8d4f..c6df829f9 100644 --- a/apps/emqx_bridge_confluent/rebar.config +++ b/apps/emqx_bridge_confluent/rebar.config @@ -2,7 +2,7 @@ {erl_opts, [debug_info]}. {deps, [ - {wolff, {git, "https://github.com/kafka4beam/wolff.git", {tag, "1.10.5"}}}, + {wolff, {git, "https://github.com/kafka4beam/wolff.git", {tag, "1.10.6"}}}, {kafka_protocol, {git, "https://github.com/kafka4beam/kafka_protocol.git", {tag, "4.1.5"}}}, {brod_gssapi, {git, "https://github.com/kafka4beam/brod_gssapi.git", {tag, "v0.1.1"}}}, {brod, {git, "https://github.com/kafka4beam/brod.git", {tag, "3.18.0"}}}, diff --git a/apps/emqx_bridge_confluent/test/emqx_bridge_confluent_producer_SUITE.erl b/apps/emqx_bridge_confluent/test/emqx_bridge_confluent_producer_SUITE.erl index 724365c58..0b3a22a99 100644 --- a/apps/emqx_bridge_confluent/test/emqx_bridge_confluent_producer_SUITE.erl +++ b/apps/emqx_bridge_confluent/test/emqx_bridge_confluent_producer_SUITE.erl @@ -40,6 +40,8 @@ init_per_suite(Config) -> emqx, emqx_management, emqx_resource, + %% Just for test helpers + brod, emqx_bridge_confluent, emqx_bridge, emqx_rule_engine, @@ -93,6 +95,9 @@ common_init_per_testcase(TestCase, Config) -> {connector_type, ?CONNECTOR_TYPE}, {connector_name, Name}, {connector_config, ConnectorConfig}, + {action_type, ?ACTION_TYPE}, + {action_name, Name}, + {action_config, BridgeConfig}, {bridge_type, ?ACTION_TYPE}, {bridge_name, Name}, {bridge_config, BridgeConfig} @@ -306,7 +311,7 @@ t_same_name_confluent_kafka_bridges(Config) -> ), %% then create a Kafka bridge with same name and delete it after creation - ConfigKafka0 = lists:keyreplace(bridge_type, 1, Config, {bridge_type, ?KAFKA_BRIDGE_TYPE}), + ConfigKafka0 = lists:keyreplace(action_type, 1, Config, {action_type, ?KAFKA_BRIDGE_TYPE}), ConfigKafka = lists:keyreplace( connector_type, 1, ConfigKafka0, {connector_type, ?KAFKA_BRIDGE_TYPE} ), @@ -378,3 +383,20 @@ t_list_v1_bridges(Config) -> [] ), ok. + +t_multiple_actions_sharing_topic(Config) -> + ActionConfig0 = ?config(action_config, Config), + ActionConfig = + emqx_utils_maps:deep_merge( + ActionConfig0, + #{<<"parameters">> => #{<<"query_mode">> => <<"sync">>}} + ), + ok = emqx_bridge_v2_kafka_producer_SUITE:t_multiple_actions_sharing_topic( + [ + {type, ?ACTION_TYPE_BIN}, + {connector_name, ?config(connector_name, Config)}, + {connector_config, ?config(connector_config, Config)}, + {action_config, ActionConfig} + ] + ), + ok. diff --git a/apps/emqx_bridge_kafka/rebar.config b/apps/emqx_bridge_kafka/rebar.config index 500c5a394..5e394a261 100644 --- a/apps/emqx_bridge_kafka/rebar.config +++ b/apps/emqx_bridge_kafka/rebar.config @@ -2,7 +2,7 @@ {erl_opts, [debug_info]}. {deps, [ - {wolff, {git, "https://github.com/kafka4beam/wolff.git", {tag, "1.10.5"}}}, + {wolff, {git, "https://github.com/kafka4beam/wolff.git", {tag, "1.10.6"}}}, {kafka_protocol, {git, "https://github.com/kafka4beam/kafka_protocol.git", {tag, "4.1.5"}}}, {brod_gssapi, {git, "https://github.com/kafka4beam/brod_gssapi.git", {tag, "v0.1.1"}}}, {brod, {git, "https://github.com/kafka4beam/brod.git", {tag, "3.18.0"}}}, 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 872e0027a..6b85cbbe4 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 @@ -324,6 +324,12 @@ on_query( }), do_send_msg(sync, KafkaMessage, Producers, SyncTimeout) catch + error:{invalid_partition_count, _Count, _Partitioner} -> + ?tp("kafka_producer_invalid_partition_count", #{ + action_id => MessageTag, + query_mode => sync + }), + {error, invalid_partition_count}; throw:{bad_kafka_header, _} = Error -> ?tp( emqx_bridge_kafka_impl_producer_sync_query_failed, @@ -385,6 +391,10 @@ on_query_async( do_send_msg(async, KafkaMessage, Producers, AsyncReplyFn) catch error:{invalid_partition_count, _Count, _Partitioner} -> + ?tp("kafka_producer_invalid_partition_count", #{ + action_id => MessageTag, + query_mode => async + }), {error, invalid_partition_count}; throw:{bad_kafka_header, _} = Error -> ?tp( @@ -690,6 +700,7 @@ producers_config(BridgeType, BridgeName, Input, IsDryRun, BridgeV2Id) -> max_batch_bytes => MaxBatchBytes, max_send_ahead => MaxInflight - 1, compression => Compression, + alias => BridgeV2Id, telemetry_meta_data => #{bridge_id => BridgeV2Id}, max_partitions => MaxPartitions }. diff --git a/apps/emqx_bridge_kafka/test/emqx_bridge_v2_kafka_producer_SUITE.erl b/apps/emqx_bridge_kafka/test/emqx_bridge_v2_kafka_producer_SUITE.erl index afd6e05a5..ba558792b 100644 --- a/apps/emqx_bridge_kafka/test/emqx_bridge_v2_kafka_producer_SUITE.erl +++ b/apps/emqx_bridge_kafka/test/emqx_bridge_v2_kafka_producer_SUITE.erl @@ -142,6 +142,9 @@ check_send_message_with_bridge(BridgeName) -> check_kafka_message_payload(Offset, Payload). send_message(ActionName) -> + send_message(?TYPE, ActionName). + +send_message(Type, ActionName) -> %% ###################################### %% Create Kafka message %% ###################################### @@ -157,8 +160,8 @@ send_message(ActionName) -> %% ###################################### %% Send message %% ###################################### - emqx_bridge_v2:send_message(?TYPE, ActionName, Msg, #{}), - #{offset => Offset, payload => Payload}. + Res = emqx_bridge_v2:send_message(Type, ActionName, Msg, #{}), + #{offset => Offset, payload => Payload, result => Res}. resolve_kafka_offset() -> KafkaTopic = emqx_bridge_kafka_impl_producer_SUITE:test_topic_one_partition(), @@ -285,6 +288,18 @@ action_api_spec_props_for_get() -> emqx_bridge_v2_testlib:actions_api_spec_schemas(), Props. +assert_status_api(Line, Type, Name, Status) -> + ?assertMatch( + {ok, + {{_, 200, _}, _, #{ + <<"status">> := Status, + <<"node_status">> := [#{<<"status">> := Status}] + }}}, + emqx_bridge_v2_testlib:get_bridge_api(Type, Name), + #{line => Line, name => Name, expected_status => Status} + ). +-define(assertStatusAPI(TYPE, NAME, STATUS), assert_status_api(?LINE, TYPE, NAME, STATUS)). + %%------------------------------------------------------------------------------ %% Testcases %%------------------------------------------------------------------------------ @@ -662,3 +677,77 @@ t_ancient_v1_config_migration_without_local_topic(Config) -> erpc:call(Node, fun emqx_bridge_v2:list/0) ), ok. + +%% Tests that deleting/disabling an action that share the same Kafka topic with other +%% actions do not disturb the latter. +t_multiple_actions_sharing_topic(Config) -> + Type = proplists:get_value(type, Config, ?TYPE), + ConnectorName = proplists:get_value(connector_name, Config, <<"c">>), + ConnectorConfig = proplists:get_value(connector_config, Config, connector_config()), + ActionConfig = proplists:get_value(action_config, Config, action_config(ConnectorName)), + ?check_trace( + begin + ConnectorParams = [ + {connector_config, ConnectorConfig}, + {connector_name, ConnectorName}, + {connector_type, Type} + ], + ActionName1 = <<"a1">>, + ActionParams1 = [ + {action_config, ActionConfig}, + {action_name, ActionName1}, + {action_type, Type} + ], + ActionName2 = <<"a2">>, + ActionParams2 = [ + {action_config, ActionConfig}, + {action_name, ActionName2}, + {action_type, Type} + ], + {ok, {{_, 201, _}, _, #{}}} = + emqx_bridge_v2_testlib:create_connector_api(ConnectorParams), + {ok, {{_, 201, _}, _, #{}}} = + emqx_bridge_v2_testlib:create_action_api(ActionParams1), + {ok, {{_, 201, _}, _, #{}}} = + emqx_bridge_v2_testlib:create_action_api(ActionParams2), + RuleTopic = <<"t/a2">>, + {ok, _} = emqx_bridge_v2_testlib:create_rule_and_action_http(Type, RuleTopic, Config), + + ?assertStatusAPI(Type, ActionName1, <<"connected">>), + ?assertStatusAPI(Type, ActionName2, <<"connected">>), + + %% Disabling a1 shouldn't disturb a2. + ?assertMatch( + {204, _}, emqx_bridge_v2_testlib:disable_kind_api(action, Type, ActionName1) + ), + + ?assertStatusAPI(Type, ActionName1, <<"disconnected">>), + ?assertStatusAPI(Type, ActionName2, <<"connected">>), + + ?assertMatch(#{result := ok}, send_message(Type, ActionName2)), + ?assertStatusAPI(Type, ActionName2, <<"connected">>), + + ?assertMatch( + {204, _}, + emqx_bridge_v2_testlib:enable_kind_api(action, Type, ActionName1) + ), + ?assertStatusAPI(Type, ActionName1, <<"connected">>), + ?assertStatusAPI(Type, ActionName2, <<"connected">>), + ?assertMatch(#{result := ok}, send_message(Type, ActionName2)), + + %% Deleting also shouldn't disrupt a2. + ?assertMatch( + {204, _}, + emqx_bridge_v2_testlib:delete_kind_api(action, Type, ActionName1) + ), + ?assertStatusAPI(Type, ActionName2, <<"connected">>), + ?assertMatch(#{result := ok}, send_message(Type, ActionName2)), + + ok + end, + fun(Trace) -> + ?assertEqual([], ?of_kind("kafka_producer_invalid_partition_count", Trace)), + ok + end + ), + ok. diff --git a/changes/ee/breaking-13327.en.md b/changes/ee/breaking-13327.en.md new file mode 100644 index 000000000..645ab0c73 --- /dev/null +++ b/changes/ee/breaking-13327.en.md @@ -0,0 +1,3 @@ +The directory path scheme for on-disk Kafka/Confluent/Azure Event Hub buffers has changed. It now uses the Action name instead of the topic name. + +Updating to this version will invalidate (not use) old paths, and will require manual cleanup of the old directories. diff --git a/changes/ee/fix-13327.en.md b/changes/ee/fix-13327.en.md new file mode 100644 index 000000000..e81490c5c --- /dev/null +++ b/changes/ee/fix-13327.en.md @@ -0,0 +1 @@ +Fixed an issue with Kafka, Confluent and Azure Event Hub bridges where different actions targeting the same topic could break one another when being deleted or disabled. diff --git a/mix.exs b/mix.exs index a94707da2..68a79c73e 100644 --- a/mix.exs +++ b/mix.exs @@ -211,7 +211,7 @@ defmodule EMQXUmbrella.MixProject do {:hstreamdb_erl, github: "hstreamdb/hstreamdb_erl", tag: "0.5.18+v0.18.1+ezstd-v1.0.5-emqx1"}, {:influxdb, github: "emqx/influxdb-client-erl", tag: "1.1.13", override: true}, - {:wolff, github: "kafka4beam/wolff", tag: "1.10.5"}, + {:wolff, github: "kafka4beam/wolff", tag: "1.10.6"}, {:kafka_protocol, github: "kafka4beam/kafka_protocol", tag: "4.1.5", override: true}, {:brod_gssapi, github: "kafka4beam/brod_gssapi", tag: "v0.1.1"}, {:brod, github: "kafka4beam/brod", tag: "3.18.0"},