yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(emqx_bridge_kafka): match example in api schema

This commit is contained in:
Stefan Strigler 2023-11-16 09:41:56 +01:00
parent e8d18b0e09
commit d2901afd1b
1 changed files with 43 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

85
apps/emqx_bridge_kafka/src/emqx_bridge_kafka.erl
View File

@ -112,16 +112,15 @@ values({put, connector}) ->
values({put, KafkaType}) -> values({put, KafkaType}) ->
maps:merge(values(common_config), values(KafkaType)); maps:merge(values(common_config), values(KafkaType));
values(bridge_v2_producer) -> values(bridge_v2_producer) ->
maps:merge( #{
#{ enable => true,
enable => true, connector => <<"my_kafka_producer_connector">>,
connector => <<"my_kafka_producer_connector">>, parameters => values(producer_values),
resource_opts => #{ local_topic => <<"mqtt/local/topic">>,
health_check_interval => "32s" resource_opts => #{
} health_check_interval => "32s"
}, }
values(producer) };
);
values(common_config) -> values(common_config) ->
#{ #{
authentication => #{ authentication => #{
@ -143,40 +142,42 @@ values(common_config) ->
}; };
values(producer) -> values(producer) ->
#{ #{
kafka => #{ kafka => values(producer_values),
topic => <<"kafka-topic">>,
message => #{
key => <<"${.clientid}">>,
value => <<"${.}">>,
timestamp => <<"${.timestamp}">>
},
max_batch_bytes => <<"896KB">>,
compression => <<"no_compression">>,
partition_strategy => <<"random">>,
required_acks => <<"all_isr">>,
partition_count_refresh_interval => <<"60s">>,
kafka_headers => <<"${pub_props}">>,
kafka_ext_headers => [
#{
kafka_ext_header_key => <<"clientid">>,
kafka_ext_header_value => <<"${clientid}">>
},
#{
kafka_ext_header_key => <<"topic">>,
kafka_ext_header_value => <<"${topic}">>
}
],
kafka_header_value_encode_mode => none,
max_inflight => 10,
buffer => #{
mode => <<"hybrid">>,
per_partition_limit => <<"2GB">>,
segment_bytes => <<"100MB">>,
memory_overload_protection => true
}
},
local_topic => <<"mqtt/local/topic">> local_topic => <<"mqtt/local/topic">>
}; };
values(producer_values) ->
#{
topic => <<"kafka-topic">>,
message => #{
key => <<"${.clientid}">>,
value => <<"${.}">>,
timestamp => <<"${.timestamp}">>
},
max_batch_bytes => <<"896KB">>,
compression => <<"no_compression">>,
partition_strategy => <<"random">>,
required_acks => <<"all_isr">>,
partition_count_refresh_interval => <<"60s">>,
kafka_headers => <<"${pub_props}">>,
kafka_ext_headers => [
#{
kafka_ext_header_key => <<"clientid">>,
kafka_ext_header_value => <<"${clientid}">>
},
#{
kafka_ext_header_key => <<"topic">>,
kafka_ext_header_value => <<"${topic}">>
}
],
kafka_header_value_encode_mode => none,
max_inflight => 10,
buffer => #{
mode => <<"hybrid">>,
per_partition_limit => <<"2GB">>,
segment_bytes => <<"100MB">>,
memory_overload_protection => true
}
};
values(consumer) -> values(consumer) ->
#{ #{
kafka => #{ kafka => #{