diff --git a/include/fluent-bit/flb_input.h b/include/fluent-bit/flb_input.h
index 0b7be267c20..d2e30314e31 100644
--- a/include/fluent-bit/flb_input.h
+++ b/include/fluent-bit/flb_input.h
@@ -75,6 +75,71 @@
 
 struct flb_input_instance;
 
+/*
+ * Tests callbacks
+ * ===============
+ */
+struct flb_test_in_formatter {
+    /*
+     * Runtime Library Mode
+     * ====================
+     * When the runtime library enable the test formatter mode, it needs to
+     * keep a reference of the context and other information:
+     *
+     * - rt_ctx : context created by flb_create()
+     *
+     * - rt_ffd : this plugin assigned 'integer' created by flb_output()
+     *
+     * - rt_in_calback: intermediary function to receive the results of
+     *                  the formatter plugin test function.
+     *
+     * - rt_data: opaque data type for rt_step_callback()
+     */
+
+    /* runtime library context */
+    void *rt_ctx;
+
+    /* runtime library: assigned plugin integer */
+    int rt_ffd;
+
+    /* optional format context */
+    void *format_ctx;
+
+    /*
+     * "runtime step callback": this function pointer is used by Fluent Bit
+     * library mode to reference a test function that must retrieve the
+     * results of 'callback'. Consider this an intermediary function to
+     * transfer the results to the runtime test.
+     *
+     * This function is private and should not be set manually in the plugin
+     * code, it's set on src/flb_lib.c .
+     */
+    void (*rt_in_callback) (void *, int, int, void *, size_t, void *);
+
+    /*
+     * opaque data type passed by the runtime library to be used on
+     * rt_step_test().
+     */
+    void *rt_data;
+
+    /*
+     * Callback
+     * =========
+     * "Formatter callback": it references the plugin function that performs
+     * data formatting (msgpack -> local data). This entry is mostly to
+     * expose the plugin local function.
+     */
+    int (*callback) (/* Fluent Bit context */
+                     struct flb_config *,
+                     /* plugin that ingested the records */
+                     struct flb_input_instance *,
+                     void *,         /* plugin instance context */
+                     const void *,   /* incoming unformatted data */
+                     size_t,         /* incoming unformatted size */
+                     void **,        /* output buffer      */
+                     size_t *);      /* output buffer size */
+};
+
 struct flb_input_plugin {
     /*
      * The type defines if this is a core-based plugin or it's handled by
@@ -138,6 +203,9 @@ struct flb_input_plugin {
     /* Destroy */
     void (*cb_destroy) (struct flb_input_plugin *);
 
+    /* Tests */
+    struct flb_test_in_formatter test_formatter;
+
     void *instance;
 
     struct mk_list _head;
@@ -177,6 +245,7 @@ struct flb_input_instance {
     int runs_in_coroutine;               /* instance runs in coroutine ? */
     char name[32];                       /* numbered name (cpu -> cpu.0) */
     char *alias;                         /* alias name for the instance  */
+    int test_mode;                       /* running tests? (default:off) */
     void *context;                       /* plugin configuration context */
     flb_pipefd_t ch_events[2];           /* channel for events           */
     struct flb_input_plugin *p;          /* original plugin              */
@@ -294,6 +363,8 @@ struct flb_input_instance {
     struct flb_metrics *metrics;         /* metrics                    */
 #endif
 
+    /* Tests */
+    struct flb_test_in_formatter test_formatter;
 
     /* is the plugin running in a separate thread ? */
     int is_threaded;
diff --git a/include/fluent-bit/flb_lib.h b/include/fluent-bit/flb_lib.h
index 897ea0098ee..88ff8cb0af3 100644
--- a/include/fluent-bit/flb_lib.h
+++ b/include/fluent-bit/flb_lib.h
@@ -57,6 +57,10 @@ FLB_EXPORT int flb_output(flb_ctx_t *ctx, const char *output, struct flb_lib_out
 FLB_EXPORT int flb_output_set_processor(flb_ctx_t *ctx, int ffd, struct flb_processor *proc);
 FLB_EXPORT int flb_filter(flb_ctx_t *ctx, const char *filter, void *data);
 FLB_EXPORT int flb_input_set(flb_ctx_t *ctx, int ffd, ...);
+FLB_EXPORT int flb_input_set_test(flb_ctx_t *ctx, int ffd, char *test_name,
+                                  void (*in_callback) (void *, int, int,
+                                                       void *, size_t, void *),
+                                  void *in_callback_data);
 FLB_EXPORT int flb_input_property_check(flb_ctx_t *ctx, int ffd, char *key, char *val);
 FLB_EXPORT int flb_output_property_check(flb_ctx_t *ctx, int ffd, char *key, char *val);
 FLB_EXPORT int flb_filter_property_check(flb_ctx_t *ctx, int ffd, char *key, char *val);
diff --git a/plugins/in_systemd/systemd.c b/plugins/in_systemd/systemd.c
index 85f36ee990a..08ba75d210e 100644
--- a/plugins/in_systemd/systemd.c
+++ b/plugins/in_systemd/systemd.c
@@ -70,13 +70,202 @@ static int tag_compose(const char *tag, const char *unit_name,
     return 0;
 }
 
+static int append_enumerate_data(struct flb_systemd_config *ctx, struct cfl_kvlist *kvlist)
+{
+    int i;
+    int ret = FLB_EVENT_ENCODER_SUCCESS;
+    struct cfl_list *head;
+    struct cfl_kvpair *kvpair = NULL;
+    struct cfl_variant *v = NULL;
+    struct cfl_array *array = NULL;
+
+    /* Interpret cfl_kvlist as logs type of events later. */
+    cfl_list_foreach(head, &kvlist->list) {
+        kvpair = cfl_list_entry(head, struct cfl_kvpair, _head);
+        if (ret == FLB_EVENT_ENCODER_SUCCESS) {
+            ret = flb_log_event_encoder_append_body_string_length(
+                    ctx->log_encoder, cfl_sds_len(kvpair->key));
+        }
+
+        if (ret == FLB_EVENT_ENCODER_SUCCESS) {
+            ret = flb_log_event_encoder_append_body_string_body(
+                    ctx->log_encoder, kvpair->key, cfl_sds_len(kvpair->key));
+        }
+
+        v = kvpair->val;
+        if (v->type == CFL_VARIANT_STRING) {
+            if (ret == FLB_EVENT_ENCODER_SUCCESS) {
+                ret = flb_log_event_encoder_append_body_string(
+                        ctx->log_encoder, v->data.as_string, cfl_variant_size_get(v));
+            }
+        }
+        else if (v->type == CFL_VARIANT_ARRAY) {
+            if (ret == FLB_EVENT_ENCODER_SUCCESS) {
+                ret = flb_log_event_encoder_body_begin_array(ctx->log_encoder);
+            }
+
+            array = v->data.as_array;
+            for (i = 0; i < array->entry_count; i++) {
+                if (ret == FLB_EVENT_ENCODER_SUCCESS) {
+                    if (array->entries[i]->type != CFL_VARIANT_STRING) {
+                        continue;
+                    }
+                    ret = flb_log_event_encoder_append_body_string(
+                            ctx->log_encoder, array->entries[i]->data.as_string,
+                            cfl_variant_size_get(array->entries[i]));
+                }
+            }
+
+            if (ret == FLB_EVENT_ENCODER_SUCCESS) {
+                ret = flb_log_event_encoder_body_commit_array(ctx->log_encoder);
+            }
+        }
+    }
+
+    return ret;
+}
+
+static int systemd_enumerate_data_store(struct flb_config *config,
+                                        struct flb_input_instance *ins,
+                                        void *plugin_context,
+                                        void *format_context,
+                                        const void *data, size_t data_size)
+{
+    int i;
+    int len;
+    int key_len;
+    size_t length = data_size;
+    const char *sep;
+    const char *key;
+    const char *val;
+    char *buf = NULL;
+    struct cfl_kvlist *kvlist = format_context;
+    struct flb_systemd_config *ctx = plugin_context;
+    struct cfl_variant *cfl_val = NULL;
+    struct cfl_array *array = NULL;
+    struct cfl_variant *tmp_val = NULL;
+    flb_sds_t list_key = NULL;
+    flb_sds_t search_key = NULL;
+
+    key = (const char *) data;
+    sep = strchr(key, '=');
+    if (sep == NULL) {
+        return -2;
+    }
+
+    len = (sep - key);
+    key_len = len;
+
+    if (ctx->lowercase == FLB_TRUE) {
+        /*
+         * Ensure buf to have enough space for the key because the libsystemd
+         * might return larger data than the threshold.
+         */
+        if (buf == NULL) {
+            buf = flb_sds_create_len(NULL, ctx->threshold);
+        }
+        if (flb_sds_alloc(buf) < len) {
+            buf = flb_sds_increase(buf, len - flb_sds_alloc(buf));
+        }
+        for (i = 0; i < len; i++) {
+            buf[i] = tolower(key[i]);
+        }
+        list_key = flb_sds_create_len(buf, key_len);
+    }
+    else {
+        list_key = flb_sds_create_len(key, key_len);
+    }
+
+    if (!list_key) {
+        return -1;
+    }
+
+    /* Check existence */
+    cfl_val = NULL;
+    cfl_val = cfl_kvlist_fetch_s(kvlist, list_key, key_len);
+
+    val = sep + 1;
+    len = length - (sep - key) - 1;
+
+    /* Initialize variable for cfl_variant operations. */
+    search_key = NULL;
+    tmp_val = NULL;
+
+    /* Store cfl_kvlist format at first to detect duplicated keys */
+    if (cfl_val) {
+        switch(cfl_val->type) {
+        case CFL_VARIANT_STRING:
+            tmp_val = cfl_variant_create_from_string(cfl_val->data.as_string);
+            if (!tmp_val) {
+                return -1;
+            }
+            break;
+        case CFL_VARIANT_ARRAY:
+            /* Just a reference */
+            tmp_val = cfl_val;
+            break;
+        default:
+            /* nop */
+            break;
+        }
+
+        switch(tmp_val->type) {
+        case CFL_VARIANT_STRING:
+            search_key = flb_sds_create_len(list_key, key_len);
+            if (search_key != NULL) {
+                cfl_kvlist_remove(kvlist, search_key);
+            }
+            flb_sds_destroy(search_key);
+
+            array = cfl_array_create(8);
+            if (!array) {
+                cfl_variant_destroy(tmp_val);
+                goto error;
+            }
+            if (cfl_array_resizable(array, CFL_TRUE) == -1) {
+                cfl_array_destroy(array);
+                cfl_variant_destroy(tmp_val);
+                goto error;
+            }
+
+            cfl_array_append_string_s(array,
+                                      tmp_val->data.as_string,
+                                      strlen(tmp_val->data.as_string),
+                                      CFL_FALSE);
+            cfl_array_append_string_s(array, (char *)val, strlen(val), CFL_FALSE);
+            cfl_kvlist_insert_array_s(kvlist, list_key, key_len, array);
+            cfl_variant_destroy(tmp_val);
+            break;
+        case CFL_VARIANT_ARRAY:
+            /* Just appending the newly arrived field(s) */
+            array = tmp_val->data.as_array;
+            cfl_array_append_string_s(array, (char *)val, strlen(val), CFL_FALSE);
+            break;
+        default:
+            /* nop */
+            break;
+        }
+    }
+    else {
+        cfl_kvlist_insert_string_s(kvlist, list_key, key_len,
+                                   (char *)val, strlen(val), CFL_FALSE);
+    }
+
+    flb_sds_destroy(list_key);
+
+    return 0;
+
+error:
+    flb_sds_destroy(list_key);
+
+    return -1;
+}
+
 static int in_systemd_collect(struct flb_input_instance *ins,
                               struct flb_config *config, void *in_context)
 {
     int ret;
     int ret_j;
-    int i;
-    int len;
     int entries = 0;
     int skip_entries = 0;
     int rows = 0;
@@ -84,10 +273,7 @@ static int in_systemd_collect(struct flb_input_instance *ins,
     long nsec;
     uint64_t usec;
     size_t length;
-    size_t threshold;
-    const char *sep;
     const char *key;
-    const char *val;
     char *buf = NULL;
 #ifdef FLB_HAVE_SQLDB
     char *cursor = NULL;
@@ -100,6 +286,7 @@ static int in_systemd_collect(struct flb_input_instance *ins,
     const void *data;
     struct flb_systemd_config *ctx = in_context;
     struct flb_time tm;
+    struct cfl_kvlist *kvlist = NULL;
 
     /* Restricted by mem_buf_limit */
     if (flb_input_buf_paused(ins) == FLB_TRUE) {
@@ -123,7 +310,7 @@ static int in_systemd_collect(struct flb_input_instance *ins,
     }
 
     if (ctx->lowercase == FLB_TRUE) {
-        ret = sd_journal_get_data_threshold(ctx->j, &threshold);
+        ret = sd_journal_get_data_threshold(ctx->j, &ctx->threshold);
         if (ret != 0) {
             flb_plg_error(ctx->ins,
                           "error setting up systemd data. "
@@ -200,6 +387,13 @@ static int in_systemd_collect(struct flb_input_instance *ins,
             ret = flb_log_event_encoder_set_timestamp(ctx->log_encoder, &tm);
         }
 
+        /* create an empty kvlist as the labels */
+        kvlist = cfl_kvlist_create();
+        if (!kvlist) {
+            flb_plg_error(ctx->ins, "error allocating kvlist");
+            break;
+        }
+
         /* Pack every field in the entry */
         entries = 0;
         skip_entries = 0;
@@ -211,58 +405,28 @@ static int in_systemd_collect(struct flb_input_instance *ins,
                 length--;
             }
 
-            sep = strchr(key, '=');
-            if (sep == NULL) {
+            ret = systemd_enumerate_data_store(config, ctx->ins,
+                                               (void *)ctx, (void *)kvlist,
+                                               key, length);
+            if (ret == -2) {
                 skip_entries++;
                 continue;
             }
-
-            len = (sep - key);
-
-            if (ret == FLB_EVENT_ENCODER_SUCCESS) {
-                ret = flb_log_event_encoder_append_body_string_length(
-                        ctx->log_encoder, len);
-            }
-
-            if (ctx->lowercase == FLB_TRUE) {
-                /*
-                 * Ensure buf to have enough space for the key because the libsystemd
-                 * might return larger data than the threshold.
-                 */
-                if (buf == NULL) {
-                    buf = flb_sds_create_len(NULL, threshold);
-                }
-                if (flb_sds_alloc(buf) < len) {
-                    buf = flb_sds_increase(buf, len - flb_sds_alloc(buf));
-                }
-                for (i = 0; i < len; i++) {
-                    buf[i] = tolower(key[i]);
-                }
-
-                if (ret == FLB_EVENT_ENCODER_SUCCESS) {
-                    ret = flb_log_event_encoder_append_body_string_body(
-                            ctx->log_encoder, buf, len);
-                }
-            }
-            else {
-                if (ret == FLB_EVENT_ENCODER_SUCCESS) {
-                    ret = flb_log_event_encoder_append_body_string_body(
-                            ctx->log_encoder, (char *) key, len);
-                }
-            }
-
-            val = sep + 1;
-            len = length - (sep - key) - 1;
-
-            if (ret == FLB_EVENT_ENCODER_SUCCESS) {
-                ret = flb_log_event_encoder_append_body_string(
-                        ctx->log_encoder, (char *) val, len);
+            else if (ret == -1) {
+                continue;
             }
 
             entries++;
         }
         rows++;
 
+        /* Interpret cfl_kvlist as logs type of events later. */
+        ret = append_enumerate_data(ctx, kvlist);
+
+        if (kvlist) {
+            cfl_kvlist_destroy(kvlist);
+        }
+
         if (skip_entries > 0) {
             flb_plg_error(ctx->ins, "Skip %d broken entries", skip_entries);
         }
@@ -483,6 +647,78 @@ static int in_systemd_exit(void *data, struct flb_config *config)
     return 0;
 }
 
+static int cb_systemd_format_test(struct flb_config *config,
+                                  struct flb_input_instance *ins,
+                                  void *plugin_context,
+                                  const void *data, size_t bytes,
+                                  void **out_data, size_t *out_size)
+{
+    int ret;
+    struct flb_systemd_config *ctx = plugin_context;
+    struct flb_time tm;
+    struct cfl_list *head = NULL;
+    struct cfl_list *kvs = NULL;
+    struct cfl_split_entry *cur = NULL;
+    struct cfl_kvlist *kvlist = NULL;
+    const char *keys;
+
+    ret = flb_log_event_encoder_begin_record(ctx->log_encoder);
+
+    if (ret == FLB_EVENT_ENCODER_SUCCESS) {
+        ret = flb_log_event_encoder_set_timestamp(ctx->log_encoder, &tm);
+    }
+
+    /* create an empty kvlist as the labels */
+    kvlist = cfl_kvlist_create();
+    if (!kvlist) {
+        flb_plg_error(ctx->ins, "error allocating kvlist");
+        return -1;
+    }
+
+    keys = (const char *) data;
+    kvs = cfl_utils_split(keys, '\n', -1 );
+    if (kvs == NULL) {
+        goto split_error;
+    }
+
+    cfl_list_foreach(head, kvs) {
+        cur = cfl_list_entry(head, struct cfl_split_entry, _head);
+        ret = systemd_enumerate_data_store(config, ctx->ins,
+                                           (void *)ctx, (void *)kvlist,
+                                           cur->value, cur->len);
+
+        if (ret == -2 || ret == -1) {
+            continue;
+        }
+    }
+
+    /* Interpret cfl_kvlist as logs type of events later. */
+    ret = append_enumerate_data(ctx, kvlist);
+
+    if (kvlist) {
+        cfl_kvlist_destroy(kvlist);
+    }
+
+    if (kvs != NULL) {
+        cfl_utils_split_free(kvs);
+    }
+
+    if (ret == FLB_EVENT_ENCODER_SUCCESS) {
+        ret = flb_log_event_encoder_commit_record(ctx->log_encoder);
+    }
+
+    *out_data = ctx->log_encoder->output_buffer;
+    *out_size = ctx->log_encoder->output_length;
+
+    return 0;
+
+split_error:
+    *out_data = NULL;
+    *out_size = 0;
+
+    return -1;
+}
+
 static struct flb_config_map config_map[] = {
     {
       FLB_CONFIG_MAP_STR, "path", (char *)NULL,
@@ -551,5 +787,9 @@ struct flb_input_plugin in_systemd_plugin = {
     .cb_resume    = in_systemd_resume,
     .cb_exit      = in_systemd_exit,
     .config_map   = config_map,
+
+    /* for testing */
+    .test_formatter.callback = cb_systemd_format_test,
+
     .flags        = 0
 };
diff --git a/plugins/in_systemd/systemd_config.h b/plugins/in_systemd/systemd_config.h
index 83e14856b4d..af789b7ea3b 100644
--- a/plugins/in_systemd/systemd_config.h
+++ b/plugins/in_systemd/systemd_config.h
@@ -63,6 +63,7 @@ struct flb_systemd_config {
     int dynamic_tag;
     int max_fields;            /* max number of fields per record */
     int max_entries;           /* max number of records per iteration */
+    size_t threshold;         /* threshold for retriveing journal */
 
 #ifdef FLB_HAVE_SQLDB
     flb_sds_t db_path;
diff --git a/src/flb_input.c b/src/flb_input.c
index 8b147b4d5ca..089070c297b 100644
--- a/src/flb_input.c
+++ b/src/flb_input.c
@@ -242,6 +242,7 @@ struct flb_input_instance *flb_input_new(struct flb_config *config,
         instance->alias    = NULL;
         instance->id       = id;
         instance->flags    = plugin->flags;
+        instance->test_mode = FLB_FALSE;
         instance->p        = plugin;
         instance->tag      = NULL;
         instance->tag_len  = 0;
@@ -345,6 +346,9 @@ struct flb_input_instance *flb_input_new(struct flb_config *config,
 
         /* processor instance */
         instance->processor = flb_processor_create(config, instance->name, instance, FLB_PLUGIN_INPUT);
+
+        /* Tests */
+        instance->test_formatter.callback = plugin->test_formatter.callback;
     }
 
     return instance;
diff --git a/src/flb_lib.c b/src/flb_lib.c
index b9674e06824..0e4cde0dbaf 100644
--- a/src/flb_lib.c
+++ b/src/flb_lib.c
@@ -347,6 +347,37 @@ int flb_input_set_processor(flb_ctx_t *ctx, int ffd, struct flb_processor *proc)
     return 0;
 }
 
+int flb_input_set_test(flb_ctx_t *ctx, int ffd, char *test_name,
+                       void (*in_callback) (void *, int, int, void *, size_t, void *),
+                       void *in_callback_data)
+{
+    struct flb_input_instance *i_ins;
+
+    i_ins = in_instance_get(ctx, ffd);
+    if (!i_ins) {
+        return -1;
+    }
+
+    /*
+     * Enabling a test, set the output instance in 'test' mode, so no real
+     * flush callback is invoked, only the desired implemented test.
+     */
+
+    /* Formatter test */
+    if (strcmp(test_name, "formatter") == 0) {
+        i_ins->test_mode = FLB_TRUE;
+        i_ins->test_formatter.rt_ctx = ctx;
+        i_ins->test_formatter.rt_ffd = ffd;
+        i_ins->test_formatter.rt_in_callback = in_callback;
+        i_ins->test_formatter.rt_data = in_callback_data;
+    }
+    else {
+        return -1;
+    }
+
+    return 0;
+}
+
 int flb_output_set_http_test(flb_ctx_t *ctx, int ffd, char *test_name,
                              void (*out_response) (void *, int, int, void *, size_t, void *),
                              void *out_callback_data)
@@ -668,6 +699,41 @@ int flb_lib_free(void* data)
     return 0;
 }
 
+static int flb_input_run_formatter(flb_ctx_t *ctx, struct flb_input_instance *i_ins,
+                                   const void *data, size_t len)
+{
+    int ret;
+    void *out_buf = NULL;
+    size_t out_size = 0;
+    struct flb_test_in_formatter *itf;
+
+    if (!i_ins) {
+        return -1;
+    }
+
+    itf = &i_ins->test_formatter;
+
+    /* Invoke the input plugin formatter test callback */
+    ret = itf->callback(ctx->config,
+                        i_ins,
+                        i_ins->context,
+                        data, len,
+                        &out_buf, &out_size);
+
+    /* Call the runtime test callback checker */
+    if (itf->rt_in_callback) {
+        itf->rt_in_callback(itf->rt_ctx,
+                            itf->rt_ffd,
+                            ret,
+                            out_buf, out_size,
+                            itf->rt_data);
+    }
+    else {
+        flb_free(out_buf);
+    }
+
+    return 0;
+}
 
 static int flb_output_run_response(flb_ctx_t *ctx, struct flb_output_instance *o_ins,
                                    int status, const void *data, size_t len)
@@ -720,10 +786,16 @@ int flb_lib_push(flb_ctx_t *ctx, int ffd, const void *data, size_t len)
         return -1;
     }
 
-    ret = flb_pipe_w(i_ins->channel[1], data, len);
-    if (ret == -1) {
-        flb_errno();
-        return -1;
+    /* If input's test_formatter is registered, priorize to run it. */
+    if (i_ins->test_formatter.callback != NULL) {
+        ret = flb_input_run_formatter(ctx, i_ins, data, len);
+    }
+    else {
+        ret = flb_pipe_w(i_ins->channel[1], data, len);
+        if (ret == -1) {
+            flb_errno();
+            return -1;
+        }
     }
     return ret;
 }
diff --git a/tests/runtime/CMakeLists.txt b/tests/runtime/CMakeLists.txt
index f355294ed9f..13acd35b728 100644
--- a/tests/runtime/CMakeLists.txt
+++ b/tests/runtime/CMakeLists.txt
@@ -58,6 +58,9 @@ if(FLB_OUT_LIB)
   FLB_RT_TEST(FLB_IN_FORWARD       "in_forward.c")
   FLB_RT_TEST(FLB_IN_FLUENTBIT_METRICS "in_fluentbit_metrics.c")
   FLB_RT_TEST(FLB_IN_KUBERNETES_EVENTS "in_kubernetes_events.c")
+  if (FLB_IN_SYSTEMD)
+    FLB_RT_TEST(FLB_IN_SYSTEMD        "in_systemd.c")
+  endif ()
 endif()
 
 if (FLB_CUSTOM_CALYPTIA)
diff --git a/tests/runtime/in_systemd.c b/tests/runtime/in_systemd.c
new file mode 100644
index 00000000000..098d5ceb0c4
--- /dev/null
+++ b/tests/runtime/in_systemd.c
@@ -0,0 +1,114 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2022 The Fluent Bit Authors
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+#include <fluent-bit.h>
+#include <fluent-bit/flb_pack.h>
+#include <fluent-bit/flb_record_accessor.h>
+#include <fluent-bit/flb_ra_key.h>
+
+#include "flb_tests_runtime.h"
+
+static void cb_check_cfl_variant_properties(void *ctx, int ffd,
+                                            int res_ret, void *res_data, size_t res_size,
+                                            void *data)
+{
+    flb_sds_t output;
+    char *result = NULL;
+
+    /* Convert from msgpack to JSON */
+    output = flb_msgpack_raw_to_json_sds(res_data, res_size);
+    TEST_CHECK(output != NULL);
+
+    result = strstr(output, "\"MESSAGE\":\"test native message with multiple values\"");
+    if (TEST_CHECK(result != NULL)) {
+        TEST_MSG("output:%s\n", output);
+    }
+
+    result = strstr(output, "\"KEY\":[\"value1\",\"value4\",\"another\"]");
+    if (TEST_CHECK(result != NULL)) {
+        TEST_MSG("output:%s\n", output);
+    }
+
+    result = strstr(output, "\"KEY2\":[\"value2\",\"value3\",\"value5\",\"value10\",\"final_field\"]");
+    if (TEST_CHECK(result != NULL)) {
+        TEST_MSG("output:%s\n", output);
+    }
+
+    result = strstr(output, "\"KEY3\":[\"howdy\",\"prettygood\",\"wow\"]");
+    if (TEST_CHECK(result != NULL)) {
+        TEST_MSG("output:%s\n", output);
+    }
+
+    flb_sds_destroy(output);
+}
+
+void flb_test_duplicated_keys()
+{
+    int ret;
+    int in_ffd;
+    int out_ffd;
+    flb_ctx_t *ctx;
+    char *message = "MESSAGE=test native message with multiple values\nKEY=value1\nKEY=value4\n"
+        "KEY2=value2\nKEY=another\nKEY2=value3\nKEY2=value5\nKEY3=howdy\nKEY3=prettygood\nKEY2=value10\n"
+        "KEY3=wow\nKEY2=final_field\n";
+
+    /* Create context, flush every second (some checks omitted here) */
+    ctx = flb_create();
+    flb_service_set(ctx,
+                    "flush", "2",
+                    "grace", "1",
+                    "Log_Level", "error",
+                    NULL);
+
+    /* Systemd */
+    in_ffd = flb_input(ctx, (char *) "systemd", NULL);
+    flb_input_set(ctx, in_ffd,
+                  "tag", "test",
+                  "Read_From_Tail", "On",
+                  NULL);
+
+
+    out_ffd = flb_output(ctx, (char *) "null", NULL);
+    flb_output_set(ctx, out_ffd,
+                   "match", "test",
+                   NULL);
+
+    /* Enable test mode */
+    ret = flb_input_set_test(ctx, in_ffd, "formatter",
+                             cb_check_cfl_variant_properties,
+                             NULL);
+
+    /* Start */
+    ret = flb_start(ctx);
+    TEST_CHECK(ret == 0);
+
+    /* Ingest data sample to run test formatter */
+    ret = flb_lib_push(ctx, in_ffd, message, strlen(message));
+    TEST_CHECK(ret == 0);
+
+    sleep(2);
+    flb_stop(ctx);
+    flb_destroy(ctx);
+}
+
+/* Test list */
+TEST_LIST = {
+    { "duplicated_keys", flb_test_duplicated_keys },
+    { NULL, NULL}
+};