feat: add connector schema scaffold and break out Kafka conector

.ci/docker-compose-file/docker-compose-kafka.yaml

@@ -18,7 +18,7 @@ services:
       - /tmp/emqx-ci/emqx-shared-secret:/var/lib/secret
   kdc:
     hostname: kdc.emqx.net
-    image:  ghcr.io/emqx/emqx-builder/5.1-4:1.14.5-25.3.2-2-ubuntu20.04
+    image:  ghcr.io/emqx/emqx-builder/5.2-3:1.14.5-25.3.2-2-ubuntu22.04
     container_name: kdc.emqx.net
     expose:
       - 88 # kdc

.ci/docker-compose-file/docker-compose.yaml

@@ -3,17 +3,17 @@ version: '3.9'
 services:
   erlang:
     container_name: erlang
-    image: ${DOCKER_CT_RUNNER_IMAGE:-ghcr.io/emqx/emqx-builder/5.1-4:1.14.5-25.3.2-2-ubuntu20.04}
+    image: ${DOCKER_CT_RUNNER_IMAGE:-ghcr.io/emqx/emqx-builder/5.2-3:1.14.5-25.3.2-2-ubuntu22.04}
     env_file:
       - conf.env
     environment:
-      GITHUB_ACTIONS: ${GITHUB_ACTIONS}
-      GITHUB_TOKEN: ${GITHUB_TOKEN}
-      GITHUB_RUN_ID: ${GITHUB_RUN_ID}
-      GITHUB_SHA: ${GITHUB_SHA}
-      GITHUB_RUN_NUMBER: ${GITHUB_RUN_NUMBER}
-      GITHUB_EVENT_NAME: ${GITHUB_EVENT_NAME}
-      GITHUB_REF: ${GITHUB_REF}
+      GITHUB_ACTIONS: ${GITHUB_ACTIONS:-}
+      GITHUB_TOKEN: ${GITHUB_TOKEN:-}
+      GITHUB_RUN_ID: ${GITHUB_RUN_ID:-}
+      GITHUB_SHA: ${GITHUB_SHA:-}
+      GITHUB_RUN_NUMBER: ${GITHUB_RUN_NUMBER:-}
+      GITHUB_EVENT_NAME: ${GITHUB_EVENT_NAME:-}
+      GITHUB_REF: ${GITHUB_REF:-}
     networks:
       - emqx_bridge
     ports:

.github/workflows/_pr_entrypoint.yaml

@@ -25,6 +25,7 @@ jobs:
       version-emqx: ${{ steps.matrix.outputs.version-emqx }}
       version-emqx-enterprise: ${{ steps.matrix.outputs.version-emqx-enterprise }}
       runner_labels: ${{ github.repository_owner == 'emqx' && '["self-hosted","ephemeral","linux","x64"]' || '["ubuntu-22.04"]' }}
+      xl_runner_labels: ${{ github.repository_owner == 'emqx' && '["self-hosted","ephemeral-xl","linux","x64"]' || '["ubuntu-22.04"]' }}
       builder: "ghcr.io/emqx/emqx-builder/5.2-3:1.14.5-25.3.2-2-ubuntu22.04"
       builder_vsn: "5.2-3"
       otp_vsn: "25.3.2-2"
@@ -115,7 +116,7 @@ jobs:
           echo "version-emqx-enterprise=$(./pkg-vsn.sh emqx-enterprise)" | tee -a $GITHUB_OUTPUT
 
   compile:
-    runs-on: ${{ fromJSON(needs.sanity-checks.outputs.runner_labels) }}
+    runs-on: ${{ fromJSON(needs.sanity-checks.outputs.xl_runner_labels) }}
     container: ${{ needs.sanity-checks.outputs.builder }}
     needs:
       - sanity-checks
@@ -153,7 +154,7 @@ jobs:
       - compile
     uses: ./.github/workflows/run_emqx_app_tests.yaml
     with:
-      runner_labels: ${{ needs.sanity-checks.outputs.runner_labels }}
+      runner_labels: ${{ needs.sanity-checks.outputs.xl_runner_labels }}
       builder: ${{ needs.sanity-checks.outputs.builder }}
       before_ref: ${{ github.event_name == 'pull_request' && github.event.pull_request.base.sha || github.event.before }}
       after_ref: ${{ github.sha }}

.github/workflows/run_helm_tests.yaml

@@ -140,12 +140,13 @@ jobs:
           echo "waiting emqx started";
           sleep 10;
         done
-    - name: Get Token
+    - name: Setup 18083 port forwarding
+      run: |
+        nohup kubectl port-forward service/${EMQX_NAME} 18083:18083 > /dev/null &
+    - name: Get auth token
       run: |
-        kubectl port-forward service/${EMQX_NAME} 18083:18083 > /dev/null &
         curl --head -X GET --retry 10 --retry-connrefused --retry-delay 6 http://localhost:18083/status
         echo "TOKEN=$(curl --silent -X 'POST' 'http://127.0.0.1:18083/api/v5/login' -H 'accept: application/json' -H 'Content-Type: application/json' -d '{"username": "admin","password": "public"}' | jq -r ".token")" >> $GITHUB_ENV
-
     - name: Check cluster
       timeout-minutes: 1
       run: |

.github/workflows/run_test_cases.yaml

@@ -69,7 +69,7 @@ jobs:
 
   ct_docker:
     runs-on: ${{ fromJSON(inputs.runner_labels) }}
-    name: "ct_docker (${{ matrix.app }}-${{ matrix.suitegroup }})"
+    name: "${{ matrix.app }}-${{ matrix.suitegroup }} (${{ matrix.profile }})"
     strategy:
       fail-fast: false
       matrix:
@@ -119,7 +119,7 @@ jobs:
 
   ct:
     runs-on: ${{ fromJSON(inputs.runner_labels) }}
-    name: "ct (${{ matrix.app }}-${{ matrix.suitegroup }})"
+    name: "${{ matrix.app }}-${{ matrix.suitegroup }} (${{ matrix.profile }})"
     strategy:
       fail-fast: false
       matrix:

Makefile

@@ -111,6 +111,11 @@ ifneq ($(CASES),)
 CASES_ARG := --case $(CASES)
 endif
 
+# Allow user-set GROUPS environment variable
+ifneq ($(GROUPS),)
+GROUPS_ARG := --groups $(GROUPS)
+endif
+
 ## example:
 ## env SUITES=apps/appname/test/test_SUITE.erl CASES=t_foo make apps/appname-ct
 define gen-app-ct-target
@@ -122,6 +127,7 @@ ifneq ($(SUITES),)
 		--name $(CT_NODE_NAME) \
 		--cover_export_name $(CT_COVER_EXPORT_PREFIX)-$(subst /,-,$1) \
 		--suite $(SUITES) \
+		$(GROUPS_ARG) \
 		$(CASES_ARG)
 else
 	@echo 'No suites found for $1'

apps/emqx/include/emqx_release.hrl

@@ -35,7 +35,7 @@
 -define(EMQX_RELEASE_CE, "5.3.1-alpha.1").
 
 %% Enterprise edition
--define(EMQX_RELEASE_EE, "5.3.1-alpha.1").
+-define(EMQX_RELEASE_EE, "5.3.1-alpha.2").
 
 %% The HTTP API version
 -define(EMQX_API_VERSION, "5.0").

apps/emqx/priv/bpapi.versions

@@ -7,12 +7,14 @@
 {emqx_bridge,2}.
 {emqx_bridge,3}.
 {emqx_bridge,4}.
+{emqx_bridge,5}.
 {emqx_broker,1}.
 {emqx_cm,1}.
 {emqx_cm,2}.
 {emqx_conf,1}.
 {emqx_conf,2}.
 {emqx_conf,3}.
+{emqx_connector, 1}.
 {emqx_dashboard,1}.
 {emqx_delayed,1}.
 {emqx_delayed,2}.

apps/emqx/src/emqx_config.erl

@@ -325,22 +325,32 @@ init_load(SchemaMod, Conf) when is_list(Conf) orelse is_binary(Conf) ->
     ok = save_schema_mod_and_names(SchemaMod),
     HasDeprecatedFile = has_deprecated_file(),
     RawConf0 = load_config_files(HasDeprecatedFile, Conf),
-    warning_deprecated_root_key(RawConf0),
-    RawConf1 =
+    RawConf1 = upgrade_raw_conf(SchemaMod, RawConf0),
+    warning_deprecated_root_key(RawConf1),
+    RawConf2 =
         case HasDeprecatedFile of
             true ->
-                overlay_v0(SchemaMod, RawConf0);
+                overlay_v0(SchemaMod, RawConf1);
             false ->
-                overlay_v1(SchemaMod, RawConf0)
+                overlay_v1(SchemaMod, RawConf1)
         end,
-    RawConf = fill_defaults_for_all_roots(SchemaMod, RawConf1),
+    RawConf3 = fill_defaults_for_all_roots(SchemaMod, RawConf2),
     %% check configs against the schema
-    {AppEnvs, CheckedConf} = check_config(SchemaMod, RawConf, #{}),
+    {AppEnvs, CheckedConf} = check_config(SchemaMod, RawConf3, #{}),
     save_to_app_env(AppEnvs),
-    ok = save_to_config_map(CheckedConf, RawConf),
+    ok = save_to_config_map(CheckedConf, RawConf3),
     maybe_init_default_zone(),
     ok.
 
+upgrade_raw_conf(SchemaMod, RawConf) ->
+    case erlang:function_exported(SchemaMod, upgrade_raw_conf, 1) of
+        true ->
+            %% TODO make it a schema module behaviour in hocon_schema
+            apply(SchemaMod, upgrade_raw_conf, [RawConf]);
+        false ->
+            RawConf
+    end.
+
 %% Merge environment variable overrides on top, then merge with overrides.
 overlay_v0(SchemaMod, RawConf) when is_map(RawConf) ->
     RawConfWithEnvs = merge_envs(SchemaMod, RawConf),

apps/emqx/src/emqx_config_handler.erl

@@ -19,7 +19,7 @@
 
 -include("logger.hrl").
 -include("emqx_schema.hrl").
--include_lib("hocon/include/hoconsc.hrl").
+-include_lib("hocon/include/hocon_types.hrl").
 
 -behaviour(gen_server).
 
@@ -736,7 +736,7 @@ remove_empty_leaf(KeyPath, Handlers) ->
     end.
 
 assert_callback_function(Mod) ->
-    _ = Mod:module_info(),
+    _ = apply(Mod, module_info, []),
     case
         erlang:function_exported(Mod, pre_config_update, 3) orelse
             erlang:function_exported(Mod, post_config_update, 5)

apps/emqx/test/emqx_common_test_helpers.erl

@@ -22,6 +22,8 @@
 
 -export([
     all/1,
+    matrix_to_groups/2,
+    group_path/1,
     init_per_testcase/3,
     end_per_testcase/3,
     boot_modules/1,
@@ -1375,3 +1377,83 @@ select_free_port(GenModule, Fun) when
     end,
     ct:pal("Select free OS port: ~p", [Port]),
     Port.
+
+%% Generate ct sub-groups from test-case's 'matrix' clause
+%% NOTE: the test cases must have a root group name which
+%% is unkonwn to this API.
+%%
+%% e.g.
+%% all() -> [{group, g1}].
+%%
+%% groups() ->
+%%   emqx_common_test_helpers:groups(?MODULE, [case1, case2]).
+%%
+%% case1(matrxi) ->
+%%   {g1, [[tcp, no_auth],
+%%         [ssl, no_auth],
+%%         [ssl, basic_auth]
+%%        ]};
+%%
+%% case2(matrxi) ->
+%%   {g1, ...}
+%% ...
+%%
+%% Return:
+%%
+%%  [{g1, [],
+%%     [ {tcp, [], [{no_auth,    [], [case1, case2]}
+%%                 ]},
+%%       {ssl, [], [{no_auth,    [], [case1, case2]},
+%%                  {basic_auth, [], [case1, case2]}
+%%                 ]}
+%%     ]
+%%   }
+%%  ]
+matrix_to_groups(Module, Cases) ->
+    lists:foldr(
+        fun(Case, Acc) ->
+            add_case_matrix(Module, Case, Acc)
+        end,
+        [],
+        Cases
+    ).
+
+add_case_matrix(Module, Case, Acc0) ->
+    {RootGroup, Matrix} = Module:Case(matrix),
+    lists:foldr(
+        fun(Row, Acc) ->
+            add_group([RootGroup | Row], Acc, Case)
+        end,
+        Acc0,
+        Matrix
+    ).
+
+add_group([], Acc, Case) ->
+    case lists:member(Case, Acc) of
+        true ->
+            Acc;
+        false ->
+            [Case | Acc]
+    end;
+add_group([Name | More], Acc, Cases) ->
+    case lists:keyfind(Name, 1, Acc) of
+        false ->
+            [{Name, [], add_group(More, [], Cases)} | Acc];
+        {Name, [], SubGroup} ->
+            New = {Name, [], add_group(More, SubGroup, Cases)},
+            lists:keystore(Name, 1, Acc, New)
+    end.
+
+group_path(Config) ->
+    try
+        Current = proplists:get_value(tc_group_properties, Config),
+        NameF = fun(Props) ->
+            {name, Name} = lists:keyfind(name, 1, Props),
+            Name
+        end,
+        Stack = proplists:get_value(tc_group_path, Config),
+        lists:reverse(lists:map(NameF, [Current | Stack]))
+    catch
+        _:_ ->
+            []
+    end.

apps/emqx/test/emqx_common_test_http.erl

@@ -61,7 +61,7 @@ request_api(Method, Url, QueryParams, Auth, Body, HttpOpts) ->
     do_request_api(Method, Request, HttpOpts).
 
 do_request_api(Method, Request, HttpOpts) ->
-    ct:pal("Method: ~p, Request: ~p", [Method, Request]),
+    % ct:pal("Method: ~p, Request: ~p", [Method, Request]),
     case httpc:request(Method, Request, HttpOpts, [{body_format, binary}]) of
         {error, socket_closed_remotely} ->
             {error, socket_closed_remotely};

apps/emqx_bridge/src/emqx_bridge.app.src

@@ -1,7 +1,7 @@
 %% -*- mode: erlang -*-
 {application, emqx_bridge, [
     {description, "EMQX bridges"},
-    {vsn, "0.1.28"},
+    {vsn, "0.1.29"},
     {registered, [emqx_bridge_sup]},
     {mod, {emqx_bridge_app, []}},
     {applications, [