diff --git a/include/fluent-bit/multiline/flb_ml.h b/include/fluent-bit/multiline/flb_ml.h
new file mode 100644
index 00000000000..4a03f760ac3
--- /dev/null
+++ b/include/fluent-bit/multiline/flb_ml.h
@@ -0,0 +1,215 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2019-2021 The Fluent Bit Authors
+ *  Copyright (C) 2015-2018 Treasure Data Inc.
+ *
+ *  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.
+ */
+
+#ifndef FLB_ML_H
+#define FLB_ML_H
+
+#include <fluent-bit/flb_info.h>
+#include <fluent-bit/flb_config.h>
+#include <fluent-bit/flb_sds.h>
+#include <fluent-bit/flb_log.h>
+#include <fluent-bit/flb_regex.h>
+#include <fluent-bit/flb_mp.h>
+#include <fluent-bit/flb_time.h>
+#include <fluent-bit/flb_parser.h>
+
+/* Types available */
+#define FLB_ML_COUNT     0    /* concatenate N consecutive records  */
+#define FLB_ML_REGEX     1    /* pattern is a regular expression    */
+#define FLB_ML_ENDSWITH  2    /* record key/content ends with 'abc' */
+#define FLB_ML_EQ        3    /* record key/content equaks 'abc'    */
+
+/* Rule types */
+#define FLB_ML_RULE_STATE_START 0
+#define FLB_ML_RULE_STATE_PART  1
+#define FLB_ML_RULE_STATE_END   2
+
+/* Payload types */
+#define FLB_ML_TYPE_TEXT        0   /* raw text */
+#define FLB_ML_TYPE_RECORD      1   /* Fluent Bit msgpack record */
+#define FLB_ML_TYPE_MAP         2   /* msgpack object/map (k/v pairs) */
+
+/* Default multiline buffer size: 4Kb */
+#define FLB_ML_BUF_SIZE  1024*4
+
+struct flb_ml;
+struct flb_ml_stream;
+
+struct flb_ml_rule {
+    /* If the rule contains a 'start_state' this flag is turned on */
+    int start_state;
+
+    /* Name of the rule (state name) */
+    struct mk_list from_states;
+
+    /*
+     * Defined 'to_state' in the rule. Note that this work only at definition
+     * since we do a mapping to multiple possible rules having the same
+     * name.
+     */
+    flb_sds_t to_state;
+
+    struct mk_list to_state_map;
+
+    /* regex content pattern */
+    struct flb_regex *regex;
+
+    /* regex end pattern */
+    struct flb_regex *regex_end;
+
+    struct mk_list _head;
+};
+
+struct flb_ml_stream {
+    flb_sds_t name;           /* Name of the stream, mostly for debugging purposes */
+
+    int counter_lines;        /* counter for the number of lines */
+
+    /* Multiline content buffer */
+    flb_sds_t buf;
+
+    /* internal state */
+    int first_line;           /* first line of multiline message ? */
+
+    struct flb_ml_rule *rule_to_state;
+
+    /* packaging buffers */
+    msgpack_sbuffer mp_sbuf;  /* temporary msgpack buffer              */
+    msgpack_packer mp_pck;    /* temporary msgpack packer              */
+    struct flb_time mp_time;  /* multiline time parsed from first line */
+
+    /* Flush callback and opaque data type */
+    int (*cb_flush) (struct flb_ml *,
+                     struct flb_ml_stream *,
+                     void *cb_data,
+                     void *buf_data,
+                     size_t buf_size);
+    void *cb_data;
+
+    struct mk_list _head;
+};
+
+struct flb_ml {
+    int type;                 /* type: COUNT, REGEX, ENDSWITH or EQ */
+    int negate;               /* negate start pattern ? */
+
+    /*
+     * If 'multiline type' is ENDSWITH or EQ, a 'match_str' string is passed
+     * so we can compare against it. We don't use a regex pattern for efficiency.
+     */
+    flb_sds_t match_str;
+
+    /*
+     * The 'key' name that contains the multiline message. For REGEX, ENDSWITH and
+     * EQ types, the conditions are applied to 'key_content' unless 'key_pattern'
+     * is set, on that case 'key_content' is used as a raw buffer and appended
+     * as part of the multiline message.
+     */
+    flb_sds_t key_content;
+
+    /*
+     * Optional: define a 'key' name that matches the pattern to decide if the
+     * line is complete or a continuation.
+     *
+     * This is not mandatory, most of the rules works directly on 'key_content'
+     * but other use-cases like 'CRI' uses a different 'key' to define if the
+     * line is complete or not.
+     */
+    flb_sds_t key_pattern;
+
+    /* Flush interval */
+    int flush_ms;
+
+    /* flush callback */
+    int (*cb_flush)(struct flb_ml *,           /* multiline context */
+                    struct flb_ml_stream *,    /* stream context */
+                    void *,                    /* opaque data */
+                    char *,                    /* text buffer */
+                    size_t);                   /* buffer length */
+
+    void *cb_data;                             /* opaque data */
+
+    /* internal */
+    struct flb_parser *parser;
+
+    /*
+     * Every multiline context has N streams, a stream represent a source
+     * of data. To avoid having 'multiple' context of 'multiline' we use
+     * streams.
+     */
+    struct mk_list streams;
+
+    /*
+     * If multiline type is REGEX, it needs a set of pre-defined rules to deal
+     * with messages.
+     */
+    struct mk_list regex_rules;
+
+    /* Fluent Bit parent context */
+    struct flb_config *config;
+};
+
+struct flb_ml *flb_ml_create(struct flb_config *ctx,
+                             int type, char *match_str, int negate,
+                             int flush_ms,
+                             char *key_content,
+                             char *key_pattern,
+                             struct flb_parser *parser);
+
+int flb_ml_destroy(struct flb_ml *ml);
+
+int flb_ml_register_context(struct flb_ml *ml, struct flb_ml_stream *mst,
+                            struct flb_time *tm, msgpack_object *map);
+
+int flb_ml_append(struct flb_ml *ml, struct flb_ml_stream *mst,
+                  int type,
+                  struct flb_time *tm, void *buf, size_t size);
+int flb_ml_append_object(struct flb_ml *ml,
+                         struct flb_ml_stream *mst,
+                         struct flb_time *tm, msgpack_object *obj);
+
+int flb_ml_flush(struct flb_ml *ml, struct flb_ml_stream *mst);
+
+/* Multiline streams */
+struct flb_ml_stream *flb_ml_stream_create(struct flb_ml *ml,
+                                           char *name,
+                                           int (*cb_flush) (struct flb_ml *,
+                                                            struct flb_ml_stream *,
+                                                            void *cb_data,
+                                                            void *buf_data,
+                                                            size_t buf_size),
+                                           void *cb_data);
+
+int flb_ml_stream_destroy(struct flb_ml_stream *mst);
+
+/* Regex Rules */
+int flb_ml_rule_create(struct flb_ml *ml,
+                       flb_sds_t from_states,
+                       char *regex_pattern,
+                       flb_sds_t to_state,
+                       char *end_pattern);
+void flb_ml_rule_destroy(struct flb_ml_rule *rule);
+void flb_ml_rule_destroy_all(struct flb_ml *ml);
+
+int flb_ml_init(struct flb_ml *ml);
+
+#include "flb_ml_mode.h"
+
+#endif
diff --git a/include/fluent-bit/multiline/flb_ml_mode.h b/include/fluent-bit/multiline/flb_ml_mode.h
new file mode 100644
index 00000000000..054fc85e4b4
--- /dev/null
+++ b/include/fluent-bit/multiline/flb_ml_mode.h
@@ -0,0 +1,42 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2019-2021 The Fluent Bit Authors
+ *  Copyright (C) 2015-2018 Treasure Data Inc.
+ *
+ *  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.
+ */
+
+#ifndef FLB_ML_MODE_H
+#define FLB_ML_MODE_H
+
+/* Docker container mode */
+struct flb_ml *flb_ml_mode_docker(struct flb_config *config, int flush_ms);
+
+/* CRI container mode */
+struct flb_ml *flb_ml_mode_cri(struct flb_config *config, int flush_ms);
+
+/* Python language mode */
+struct flb_ml *flb_ml_mode_python(struct flb_config *config,
+                                  int flush_ms, char *key);
+
+/* Java language mode */
+struct flb_ml *flb_ml_mode_java(struct flb_config *config,
+                                int flush_ms, char *key);
+
+/* Go language mode */
+struct flb_ml *flb_ml_mode_go(struct flb_config *config,
+                              int flush_ms, char *key);
+
+#endif
diff --git a/include/fluent-bit/multiline/flb_ml_rule.h b/include/fluent-bit/multiline/flb_ml_rule.h
new file mode 100644
index 00000000000..18ceeff3510
--- /dev/null
+++ b/include/fluent-bit/multiline/flb_ml_rule.h
@@ -0,0 +1,43 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2019-2021 The Fluent Bit Authors
+ *  Copyright (C) 2015-2018 Treasure Data Inc.
+ *
+ *  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.
+ */
+
+#ifndef FLB_ML_RULE_H
+#define FLB_ML_RULE_H
+
+#include <fluent-bit/flb_info.h>
+#include <fluent-bit/flb_sds.h>
+#include <fluent-bit/multiline/flb_ml.h>
+
+int flb_ml_rule_create(struct flb_ml *ml,
+                       flb_sds_t from_states,
+                       char *regex_pattern,
+                       flb_sds_t to_state,
+                       char *end_pattern);
+void flb_ml_rule_destroy(struct flb_ml_rule *rule);
+void flb_ml_rule_destroy_all(struct flb_ml *ml);
+int flb_ml_rule_process(struct flb_ml *ml,
+                        struct flb_ml_stream *mst,
+                        msgpack_object *full_map,
+                        void *buf, size_t size, struct flb_time *tm,
+                        msgpack_object *val_content,
+                        msgpack_object *val_pattern);
+int flb_ml_rule_init(struct flb_ml *ml);
+
+#endif
diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index 588a24e5ae9..3039fcd2cad 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -56,6 +56,13 @@ set(src
   flb_routes_mask.c
   )
 
+# Multiline subsystem
+add_subdirectory(multiline)
+set(src
+  ${src}
+  ${src_multiline}
+  )
+
 if(FLB_SYSTEM_WINDOWS)
   set(src
     ${src}
diff --git a/src/flb_config.c b/src/flb_config.c
index a26fd8906b8..0b491b90e64 100644
--- a/src/flb_config.c
+++ b/src/flb_config.c
@@ -345,6 +345,11 @@ void flb_config_exit(struct flb_config *config)
     }
 #endif
 
+#ifdef FLB_HAVE_PARSER
+    /* parsers */
+    flb_parser_exit(config);
+#endif
+
     if (config->storage_path) {
         flb_free(config->storage_path);
     }
diff --git a/src/flb_engine.c b/src/flb_engine.c
index 61ee500774f..d2c42753ec1 100644
--- a/src/flb_engine.c
+++ b/src/flb_engine.c
@@ -713,11 +713,6 @@ int flb_engine_shutdown(struct flb_config *config)
     /* router */
     flb_router_exit(config);
 
-#ifdef FLB_HAVE_PARSER
-    /* parsers */
-    flb_parser_exit(config);
-#endif
-
     /* cleanup plugins */
     flb_filter_exit(config);
     flb_input_exit_all(config);
diff --git a/src/multiline/CMakeLists.txt b/src/multiline/CMakeLists.txt
new file mode 100644
index 00000000000..a38aac07e67
--- /dev/null
+++ b/src/multiline/CMakeLists.txt
@@ -0,0 +1,9 @@
+set(src_multiline
+  multiline/flb_ml_mode_cri.c
+  multiline/flb_ml_mode_docker.c
+  multiline/flb_ml_mode_python.c
+  multiline/flb_ml_mode_java.c
+  multiline/flb_ml_mode_go.c
+  multiline/flb_ml_rule.c
+  multiline/flb_ml.c PARENT_SCOPE
+  )
diff --git a/src/multiline/flb_ml.c b/src/multiline/flb_ml.c
new file mode 100644
index 00000000000..3d55385bbed
--- /dev/null
+++ b/src/multiline/flb_ml.c
@@ -0,0 +1,647 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2019-2021 The Fluent Bit Authors
+ *  Copyright (C) 2015-2018 Treasure Data Inc.
+ *
+ *  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/flb_info.h>
+#include <fluent-bit/flb_mem.h>
+#include <fluent-bit/flb_log.h>
+#include <fluent-bit/flb_regex.h>
+#include <fluent-bit/flb_time.h>
+#include <fluent-bit/flb_pack.h>
+#include <fluent-bit/multiline/flb_ml.h>
+#include <fluent-bit/multiline/flb_ml_rule.h>
+
+struct flb_config_map multiline_map[] = {
+    {
+     FLB_CONFIG_MAP_BOOL, "multiline", "true",
+     0, FLB_TRUE, offsetof(struct flb_net_setup, keepalive),
+     "Enable or disable Keepalive support"
+    },
+
+    /* A pre-defined & built-in mode */
+    {
+     FLB_CONFIG_MAP_BOOL, "multiline.mode", NULL,
+     0, FLB_TRUE, offsetof(struct flb_net_setup, keepalive),
+     "Specify a pre-defined built-in mode to use"
+    },
+
+    {
+     FLB_CONFIG_MAP_BOOL, "multiline.mode", NULL,
+     0, FLB_TRUE, offsetof(struct flb_net_setup, keepalive),
+     "Enable or disable Keepalive support"
+    },
+
+};
+
+static int ml_flush_stdout(struct flb_ml *ml,
+                           struct flb_ml_stream *mst,
+                           void *data, void *buf_data, size_t buf_size)
+{
+    fprintf(stdout, "\n%s----- MULTILINE FLUSH -----%s\n", ANSI_GREEN, ANSI_RESET);
+    fprintf(stdout, "%s%s----------- EOF -----------%s\n",
+            (char *) buf_data, ANSI_GREEN, ANSI_RESET);
+
+    return 0;
+}
+
+static inline int match_negate(struct flb_ml *ml, int matched)
+{
+    int rule_match = matched;
+
+    /* Validate pattern matching against expected 'negate' condition */
+    if (matched == FLB_TRUE) {
+        if (ml->negate == FLB_FALSE) {
+            rule_match = FLB_TRUE;
+        }
+        else {
+            rule_match = FLB_FALSE;
+        }
+    }
+    else {
+        if (ml->negate == FLB_TRUE) {
+            rule_match = FLB_TRUE;
+        }
+    }
+
+    return rule_match;
+}
+
+int flb_ml_register_context(struct flb_ml *ml, struct flb_ml_stream *mst,
+                            struct flb_time *tm, msgpack_object *map)
+{
+    if (tm) {
+        flb_time_copy(&mst->mp_time, tm);
+    }
+
+    if (map) {
+        msgpack_pack_object(&mst->mp_pck, *map);
+    }
+
+    return 0;
+}
+
+/*
+ * package content into a multiline stream:
+ *
+ * full_map: if the original content to process comes in msgpack map, this variable
+ * reference the map. It's only used in case we will package a first line so we
+ * store a copy of the other key values in the map for flush time.
+ */
+static int package_content(struct flb_ml *ml,
+                           struct flb_ml_stream *mst,
+                           msgpack_object *full_map,
+                           void *buf, size_t size, struct flb_time *tm,
+                           msgpack_object *val_content, msgpack_object *val_pattern)
+{
+    int len;
+    int ret;
+    int rule_match = FLB_FALSE;
+    size_t offset = 0;
+    msgpack_object *val = val_content;
+
+    if (val_pattern) {
+        val = val_pattern;
+    }
+
+    if (ml->type == FLB_ML_COUNT) {
+
+    }
+    else if (ml->type == FLB_ML_REGEX) {
+        ret = flb_ml_rule_process(ml, mst, full_map, buf, size, tm,
+                                  val_content, val_pattern);
+    }
+    else if (ml->type == FLB_ML_ENDSWITH) {
+        len = flb_sds_len(ml->match_str);
+        if (val && len <= val->via.str.size) {
+            /* Validate if content ends with expected string */
+            offset = val->via.str.size - len;
+            ret = memcmp(val->via.str.ptr + offset, ml->match_str, len);
+            if (ret == 0) {
+                rule_match = match_negate(ml, FLB_TRUE);
+            }
+            else {
+                rule_match = match_negate(ml, FLB_FALSE);
+            }
+
+            if (mst->mp_sbuf.size == 0) {
+                flb_ml_register_context(ml, mst, tm, full_map);
+            }
+
+            /* Concatenate value */
+            flb_sds_cat(mst->buf,
+                        val_content->via.str.ptr,
+                        val_content->via.str.size);
+
+            /* on ENDSWITH mode, a rule match means flush the content */
+            if (rule_match) {
+                flb_ml_flush(ml, mst);
+            }
+        }
+    }
+    else if (ml->type == FLB_ML_EQ) {
+        if (val->via.str.size == flb_sds_len(ml->match_str) &&
+            memcmp(val->via.str.ptr, ml->match_str, val->via.str.size) == 0) {
+            /* EQ match */
+            rule_match = match_negate(ml, FLB_TRUE);
+        }
+        else {
+            rule_match = match_negate(ml, FLB_FALSE);
+        }
+
+        if (mst->mp_sbuf.size == 0) {
+            flb_ml_register_context(ml, mst, tm, full_map);
+        }
+
+        /* Concatenate value */
+        flb_sds_cat(mst->buf,
+                    val_content->via.str.ptr,
+                    val_content->via.str.size);
+
+        /* on ENDSWITH mode, a rule match means flush the content */
+        if (rule_match) {
+            flb_ml_flush(ml, mst);
+        }
+    }
+
+    return rule_match;
+}
+
+/*
+ * Retrieve the ID of a specific key name in a map. This function might be
+ * extended later to use record accessor, since all current cases are solved
+ * now quering the first level of keys in the map, we avoid record accessor
+ * to avoid extra memory allocations.
+ */
+static int get_key_id(msgpack_object *map, flb_sds_t key_name)
+{
+    int i;
+    int len;
+    int found = FLB_FALSE;
+    msgpack_object key;
+    msgpack_object val;
+
+    len = flb_sds_len(key_name);
+    for (i = 0; i < map->via.map.size; i++) {
+        key = map->via.map.ptr[i].key;
+        val = map->via.map.ptr[i].val;
+
+        if (key.type != MSGPACK_OBJECT_STR || val.type != MSGPACK_OBJECT_STR) {
+            continue;
+        }
+
+        if (key.via.str.size != len) {
+            continue;
+        }
+
+        if (strncmp(key.via.str.ptr, key_name, len) == 0) {
+            found = FLB_TRUE;
+            break;
+        }
+    }
+
+    if (found) {
+        return i;
+    }
+
+    return -1;
+}
+
+static int process_append(struct flb_ml *ml,
+                          struct flb_ml_stream *mst,
+                          int type,
+                          struct flb_time *tm, msgpack_object *obj,
+                          void *buf, size_t size)
+{
+    int ret;
+    int id_content = -1;
+    int id_pattern = -1;
+    int unpacked = FLB_FALSE;
+    size_t off = 0;
+    msgpack_object *full_map = NULL;
+    msgpack_object *val_content;
+    msgpack_object val_pattern;
+    msgpack_unpacked result;
+    struct flb_time tm_record;
+
+    /* Lookup the key */
+    if (type == FLB_ML_TYPE_TEXT) {
+        package_content(ml, mst, NULL, buf, size, tm, NULL, NULL);
+        return 0;
+    }
+    else if (type == FLB_ML_TYPE_RECORD) {
+        off = 0;
+        msgpack_unpacked_init(&result);
+        ret = msgpack_unpack_next(&result, buf, size, &off);
+        if (ret != MSGPACK_UNPACK_SUCCESS) {
+            return -1;
+        }
+        flb_time_pop_from_msgpack(&tm_record, &result, &full_map);
+        unpacked = FLB_TRUE;
+    }
+    else if (type == FLB_ML_TYPE_MAP) {
+        full_map = obj;
+        if (!full_map) {
+            msgpack_unpacked_init(&result);
+            off = 0;
+            ret = msgpack_unpack_next(&result, buf, size, &off);
+            if (ret != MSGPACK_UNPACK_SUCCESS) {
+                return -1;
+            }
+            full_map = &result.data;
+            unpacked = FLB_TRUE;
+        }
+        else if (full_map->type != MSGPACK_OBJECT_MAP) {
+            msgpack_unpacked_destroy(&result);
+            return -1;
+        }
+    }
+
+    /* Lookup for key_content entry */
+    id_content = get_key_id(full_map, ml->key_content);
+    if (id_content == -1) {
+        if (unpacked) {
+            msgpack_unpacked_destroy(&result);
+        }
+        return -1;
+    }
+    val_content = &full_map->via.map.ptr[id_content].val;
+
+    /* Optional: Lookup for key_pattern entry */
+    if (ml->key_pattern) {
+        id_pattern = get_key_id(full_map, ml->key_pattern);
+        if (id_pattern >= 0) {
+            val_pattern = full_map->via.map.ptr[id_pattern].val;
+        }
+    }
+
+    if (id_pattern >= 0) {
+        package_content(ml, mst, full_map, buf, size, tm, val_content, &val_pattern);
+    }
+    else {
+        package_content(ml, mst, full_map, buf, size, tm, val_content, NULL);
+    }
+
+    if (unpacked) {
+        msgpack_unpacked_destroy(&result);
+    }
+    return 0;
+}
+
+int flb_ml_append(struct flb_ml *ml, struct flb_ml_stream *mst,
+                  int type,
+                  struct flb_time *tm, void *buf, size_t size)
+{
+    int ret;
+    int release = FLB_FALSE;
+    void *out_buf = NULL;
+    size_t out_size = 0;
+    struct flb_time out_time = {0};
+
+    if (ml->parser && type == FLB_ML_TYPE_TEXT) {
+        /* Parse incoming content */
+        ret = flb_parser_do(ml->parser, (char *) buf, size,
+                            &out_buf, &out_size, &out_time);
+        if (ret >= 0) {
+            if (flb_time_to_double(&out_time) == 0.0) {
+                flb_time_copy(&out_time, tm);
+            }
+            release = FLB_TRUE;
+            type = FLB_ML_TYPE_MAP;
+        }
+        else {
+            return -1;
+        }
+    }
+    else if (type == FLB_ML_TYPE_TEXT) {
+        out_buf = buf;
+        out_size = size;
+    }
+
+    if (flb_time_to_double(&out_time) == 0.0 && tm) {
+        flb_time_copy(&out_time, tm);
+    }
+    else {
+        flb_time_get(&out_time);
+    }
+
+    /* Process the binary record */
+    process_append(ml, mst, type, &out_time, NULL, out_buf, out_size);
+
+    if (release == FLB_TRUE) {
+        flb_free(out_buf);
+    }
+
+    return 0;
+}
+
+int flb_ml_append_object(struct flb_ml *ml,
+                         struct flb_ml_stream *mst,
+                         struct flb_time *tm, msgpack_object *obj)
+{
+    int type;
+
+    /*
+     * As incoming objects, we only accept Fluent Bit array format
+     * and Map containing key/value pairs.
+     */
+    if (obj->type == MSGPACK_OBJECT_ARRAY) {
+        if (obj->via.array.size != 2) {
+            flb_error("[multiline] appending object with invalid size");
+            return -1;
+        }
+        type = FLB_ML_TYPE_RECORD;
+    }
+    else if (obj->type != MSGPACK_OBJECT_MAP) {
+        flb_error("[multiline] appending object with invalid type, expected "
+                  "array or map, received type=%i", obj->type);
+        return -1;
+    }
+    else {
+        type = FLB_ML_TYPE_MAP;
+    }
+
+    process_append(ml, mst, type, tm, obj, NULL, 0);
+    return 0;
+}
+
+struct flb_ml *flb_ml_create(struct flb_config *ctx,
+                             int type, char *match_str, int negate,
+                             int flush_ms,
+                             char *key_content,
+                             char *key_pattern,
+                             struct flb_parser *parser)
+{
+    struct flb_ml *ml;
+
+    ml = flb_calloc(1, sizeof(struct flb_ml));
+    if (!ml) {
+        flb_errno();
+        return NULL;
+    }
+    ml->type = type;
+
+    if (match_str) {
+        ml->match_str = flb_sds_create(match_str);
+        if (!ml->match_str) {
+            flb_free(ml);
+            return NULL;
+        }
+    }
+    ml->parser = parser;
+    ml->negate = negate;
+
+    if (key_content) {
+        ml->key_content = flb_sds_create(key_content);
+        if (!ml->key_content) {
+            flb_ml_destroy(ml);
+            return NULL;
+        }
+    }
+
+    if (key_pattern) {
+        ml->key_pattern = flb_sds_create(key_pattern);
+        if (!ml->key_pattern) {
+            flb_ml_destroy(ml);
+            return NULL;
+        }
+    }
+    mk_list_init(&ml->streams);
+    mk_list_init(&ml->regex_rules);
+
+    return ml;
+}
+
+int flb_ml_destroy(struct flb_ml *ml)
+{
+    struct mk_list *tmp;
+    struct mk_list *head;
+    struct flb_ml_stream *mst;
+
+    if (ml->match_str) {
+        flb_sds_destroy(ml->match_str);
+    }
+    if (ml->key_content) {
+        flb_sds_destroy(ml->key_content);
+    }
+    if (ml->key_pattern) {
+        flb_sds_destroy(ml->key_pattern);
+    }
+
+    /* Destroy streams */
+    mk_list_foreach_safe(head, tmp, &ml->streams) {
+        mst = mk_list_entry(head, struct flb_ml_stream, _head);
+        flb_ml_stream_destroy(mst);
+    }
+
+    /* Regex rules */
+    flb_ml_rule_destroy_all(ml);
+
+    flb_free(ml);
+    return 0;
+}
+
+int flb_ml_flush(struct flb_ml *ml, struct flb_ml_stream *mst)
+{
+    int i;
+    int ret;
+    int size;
+    int len;
+    size_t off = 0;
+    msgpack_object map;
+    msgpack_object k;
+    msgpack_object v;
+    msgpack_sbuffer mp_sbuf;
+    msgpack_packer mp_pck;
+    msgpack_unpacked result;
+
+    /* init msgpack buffer */
+    msgpack_sbuffer_init(&mp_sbuf);
+    msgpack_packer_init(&mp_pck, &mp_sbuf, msgpack_sbuffer_write);
+
+    /* compose final record if we have a first line context */
+    if (mst->mp_sbuf.size > 0) {
+        msgpack_unpacked_init(&result);
+        ret = msgpack_unpack_next(&result,
+                                  mst->mp_sbuf.data, mst->mp_sbuf.size,
+                                  &off);
+        if (ret != MSGPACK_UNPACK_SUCCESS) {
+            flb_error("[multiline] could not unpack first line state buffer");
+            msgpack_unpacked_destroy(&result);
+            mst->mp_sbuf.size = 0;
+            return -1;
+        }
+        map = result.data;
+
+        if (map.type != MSGPACK_OBJECT_MAP) {
+            flb_error("[multiline] expected MAP type in first line state buffer");
+            msgpack_unpacked_destroy(&result);
+            mst->mp_sbuf.size = 0;
+            return -1;
+        }
+
+        /* Take the first line keys and repack */
+        msgpack_pack_array(&mp_pck, 2);
+        flb_time_append_to_msgpack(&mst->mp_time, &mp_pck, 0);
+
+        len = flb_sds_len(ml->key_content);
+        size = map.via.map.size;
+        msgpack_pack_map(&mp_pck, size);
+
+        for (i = 0; i < size; i++) {
+            k = map.via.map.ptr[i].key;
+            v = map.via.map.ptr[i].val;
+
+            /*
+             * Check if the current key is the key that will contain the
+             * concatenated multiline buffer
+             */
+            if (k.type == MSGPACK_OBJECT_STR &&
+                ml->key_content &&
+                k.via.str.size == len &&
+                strncmp(k.via.str.ptr, ml->key_content, len) == 0) {
+
+                /* key */
+                msgpack_pack_object(&mp_pck, k);
+
+                /* value */
+                len = flb_sds_len(mst->buf);
+                msgpack_pack_str(&mp_pck, len);
+                msgpack_pack_str_body(&mp_pck, mst->buf, len);
+            }
+            else {
+                /* key / val */
+                msgpack_pack_object(&mp_pck, k);
+                msgpack_pack_object(&mp_pck, v);
+            }
+        }
+        msgpack_unpacked_destroy(&result);
+        mst->mp_sbuf.size = 0;
+    }
+    else {
+        /* Pack raw content as Fluent Bit record */
+        msgpack_pack_array(&mp_pck, 2);
+        flb_time_append_to_msgpack(&mst->mp_time, &mp_pck, 0);
+        msgpack_pack_map(&mp_pck, 1);
+
+        /* key */
+        if (ml->key_content) {
+            len = flb_sds_len(ml->key_content);
+            msgpack_pack_str(&mp_pck, len);
+            msgpack_pack_str_body(&mp_pck, ml->key_content, len);
+        }
+        else {
+            msgpack_pack_str(&mp_pck, 3);
+            msgpack_pack_str_body(&mp_pck, "log", 3);
+        }
+
+        /* val */
+        len = flb_sds_len(mst->buf);
+        msgpack_pack_str(&mp_pck, len);
+        msgpack_pack_str_body(&mp_pck, mst->buf, len);
+    }
+
+    mst->cb_flush(ml, mst, mst->cb_data, mp_sbuf.data, mp_sbuf.size);
+    msgpack_sbuffer_destroy(&mp_sbuf);
+    flb_sds_len_set(mst->buf, 0);
+    return 0;
+}
+
+struct flb_ml_stream *flb_ml_stream_create(struct flb_ml *ml,
+                                           char *name,
+                                           int (*cb_flush) (struct flb_ml *,
+                                                            struct flb_ml_stream *,
+                                                            void *cb_data,
+                                                            void *buf_data,
+                                                            size_t buf_size),
+                                           void *cb_data)
+{
+    char tmp[64];
+    struct flb_ml_stream *mst;
+
+    mst = flb_calloc(1, sizeof(struct flb_ml_stream));
+    if (!mst) {
+        flb_errno();
+        return NULL;
+    }
+
+    if (!name) {
+        snprintf(tmp, sizeof(tmp) - 1, "stream-%03i", mk_list_size(&ml->streams));
+        name = tmp;
+    }
+
+    mst->name = flb_sds_create(name);
+    if (!mst->name) {
+        flb_free(mst);
+        return NULL;
+    }
+
+    /* status */
+    mst->first_line = FLB_TRUE;
+    mst->buf = flb_sds_create_size(FLB_ML_BUF_SIZE);
+    if (!mst->buf) {
+        flb_error("cannot allocate multiline stream buffer");
+        flb_free(mst);
+        return NULL;
+    }
+
+    /* Flush Callback and opaque data type */
+    if (cb_flush) {
+        mst->cb_flush = cb_flush;
+    }
+    else {
+        mst->cb_flush = ml_flush_stdout;
+    }
+    mst->cb_data = cb_data;
+
+    /* msgpack buffers */
+    msgpack_sbuffer_init(&mst->mp_sbuf);
+    msgpack_packer_init(&mst->mp_pck, &mst->mp_sbuf, msgpack_sbuffer_write);
+
+    mk_list_add(&mst->_head, &ml->streams);
+    return mst;
+}
+
+int flb_ml_stream_destroy(struct flb_ml_stream *mst)
+{
+    mk_list_del(&mst->_head);
+
+    if (mst->name) {
+        flb_sds_destroy(mst->name);
+    }
+
+    if (mst->buf) {
+        flb_sds_destroy(mst->buf);
+    }
+    msgpack_sbuffer_destroy(&mst->mp_sbuf);
+    flb_free(mst);
+
+    return 0;
+}
+
+int flb_ml_init(struct flb_ml *ml)
+{
+    int ret;
+
+    ret = flb_ml_rule_init(ml);
+    if (ret == -1) {
+        return -1;
+    }
+
+    return 0;
+}
diff --git a/src/multiline/flb_ml_mode_cri.c b/src/multiline/flb_ml_mode_cri.c
new file mode 100644
index 00000000000..277a74d639a
--- /dev/null
+++ b/src/multiline/flb_ml_mode_cri.c
@@ -0,0 +1,75 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2019-2021 The Fluent Bit Authors
+ *  Copyright (C) 2015-2018 Treasure Data Inc.
+ *
+ *  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/flb_info.h>
+#include <fluent-bit/multiline/flb_ml.h>
+
+#define FLB_ML_CRI_REGEX                                                \
+  "^(?<time>.+) (?<stream>stdout|stderr) (?<_p>F|P) (?<log>.*)$"
+#define FLB_ML_CRI_TIME                         \
+  "%Y-%m-%dT%H:%M:%S.%L%:z"
+
+/* Creates a parser for Docker */
+static struct flb_parser *cri_parser_create(struct flb_config *config)
+{
+    struct flb_parser *p;
+
+    p = flb_parser_create("_ml_cri",               /* parser name */
+                          "regex",                 /* backend type */
+                          FLB_ML_CRI_REGEX,        /* regex */
+                          FLB_ML_CRI_TIME,         /* time format */
+                          "time",                  /* time key */
+                          NULL,                    /* time offset */
+                          FLB_TRUE,                /* time keep */
+                          FLB_FALSE,               /* time strict */
+                          NULL,                    /* parser types */
+                          0,                       /* types len */
+                          NULL,                    /* decoders */
+                          config);                 /* Fluent Bit context */
+    return p;
+}
+
+/* Our first multiline mode: 'docker' */
+struct flb_ml *flb_ml_mode_cri(struct flb_config *config, int flush_ms)
+{
+    struct flb_parser *parser;
+    struct flb_ml *ml;
+
+    /* Create a Docker parser */
+    parser = cri_parser_create(config);
+    if (!parser) {
+        return NULL;
+    }
+
+    ml = flb_ml_create(config,
+                       FLB_ML_EQ,           /* type      */
+                       "F",                 /* match_str */
+                       FLB_FALSE,           /* negate    */
+                       flush_ms,            /* flush_ms  */
+                       "log",               /* key_content */
+                       "_p",                /* key_pattern */
+                       parser);             /* parser */
+    if (!ml) {
+        flb_error("[multiline] could not create 'docker mode'");
+        return NULL;
+    }
+
+    return ml;
+}
diff --git a/src/multiline/flb_ml_mode_docker.c b/src/multiline/flb_ml_mode_docker.c
new file mode 100644
index 00000000000..b42ab3fa46d
--- /dev/null
+++ b/src/multiline/flb_ml_mode_docker.c
@@ -0,0 +1,106 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2019-2021 The Fluent Bit Authors
+ *  Copyright (C) 2015-2018 Treasure Data Inc.
+ *
+ *  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/flb_info.h>
+#include <fluent-bit/multiline/flb_ml.h>
+
+/* Creates a parser for Docker */
+static struct flb_parser *docker_parser_create(struct flb_config *config)
+{
+    struct flb_parser *p;
+
+    p = flb_parser_create("_ml_json_docker",      /* parser name */
+                          "json",                 /* backend type */
+                          NULL,                   /* regex */
+                          "%Y-%m-%dT%H:%M:%S.%L", /* time format */
+                          "time",                 /* time key */
+                          NULL,                   /* time offset */
+                          FLB_TRUE,               /* time keep */
+                          FLB_FALSE,              /* time strict */
+                          NULL,                   /* parser types */
+                          0,                      /* types len */
+                          NULL,                   /* decoders */
+                          config);                /* Fluent Bit context */
+    return p;
+}
+
+/* Our first multiline mode: 'docker' */
+struct flb_ml *flb_ml_mode_docker(struct flb_config *config,
+                                                int flush_ms)
+{
+    struct flb_parser *parser;
+    struct flb_ml *ml;
+
+    /* Create a Docker parser */
+    parser = docker_parser_create(config);
+    if (!parser) {
+        return NULL;
+    }
+
+    /*
+     * Let's explain this multiline mode, then you (the reader) might want
+     * to submit a PR with new built-in modes :)
+     *
+     * Containerized apps under Docker writes logs to stdout/stderr. These streams
+     * (stdout/stderr) are handled by Docker, in most of cases the content is
+     * stored in a .json file in your file system. A message like "hey!" gets into
+     * a JSON map like this:
+     *
+     * {"log": "hey!\n", "stream": "stdout", "time": "2021-02-01T01:40:03.53412Z"}
+     *
+     * By Docker log spec, any 'log' key that "ends with a \n" it's a complete
+     * log record, but Docker also limits the log record size to 16KB, so a long
+     * message that does not fit into 16KB can be split in multiple JSON lines,
+     * the following example use short words to describe the context:
+     *
+     * - original message: 'one, two, three\n'
+     *
+     * Docker log interpretation:
+     *
+     * - {"log": "one, ", "stream": "stdout", "time": "2021-02-01T01:40:03.53413Z"}
+     * - {"log": "two, ", "stream": "stdout", "time": "2021-02-01T01:40:03.53414Z"}
+     * - {"log": "three\n", "stream": "stdout", "time": "2021-02-01T01:40:03.53415Z"}
+     *
+     * So every 'log' key that does not ends with '\n', it's a partial log record
+     * and for logging purposes it needs to be concatenated with further messages
+     * until a final '\n' is found.
+     *
+     * We setup the Multiline mode as follows:
+     *
+     * - Use the type 'FLB_ML_ENDSWITH' to specify that we expect the 'log'
+     *   key must ends with a '\n' for complete messages, otherwise it means is
+     *   a continuation message. In case a message is not complete just wait until
+     *   500 milliseconds (0.5 second) and flush the buffer.
+     */
+    ml = flb_ml_create(config,                  /* Fluent Bit context */
+                       FLB_ML_ENDSWITH,         /* type           */
+                       "\n",                    /* match_str      */
+                       FLB_FALSE,               /* negate         */
+                       flush_ms,                /* flush_ms       */
+                       "log",                   /* key_content    */
+                       NULL,                    /* key_pattern    */
+                       parser);                 /* parser         */
+    if (!ml) {
+        flb_error("[multiline] could not create 'docker mode'");
+        return NULL;
+    }
+
+    return ml;
+}
diff --git a/src/multiline/flb_ml_mode_go.c b/src/multiline/flb_ml_mode_go.c
new file mode 100644
index 00000000000..9cc3666621b
--- /dev/null
+++ b/src/multiline/flb_ml_mode_go.c
@@ -0,0 +1,137 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2019-2021 The Fluent Bit Authors
+ *  Copyright (C) 2015-2018 Treasure Data Inc.
+ *
+ *  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/flb_info.h>
+#include <fluent-bit/multiline/flb_ml.h>
+
+#define rule flb_ml_rule_create
+
+static void rule_error(struct flb_ml *ml)
+{
+    int id;
+
+    id = mk_list_size(&ml->regex_rules);
+    flb_error("[multiline: go] rule #%i could not be created", id);
+    flb_ml_destroy(ml);
+}
+
+/* Golang mode */
+struct flb_ml *flb_ml_mode_go(struct flb_config *config, int flush_ms,
+                              char *key)
+{
+    int ret;
+    struct flb_ml *ml;
+
+    ml = flb_ml_create(config,          /* Fluent Bit context */
+                       FLB_ML_REGEX,    /* type      */
+                       NULL,            /* match_str */
+                       FLB_FALSE,       /* negate    */
+                       flush_ms,        /* flush_ms  */
+                       key,             /* key_content */
+                       NULL,            /* key_pattern */
+                       NULL);           /* parser */
+
+    if (!ml) {
+        flb_error("[multiline] could not create 'python mode'");
+        return NULL;
+    }
+
+    ret = rule(ml,
+               "start_state",
+               "/\\bpanic: /",
+               "go_after_panic", NULL);
+    if (ret != 0) {
+        rule_error(ml);
+        return NULL;
+    }
+
+    ret = rule(ml,
+               "start_state",
+               "/http: panic serving/",
+               "go_goroutine", NULL);
+    if (ret != 0) {
+        rule_error(ml);
+        return NULL;
+    }
+
+    ret = rule(ml,
+               "go_after_panic",
+               "/^$/",
+               "go_goroutine", NULL);
+    if (ret != 0) {
+        rule_error(ml);
+        return NULL;
+    }
+
+    ret = rule(ml,
+               "go_after_panic, go_after_signal, go_frame_1",
+               "/^$/",
+               "go_goroutine", NULL);
+    if (ret != 0) {
+        rule_error(ml);
+        return NULL;
+    }
+
+    ret = rule(ml,
+               "go_after_panic",
+               "/^\\[signal /",
+               "go_after_signal", NULL);
+    if (ret != 0) {
+        rule_error(ml);
+        return NULL;
+    }
+
+    ret = rule(ml,
+               "go_goroutine",
+               "/^goroutine \\d+ \\[[^\\]]+\\]:$/",
+               "go_frame_1", NULL);
+    if (ret != 0) {
+        rule_error(ml);
+        return NULL;
+    }
+
+    ret = rule(ml,
+               "go_frame_1",
+               "/^(?:[^\\s.:]+\\.)*[^\\s.():]+\\(|^created by /",
+               "go_frame_2", NULL);
+    if (ret != 0) {
+        rule_error(ml);
+        return NULL;
+    }
+
+    ret = rule(ml,
+               "go_frame_2",
+               "/^\\s/",
+               "go_frame_1", NULL);
+    if (ret != 0) {
+        rule_error(ml);
+        return NULL;
+    }
+
+    /* Map the rules (mandatory for regex rules) */
+    ret = flb_ml_init(ml);
+    if (ret != 0) {
+        flb_error("[multiline: python] error on mapping rules");
+        flb_ml_destroy(ml);
+        return NULL;
+    }
+
+    return ml;
+}
diff --git a/src/multiline/flb_ml_mode_java.c b/src/multiline/flb_ml_mode_java.c
new file mode 100644
index 00000000000..7f4073514d1
--- /dev/null
+++ b/src/multiline/flb_ml_mode_java.c
@@ -0,0 +1,140 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2019-2021 The Fluent Bit Authors
+ *  Copyright (C) 2015-2018 Treasure Data Inc.
+ *
+ *  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/flb_info.h>
+#include <fluent-bit/multiline/flb_ml.h>
+
+#define rule flb_ml_rule_create
+
+static void rule_error(struct flb_ml *ml)
+{
+    int id;
+
+    id = mk_list_size(&ml->regex_rules);
+    flb_error("[multiline: java] rule #%i could not be created", id);
+    flb_ml_destroy(ml);
+}
+
+/* Java mode */
+struct flb_ml *flb_ml_mode_java(struct flb_config *config, int flush_ms,
+                                char *key)
+{
+    int ret;
+    struct flb_ml *ml;
+
+    ml = flb_ml_create(config,          /* Fluent Bit context */
+                       FLB_ML_REGEX,    /* type      */
+                       NULL,            /* match_str */
+                       FLB_FALSE,       /* negate    */
+                       flush_ms,        /* flush_ms  */
+                       key,             /* key_content */
+                       NULL,            /* key_pattern */
+                       NULL);           /* parser */
+
+    if (!ml) {
+        flb_error("[multiline] could not create 'python mode'");
+        return NULL;
+    }
+
+    ret = rule(ml,
+               "start_state, java_start_exception",
+               "/(?:Exception|Error|Throwable|V8 errors stack trace)[:\\r\\n]/",
+               "java_after_exception", NULL);
+    if (ret != 0) {
+      rule_error(ml);
+      return NULL;
+    }
+
+    ret = rule(ml,
+               "java_after_exception",
+               "/^[\\t ]*nested exception is:[\\t ]*/",
+               "java_start_exception", NULL);
+    if (ret != 0) {
+        rule_error(ml);
+        return NULL;
+    }
+
+    ret = rule(ml,
+               "java_after_exception",
+               "/^[\\r\\n]*$/",
+               "java_after_exception", NULL);
+    if (ret != 0) {
+        rule_error(ml);
+        return NULL;
+    }
+
+    ret = rule(ml,
+               "java_after_exception, java",
+               "/^[\t ]+(?:eval )?at /",
+               "java", NULL);
+    if (ret != 0) {
+        rule_error(ml);
+        return NULL;
+    }
+
+    ret = rule(ml,
+               "java_after_exception, java",
+               /* C# nested exception */
+               "/^[\\t ]+--- End of inner exception stack trace ---$/",
+               "java", NULL);
+    if (ret != 0) {
+        rule_error(ml);
+        return NULL;
+    }
+
+    ret = rule(ml,
+               "java_after_exception, java",
+               /* C# exception from async code */
+               "/^--- End of stack trace from previous (?x:"
+               ")location where exception was thrown ---$/",
+               "java", NULL);
+    if (ret != 0) {
+        rule_error(ml);
+        return NULL;
+    }
+
+    ret = rule(ml,
+               "java_after_exception, java",
+               "/^[\t ]*(?:Caused by|Suppressed):/",
+               "java_after_exception", NULL);
+    if (ret != 0) {
+        rule_error(ml);
+        return NULL;
+    }
+
+    ret = rule(ml,
+               "java_after_exception, java",
+               "/^[\t ]*... \\d+ (?:more|common frames omitted)/",
+               "java", NULL);
+    if (ret != 0) {
+        rule_error(ml);
+        return NULL;
+    }
+
+    /* Map the rules (mandatory for regex rules) */
+    ret = flb_ml_init(ml);
+    if (ret != 0) {
+        flb_error("[multiline: python] error on mapping rules");
+        flb_ml_destroy(ml);
+        return NULL;
+    }
+
+    return ml;
+}
diff --git a/src/multiline/flb_ml_mode_python.c b/src/multiline/flb_ml_mode_python.c
new file mode 100644
index 00000000000..d0b2d3a3f89
--- /dev/null
+++ b/src/multiline/flb_ml_mode_python.c
@@ -0,0 +1,96 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2019-2021 The Fluent Bit Authors
+ *  Copyright (C) 2015-2018 Treasure Data Inc.
+ *
+ *  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/flb_info.h>
+#include <fluent-bit/multiline/flb_ml.h>
+
+#define rule flb_ml_rule_create
+
+static void rule_error(struct flb_ml *ml)
+{
+    int id;
+
+    id = mk_list_size(&ml->regex_rules);
+    flb_error("[multiline: python] rule #%i could not be created", id);
+    flb_ml_destroy(ml);
+}
+
+/* Our first multiline mode: 'docker' */
+struct flb_ml *flb_ml_mode_python(struct flb_config *config,
+                                                int flush_ms,
+                                                char *key)
+{
+    int ret;
+    struct flb_ml *ml;
+
+    ml = flb_ml_create(config,          /* Fluent Bit context */
+                       FLB_ML_REGEX,    /* type      */
+                       NULL,            /* match_str */
+                       FLB_FALSE,       /* negate    */
+                       flush_ms,        /* flush_ms  */
+                       key,             /* key_content */
+                       NULL,            /* key_pattern */
+                       NULL);           /* parser */
+
+    if (!ml) {
+        flb_error("[multiline] could not create 'python mode'");
+        return NULL;
+    }
+
+    /* rule(:start_state, /^Traceback \(most recent call last\):$/, :python) */
+    ret = rule(ml,
+               "start_state", "/^Traceback \\(most recent call last\\):$/",
+               "python", NULL);
+    if (ret != 0) {
+        rule_error(ml);
+        return NULL;
+    }
+
+    /* rule(:python, /^[\t ]+File /, :python_code) */
+    ret = rule(ml, "python", "/^[\\t ]+File /", "python_code", NULL);
+    if (ret != 0) {
+        rule_error(ml);
+        return NULL;
+    }
+
+    /* rule(:python_code, /[^\t ]/, :python) */
+    ret = rule(ml, "python_code", "/[^\\t ]/", "python", NULL);
+    if (ret != 0) {
+        rule_error(ml);
+        return NULL;
+    }
+
+    /* rule(:python, /^(?:[^\s.():]+\.)*[^\s.():]+:/, :start_state) */
+    ret = rule(ml, "python", "/^(?:[^\\s.():]+\\.)*[^\\s.():]+:/", "start_state", NULL);
+    if (ret != 0) {
+        rule_error(ml);
+        return NULL;
+    }
+
+    /* Map the rules (mandatory for regex rules) */
+    ret = flb_ml_init(ml);
+    if (ret != 0) {
+        flb_error("[multiline: python] error on mapping rules");
+        flb_ml_destroy(ml);
+        return NULL;
+    }
+
+    return ml;
+}
diff --git a/src/multiline/flb_ml_rule.c b/src/multiline/flb_ml_rule.c
new file mode 100644
index 00000000000..f77f0a8b697
--- /dev/null
+++ b/src/multiline/flb_ml_rule.c
@@ -0,0 +1,360 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2019-2021 The Fluent Bit Authors
+ *  Copyright (C) 2015-2018 Treasure Data Inc.
+ *
+ *  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/flb_info.h>
+#include <fluent-bit/flb_regex.h>
+#include <fluent-bit/flb_slist.h>
+#include <fluent-bit/multiline/flb_ml.h>
+
+struct to_state {
+    struct flb_ml_rule *rule;
+    struct mk_list _head;
+};
+
+struct flb_slist_entry *get_start_state(struct mk_list *list)
+{
+    struct mk_list *head;
+    struct flb_slist_entry *e;
+
+    mk_list_foreach(head, list) {
+        e = mk_list_entry(head, struct flb_slist_entry, _head);
+        if (strcmp(e->str, "start_state") == 0) {
+            return e;
+        }
+    }
+
+    return NULL;
+}
+
+int flb_ml_rule_create(struct flb_ml *ml,
+                              flb_sds_t from_states,
+                              char *regex_pattern,
+                              flb_sds_t to_state,
+                              char *end_pattern)
+{
+    int ret;
+    struct flb_ml_rule *rule;
+
+    rule = flb_calloc(1, sizeof(struct flb_ml_rule));
+    if (!rule) {
+        flb_errno();
+        return -1;
+    }
+    flb_slist_create(&rule->from_states);
+    mk_list_init(&rule->to_state_map);
+    mk_list_add(&rule->_head, &ml->regex_rules);
+
+    /* from_states */
+    ret = flb_slist_split_string(&rule->from_states, from_states, ',', -1);
+    if (ret <= 0) {
+        flb_error("[multiline] rule is empty or has invalid 'from_states' tokens");
+        flb_ml_rule_destroy(rule);
+        return -1;
+    }
+
+    /* Check if the rule contains a 'start_state' */
+    if (get_start_state(&rule->from_states)) {
+        rule->start_state = FLB_TRUE;
+    }
+
+    /* regex content pattern */
+    rule->regex = flb_regex_create(regex_pattern);
+    if (!rule->regex) {
+        flb_ml_rule_destroy(rule);
+        return -1;
+    }
+
+    /* to_state */
+    rule->to_state = flb_sds_create(to_state);
+    if (!rule->to_state) {
+        flb_ml_rule_destroy(rule);
+        return -1;
+    }
+
+    /* regex end pattern */
+    if (end_pattern) {
+        rule->regex_end = flb_regex_create(end_pattern);
+        if (!rule->regex_end) {
+            flb_ml_rule_destroy(rule);
+            return -1;
+        }
+    }
+
+    return 0;
+}
+
+void flb_ml_rule_destroy(struct flb_ml_rule *rule)
+{
+    struct mk_list *tmp;
+    struct mk_list *head;
+    struct to_state *st;
+
+    flb_slist_destroy(&rule->from_states);
+
+    if (rule->regex) {
+        flb_regex_destroy(rule->regex);
+    }
+
+
+    if (rule->to_state) {
+        flb_sds_destroy(rule->to_state);
+    }
+
+    mk_list_foreach_safe(head, tmp, &rule->to_state_map) {
+        st = mk_list_entry(head, struct to_state, _head);
+        mk_list_del(&st->_head);
+        flb_free(st);
+    }
+
+    if (rule->regex_end) {
+        flb_regex_destroy(rule->regex_end);
+    }
+
+    mk_list_del(&rule->_head);
+    flb_free(rule);
+}
+
+void flb_ml_rule_destroy_all(struct flb_ml *ml)
+{
+    struct mk_list *tmp;
+    struct mk_list *head;
+    struct flb_ml_rule *rule;
+
+    mk_list_foreach_safe(head, tmp, &ml->regex_rules) {
+        rule = mk_list_entry(head, struct flb_ml_rule, _head);
+        flb_ml_rule_destroy(rule);
+    }
+}
+
+static inline int to_states_matches_rule(struct flb_ml_rule *rule,
+                                         flb_sds_t state)
+{
+    struct mk_list *head;
+    struct flb_slist_entry *e;
+
+    mk_list_foreach(head, &rule->from_states) {
+        e = mk_list_entry(head, struct flb_slist_entry, _head);
+        if (strcmp(e->str, state) == 0) {
+            return FLB_TRUE;
+        }
+    }
+
+    return FLB_FALSE;
+}
+
+static int set_to_state_map(struct flb_ml *ml,
+                            struct flb_ml_rule *rule)
+{
+    int ret;
+    struct to_state *s;
+    struct mk_list *head;
+    struct flb_ml_rule *r;
+
+    if (!rule->to_state) {
+        /* no to_state */
+        return 0;
+    }
+
+    /* Iterate all rules that matches the to_state */
+    mk_list_foreach(head, &ml->regex_rules) {
+        r = mk_list_entry(head, struct flb_ml_rule, _head);
+
+        /*
+         * A rule can have many 'from_states', check if the current 'rule->to_state'
+         * matches any 'r->from_states'
+         */
+        ret = to_states_matches_rule(r, rule->to_state);
+        if (!ret) {
+            continue;
+        }
+
+        /* We have a match. Create a 'to_state' entry into the 'to_state_map' list */
+        s = flb_malloc(sizeof(struct to_state));
+        if (!s) {
+            flb_errno();
+            return -1;
+        }
+        s->rule = r;
+        mk_list_add(&s->_head, &rule->to_state_map);
+    }
+
+    return 0;
+}
+
+static int try_flushing_buffer(struct flb_ml *ml,
+                               struct flb_ml_stream *mst)
+{
+    int next_start = FLB_FALSE;
+    struct mk_list *head;
+    struct to_state *st;
+    struct flb_ml_rule *rule;
+
+    rule = mst->rule_to_state;
+    if (!rule) {
+        if (flb_sds_len(mst->buf) > 0) {
+            flb_ml_flush(ml, mst);
+            mst->first_line = FLB_TRUE;
+        }
+        return 0;
+    }
+
+    /* Check if any 'to_state_map' referenced rules is a possible start */
+    mk_list_foreach(head, &rule->to_state_map) {
+        st = mk_list_entry(head, struct to_state, _head);
+        if (st->rule->start_state) {
+            next_start = FLB_TRUE;
+            break;
+        }
+    }
+
+    if (next_start && flb_sds_len(mst->buf) > 0) {
+        flb_ml_flush(ml, mst);
+        mst->first_line = FLB_TRUE;
+    }
+
+    return 0;
+}
+
+/* Initialize all rules */
+int flb_ml_rule_init(struct flb_ml *ml)
+{
+    int ret;
+    struct mk_list *head;
+    struct flb_ml_rule *rule;
+
+    /* FIXME: sort rules by start_state first (let's trust in the caller) */
+
+    /* For each rule, compose it to_state_map list */
+    mk_list_foreach(head, &ml->regex_rules) {
+        rule = mk_list_entry(head, struct flb_ml_rule, _head);
+        /* Populate 'rule->to_state_map' list */
+        ret = set_to_state_map(ml, rule);
+        if (ret == -1) {
+            return -1;
+        }
+    }
+
+    return 0;
+}
+
+
+int flb_ml_rule_process(struct flb_ml *ml,
+                        struct flb_ml_stream *mst,
+                        msgpack_object *full_map,
+                        void *buf, size_t size, struct flb_time *tm,
+                        msgpack_object *val_content,
+                        msgpack_object *val_pattern)
+{
+    int ret;
+    char *buf_data;
+    size_t buf_size;
+    struct mk_list *head;
+    struct to_state *st = NULL;
+    struct flb_ml_rule *rule = NULL;
+    struct flb_ml_rule *tmp_rule = NULL;
+
+    if (val_content) {
+        buf_data = (char *) val_content->via.str.ptr;
+        buf_size = val_content->via.str.size;
+    }
+    else {
+        buf_data = buf;
+        buf_size = size;
+    }
+
+    if (mst->first_line) {
+        mst->rule_to_state = NULL;
+
+        /* If a previous content exists, just flush it */
+        if (flb_sds_len(mst->buf) > 0) {
+            flb_ml_flush(ml, mst);
+        }
+
+        /* If this is a first line, check for any rule that matches a start state */
+        mk_list_foreach(head, &ml->regex_rules) {
+            rule = mk_list_entry(head, struct flb_ml_rule, _head);
+
+            /* Is this rule matching a start_state ? */
+            if (!rule->start_state) {
+                rule = NULL;
+                continue;
+            }
+
+            /* Matched a start_state. Check if we have a regex match */
+            ret = flb_regex_match(rule->regex,
+                                  (unsigned char *) buf_data,
+                                  buf_size);
+            if (ret) {
+                /* Regex matched */
+                flb_sds_cat(mst->buf, buf_data, buf_size);
+                mst->first_line = FLB_FALSE;
+
+                /* Copy full map content in stream buffer */
+                flb_ml_register_context(ml, mst, tm, full_map);
+                break;
+            }
+            rule = NULL;
+        }
+    }
+    else if (mst->rule_to_state) {
+        tmp_rule = mst->rule_to_state;
+        rule = NULL;
+
+        /* Lookup all possible next rules by state reference */
+        mk_list_foreach(head, &tmp_rule->to_state_map) {
+            st = mk_list_entry(head, struct to_state, _head);
+
+            /* Try regex match */
+            ret = flb_regex_match(st->rule->regex,
+                                  (unsigned char *) buf_data, buf_size);
+            if (ret) {
+                /* Regex matched */
+                flb_sds_cat(mst->buf, buf_data, buf_size);
+                rule = st->rule;
+                break;
+            }
+            rule = NULL;
+        }
+    }
+
+    /*
+     * If 'rule' is set means we got a rule regex  match. This rule might have a
+     * to_state defined.
+     */
+    if (rule) {
+        /*
+         * reference the rule that recently matched the pattern. So on the
+         * next iteration we can query the possible 'to_states' in the list.
+         */
+        mst->rule_to_state = rule;
+        try_flushing_buffer(ml, mst);
+    }
+    else {
+        /* Flush any previous content */
+        mst->rule_to_state = NULL;
+        try_flushing_buffer(ml, mst);
+
+        /* Append un-matched content to buffer and flush */
+        flb_sds_cat(mst->buf, buf_data, buf_size);
+        try_flushing_buffer(ml, mst);
+    }
+
+    return 0;
+}
diff --git a/tests/internal/CMakeLists.txt b/tests/internal/CMakeLists.txt
index e4ad3626ffa..f49930e1bfc 100644
--- a/tests/internal/CMakeLists.txt
+++ b/tests/internal/CMakeLists.txt
@@ -20,6 +20,7 @@ set(UNIT_TESTS_FILES
   mp.c
   input_chunk.c
   flb_time.c
+  multiline.c
   )
 
 if (NOT WIN32)
diff --git a/tests/internal/multiline.c b/tests/internal/multiline.c
new file mode 100644
index 00000000000..66179c4b83f
--- /dev/null
+++ b/tests/internal/multiline.c
@@ -0,0 +1,685 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+#include <fluent-bit/flb_info.h>
+#include <fluent-bit/flb_pack.h>
+#include <fluent-bit/flb_parser.h>
+#include <fluent-bit/flb_mem.h>
+#include <fluent-bit/multiline/flb_ml.h>
+#include "flb_tests_internal.h"
+
+struct record_check {
+    char *buf;
+};
+
+struct expected_result {
+    int current_record;
+    char *key;
+    struct record_check *out_records;
+};
+
+struct record_check docker_input[] = {
+  {"{\"log\": \"aa\\n\", \"stream\": \"stdout\", \"time\": \"2021-02-01T16:45:03.01231z\"}"},
+  {"{\"log\": \"bb\", \"stream\": \"stdout\", \"time\": \"2021-02-01T16:45:03.01232z\"}"},
+  {"{\"log\": \"cc\", \"stream\": \"stdout\", \"time\": \"2021-02-01T16:45:03.01233z\"}"},
+  {"{\"log\": \"dd\\n\", \"stream\": \"stdout\", \"time\": \"2021-02-01T16:45:03.01234z\"}"},
+  {"{\"log\": \"ee\\n\", \"stream\": \"stdout\", \"time\": \"2021-02-01T16:45:03.01235z\"}"},
+  {"{\"log\": \"ff\", \"stream\": \"stdout\", \"time\": \"2021-02-01T16:45:03.01236z\"}"},
+  {"{\"log\": \"gg\\n\", \"stream\": \"stdout\", \"time\": \"2021-02-01T16:45:03.01237z\"}"},
+};
+
+struct record_check docker_output[] = {
+  {"aa\n"},
+  {"bbccdd\n"},
+  {"ee\n"},
+  {"ffgg\n"}
+};
+
+struct record_check cri_input[] = {
+  {"2019-05-07T18:57:50.904275087+00:00 stdout P 1. some "},
+  {"2019-05-07T18:57:51.904275088+00:00 stdout P multiline "},
+  {"2019-05-07T18:57:52.904275089+00:00 stdout F log"},
+  {"2019-05-07T18:57:53.904275090+00:00 stdout P 2. another "},
+  {"2019-05-07T18:57:54.904275091+00:00 stdout P multiline "},
+  {"2019-05-07T18:57:55.904275092+00:00 stdout F log"},
+  {"2019-05-07T18:57:56.904275093+00:00 stdout F 3. non multiline 1"},
+  {"2019-05-07T18:57:57.904275094+00:00 stdout F 4. non multiline 2"}
+};
+
+struct record_check cri_output[] = {
+  {"1. some multiline log"},
+  {"2. another multiline log"},
+  {"3. non multiline 1"},
+  {"4. non multiline 2"}
+};
+
+struct record_check java_input[] = {
+  {"Exception in thread \"main\" java.lang.IllegalStateException: ..null property\n"},
+  {"     at com.example.myproject.Author.getBookIds(xx.java:38)\n"},
+  {"     at com.example.myproject.Bootstrap.main(Bootstrap.java:14)\n"},
+  {"Caused by: java.lang.NullPointerException\n"},
+  {"     at com.example.myproject.Book.getId(Book.java:22)\n"},
+  {"     at com.example.myproject.Author.getBookIds(Author.java:35)\n"},
+  {"     ... 1 more"},
+  {"single line"}
+};
+
+struct record_check java_output[] = {
+  {
+    "Exception in thread \"main\" java.lang.IllegalStateException: ..null property\n"
+    "     at com.example.myproject.Author.getBookIds(xx.java:38)\n"
+    "     at com.example.myproject.Bootstrap.main(Bootstrap.java:14)\n"
+    "Caused by: java.lang.NullPointerException\n"
+    "     at com.example.myproject.Book.getId(Book.java:22)\n"
+    "     at com.example.myproject.Author.getBookIds(Author.java:35)\n"
+    "     ... 1 more"
+  },
+  {
+    "single line"
+  }
+};
+
+struct record_check python_input[] = {
+  {"Traceback (most recent call last):\n"},
+  {"  File \"/base/data/home/runtimes/python27/python27_lib/versions/third_party/webapp2-2.5.2/webapp2.py\", line 1535, in __call__\n"},
+  {"    rv = self.handle_exception(request, response, e)\n"},
+  {"  File \"/base/data/home/apps/s~nearfieldspy/1.378705245900539993/nearfieldspy.py\", line 17, in start\n"},
+  {"    return get()\n"},
+  {"  File \"/base/data/home/apps/s~nearfieldspy/1.378705245900539993/nearfieldspy.py\", line 5, in get\n"},
+  {"    raise Exception('spam', 'eggs')\n"},
+  {"Exception: ('spam', 'eggs')\n"},
+  {"hello world, not multiline\n"}
+};
+
+struct record_check python_output[] = {
+  {
+      "Traceback (most recent call last):\n"
+      "  File \"/base/data/home/runtimes/python27/python27_lib/versions/third_party/webapp2-2.5.2/webapp2.py\", line 1535, in __call__\n"
+      "    rv = self.handle_exception(request, response, e)\n"
+      "  File \"/base/data/home/apps/s~nearfieldspy/1.378705245900539993/nearfieldspy.py\", line 17, in start\n"
+      "    return get()\n"
+      "  File \"/base/data/home/apps/s~nearfieldspy/1.378705245900539993/nearfieldspy.py\", line 5, in get\n"
+      "    raise Exception('spam', 'eggs')\n"
+      "Exception: ('spam', 'eggs')\n"
+  },
+  {"hello world, not multiline\n"}
+};
+
+struct record_check elastic_input[] = {
+  {"[some weird test] IndexNotFoundException[no such index]\n"},
+  {"    at org.elasticsearch.cluster.metadata.IndexNameExpressionResolver....\n"},
+  {"    at org.elasticsearch.cluster.metadata.IndexNameExpressionResolver.java:133)\n"},
+  {"    at org.elasticsearch.action.admin.indices.delete.java:75)\n"},
+  {"another separate log line\n"}
+};
+
+struct record_check elastic_output[] = {
+  {
+      "[some weird test] IndexNotFoundException[no such index]\n"
+      "    at org.elasticsearch.cluster.metadata.IndexNameExpressionResolver....\n"
+      "    at org.elasticsearch.cluster.metadata.IndexNameExpressionResolver.java:133)\n"
+      "    at org.elasticsearch.action.admin.indices.delete.java:75)\n"
+  },
+  {
+      "another separate log line\n"
+  }
+};
+
+struct record_check go_input[] = {
+    {"panic: my panic\n"},
+    {"\n"},
+    {"goroutine 4 [running]:\n"},
+    {"panic(0x45cb40, 0x47ad70)\n"},
+    {"	/usr/local/go/src/runtime/panic.go:542 +0x46c fp=0xc42003f7b8 sp=0xc42003f710 pc=0x422f7c\n"},
+    {"main.main.func1(0xc420024120)\n"},
+    {"	foo.go:6 +0x39 fp=0xc42003f7d8 sp=0xc42003f7b8 pc=0x451339\n"},
+    {"runtime.goexit()\n"},
+    {"	/usr/local/go/src/runtime/asm_amd64.s:2337 +0x1 fp=0xc42003f7e0 sp=0xc42003f7d8 pc=0x44b4d1\n"},
+    {"created by main.main\n"},
+    {"	foo.go:5 +0x58\n"},
+    {"\n"},
+    {"goroutine 1 [chan receive]:\n"},
+    {"runtime.gopark(0x4739b8, 0xc420024178, 0x46fcd7, 0xc, 0xc420028e17, 0x3)\n"},
+    {"	/usr/local/go/src/runtime/proc.go:280 +0x12c fp=0xc420053e30 sp=0xc420053e00 pc=0x42503c\n"},
+    {"runtime.goparkunlock(0xc420024178, 0x46fcd7, 0xc, 0x1000f010040c217, 0x3)\n"},
+    {"	/usr/local/go/src/runtime/proc.go:286 +0x5e fp=0xc420053e70 sp=0xc420053e30 pc=0x42512e\n"},
+    {"runtime.chanrecv(0xc420024120, 0x0, 0xc420053f01, 0x4512d8)\n"},
+    {"	/usr/local/go/src/runtime/chan.go:506 +0x304 fp=0xc420053f20 sp=0xc420053e70 pc=0x4046b4\n"},
+    {"runtime.chanrecv1(0xc420024120, 0x0)\n"},
+    {"	/usr/local/go/src/runtime/chan.go:388 +0x2b fp=0xc420053f50 sp=0xc420053f20 pc=0x40439b\n"},
+    {"main.main()\n"},
+    {"	foo.go:9 +0x6f fp=0xc420053f80 sp=0xc420053f50 pc=0x4512ef\n"},
+    {"runtime.main()\n"},
+    {"	/usr/local/go/src/runtime/proc.go:185 +0x20d fp=0xc420053fe0 sp=0xc420053f80 pc=0x424bad\n"},
+    {"runtime.goexit()\n"},
+    {"	/usr/local/go/src/runtime/asm_amd64.s:2337 +0x1 fp=0xc420053fe8 sp=0xc420053fe0 pc=0x44b4d1\n"},
+    {"\n"},
+    {"goroutine 2 [force gc (idle)]:\n"},
+    {"runtime.gopark(0x4739b8, 0x4ad720, 0x47001e, 0xf, 0x14, 0x1)\n"},
+    {"	/usr/local/go/src/runtime/proc.go:280 +0x12c fp=0xc42003e768 sp=0xc42003e738 pc=0x42503c\n"},
+    {"runtime.goparkunlock(0x4ad720, 0x47001e, 0xf, 0xc420000114, 0x1)\n"},
+    {"	/usr/local/go/src/runtime/proc.go:286 +0x5e fp=0xc42003e7a8 sp=0xc42003e768 pc=0x42512e\n"},
+    {"runtime.forcegchelper()\n"},
+    {"	/usr/local/go/src/runtime/proc.go:238 +0xcc fp=0xc42003e7e0 sp=0xc42003e7a8 pc=0x424e5c\n"},
+    {"runtime.goexit()\n"},
+    {"	/usr/local/go/src/runtime/asm_amd64.s:2337 +0x1 fp=0xc42003e7e8 sp=0xc42003e7e0 pc=0x44b4d1\n"},
+    {"created by runtime.init.4\n"},
+    {"	/usr/local/go/src/runtime/proc.go:227 +0x35\n"},
+    {"\n"},
+    {"goroutine 3 [GC sweep wait]:\n"},
+    {"runtime.gopark(0x4739b8, 0x4ad7e0, 0x46fdd2, 0xd, 0x419914, 0x1)\n"},
+    {"	/usr/local/go/src/runtime/proc.go:280 +0x12c fp=0xc42003ef60 sp=0xc42003ef30 pc=0x42503c\n"},
+    {"runtime.goparkunlock(0x4ad7e0, 0x46fdd2, 0xd, 0x14, 0x1)\n"},
+    {"	/usr/local/go/src/runtime/proc.go:286 +0x5e fp=0xc42003efa0 sp=0xc42003ef60 pc=0x42512e\n"},
+    {"runtime.bgsweep(0xc42001e150)\n"},
+    {"	/usr/local/go/src/runtime/mgcsweep.go:52 +0xa3 fp=0xc42003efd8 sp=0xc42003efa0 pc=0x419973\n"},
+    {"runtime.goexit()\n"},
+    {"	/usr/local/go/src/runtime/asm_amd64.s:2337 +0x1 fp=0xc42003efe0 sp=0xc42003efd8 pc=0x44b4d1\n"},
+    {"created by runtime.gcenable\n"},
+    {"	/usr/local/go/src/runtime/mgc.go:216 +0x58\n"},
+    {"one more line, no multiline"}
+};
+
+struct record_check go_output[] = {
+    {
+        "panic: my panic\n"
+        "\n"
+        "goroutine 4 [running]:\n"
+        "panic(0x45cb40, 0x47ad70)\n"
+        "	/usr/local/go/src/runtime/panic.go:542 +0x46c fp=0xc42003f7b8 sp=0xc42003f710 pc=0x422f7c\n"
+        "main.main.func1(0xc420024120)\n"
+        "	foo.go:6 +0x39 fp=0xc42003f7d8 sp=0xc42003f7b8 pc=0x451339\n"
+        "runtime.goexit()\n"
+        "	/usr/local/go/src/runtime/asm_amd64.s:2337 +0x1 fp=0xc42003f7e0 sp=0xc42003f7d8 pc=0x44b4d1\n"
+        "created by main.main\n"
+        "	foo.go:5 +0x58\n"
+        "\n"
+        "goroutine 1 [chan receive]:\n"
+        "runtime.gopark(0x4739b8, 0xc420024178, 0x46fcd7, 0xc, 0xc420028e17, 0x3)\n"
+        "	/usr/local/go/src/runtime/proc.go:280 +0x12c fp=0xc420053e30 sp=0xc420053e00 pc=0x42503c\n"
+        "runtime.goparkunlock(0xc420024178, 0x46fcd7, 0xc, 0x1000f010040c217, 0x3)\n"
+        "	/usr/local/go/src/runtime/proc.go:286 +0x5e fp=0xc420053e70 sp=0xc420053e30 pc=0x42512e\n"
+        "runtime.chanrecv(0xc420024120, 0x0, 0xc420053f01, 0x4512d8)\n"
+        "	/usr/local/go/src/runtime/chan.go:506 +0x304 fp=0xc420053f20 sp=0xc420053e70 pc=0x4046b4\n"
+        "runtime.chanrecv1(0xc420024120, 0x0)\n"
+        "	/usr/local/go/src/runtime/chan.go:388 +0x2b fp=0xc420053f50 sp=0xc420053f20 pc=0x40439b\n"
+        "main.main()\n"
+        "	foo.go:9 +0x6f fp=0xc420053f80 sp=0xc420053f50 pc=0x4512ef\n"
+        "runtime.main()\n"
+        "	/usr/local/go/src/runtime/proc.go:185 +0x20d fp=0xc420053fe0 sp=0xc420053f80 pc=0x424bad\n"
+        "runtime.goexit()\n"
+        "	/usr/local/go/src/runtime/asm_amd64.s:2337 +0x1 fp=0xc420053fe8 sp=0xc420053fe0 pc=0x44b4d1\n"
+        "\n"
+        "goroutine 2 [force gc (idle)]:\n"
+        "runtime.gopark(0x4739b8, 0x4ad720, 0x47001e, 0xf, 0x14, 0x1)\n"
+        "	/usr/local/go/src/runtime/proc.go:280 +0x12c fp=0xc42003e768 sp=0xc42003e738 pc=0x42503c\n"
+        "runtime.goparkunlock(0x4ad720, 0x47001e, 0xf, 0xc420000114, 0x1)\n"
+        "	/usr/local/go/src/runtime/proc.go:286 +0x5e fp=0xc42003e7a8 sp=0xc42003e768 pc=0x42512e\n"
+        "runtime.forcegchelper()\n"
+        "	/usr/local/go/src/runtime/proc.go:238 +0xcc fp=0xc42003e7e0 sp=0xc42003e7a8 pc=0x424e5c\n"
+        "runtime.goexit()\n"
+        "	/usr/local/go/src/runtime/asm_amd64.s:2337 +0x1 fp=0xc42003e7e8 sp=0xc42003e7e0 pc=0x44b4d1\n"
+        "created by runtime.init.4\n"
+        "	/usr/local/go/src/runtime/proc.go:227 +0x35\n"
+        "\n"
+        "goroutine 3 [GC sweep wait]:\n"
+        "runtime.gopark(0x4739b8, 0x4ad7e0, 0x46fdd2, 0xd, 0x419914, 0x1)\n"
+        "	/usr/local/go/src/runtime/proc.go:280 +0x12c fp=0xc42003ef60 sp=0xc42003ef30 pc=0x42503c\n"
+        "runtime.goparkunlock(0x4ad7e0, 0x46fdd2, 0xd, 0x14, 0x1)\n"
+        "	/usr/local/go/src/runtime/proc.go:286 +0x5e fp=0xc42003efa0 sp=0xc42003ef60 pc=0x42512e\n"
+        "runtime.bgsweep(0xc42001e150)\n"
+        "	/usr/local/go/src/runtime/mgcsweep.go:52 +0xa3 fp=0xc42003efd8 sp=0xc42003efa0 pc=0x419973\n"
+        "runtime.goexit()\n"
+        "	/usr/local/go/src/runtime/asm_amd64.s:2337 +0x1 fp=0xc42003efe0 sp=0xc42003efd8 pc=0x44b4d1\n"
+        "created by runtime.gcenable\n"
+        "	/usr/local/go/src/runtime/mgc.go:216 +0x58\n"
+    },
+    {"one more line, no multiline"}
+};
+
+/*
+ * Flush callback is invoked every time a multiline stream has completed a multiline
+ * message or a message is not multiline.
+ */
+static int flush_callback(struct flb_ml *ml,
+                          struct flb_ml_stream *mst,
+                          void *data, void *buf_data, size_t buf_size)
+{
+    int i;
+    int ret;
+    int len;
+    int found = FLB_FALSE;
+    size_t off = 0;
+    msgpack_unpacked result;
+    msgpack_object *map;
+    msgpack_object key;
+    msgpack_object val;
+    struct flb_time tm;
+    struct expected_result *res = data;
+    struct record_check *exp;
+
+    fprintf(stdout, "\n%s----- MULTILINE FLUSH -----%s\n", ANSI_YELLOW, ANSI_RESET);
+
+    /* Print incoming flush buffer */
+    flb_pack_print(buf_data, buf_size);
+
+    fprintf(stdout, "%s----------- EOF -----------%s\n",
+            ANSI_YELLOW, ANSI_RESET);
+
+    /* Validate content */
+    msgpack_unpacked_init(&result);
+
+    off = 0;
+    ret = msgpack_unpack_next(&result, buf_data, buf_size, &off);
+    TEST_CHECK(ret == MSGPACK_UNPACK_SUCCESS);
+
+    flb_time_pop_from_msgpack(&tm, &result, &map);
+
+    exp = &res->out_records[res->current_record];
+    len = strlen(res->key);
+    for (i = 0; i < map->via.map.size; i++) {
+        key = map->via.map.ptr[i].key;
+        val = map->via.map.ptr[i].val;
+
+        if (key.via.str.size != len) {
+            continue;
+        }
+
+        if (strncmp(key.via.str.ptr, res->key, len) == 0) {
+            found = FLB_TRUE;
+            break;
+        }
+    }
+    TEST_CHECK(found == FLB_TRUE);
+
+    len = strlen(exp->buf);
+    TEST_CHECK(val.via.str.size == len);
+    if (val.via.str.size != len) {
+        printf("expected length: %i, received: %i\n", len, val.via.str.size);
+    }
+    TEST_CHECK(memcmp(val.via.str.ptr, exp->buf, len) == 0);
+    res->current_record++;
+
+    msgpack_unpacked_destroy(&result);
+    return 0;
+}
+
+static void test_mode_docker()
+{
+    int i;
+    int len;
+    int entries;
+    struct record_check *r;
+    struct flb_config *config;
+    struct flb_time tm;
+    struct flb_ml *ml;
+    struct flb_ml_stream *mst;
+    struct expected_result res = {0};
+
+    /* Expected results context */
+    res.key = "log";
+    res.out_records = docker_output;
+
+    /*
+     * Retrieve a docker parser. We test the multiline mode in two separate ways:
+     *
+     * 1. packed JSON as Fluent Bit msgpack format
+     * 2. raw text with a parser context
+     */
+    config = flb_config_init();
+
+    /* Create docker multiline mode */
+    ml = flb_ml_mode_docker(config, 1000);
+    TEST_CHECK(ml != NULL);
+
+    mst = flb_ml_stream_create(ml, "docker", flush_callback, (void *) &res);
+    TEST_CHECK(mst != NULL);
+
+    entries = sizeof(docker_input) / sizeof(struct record_check);
+    for (i = 0; i < entries; i++) {
+        r = &docker_input[i];
+        len = strlen(r->buf);
+
+        flb_time_get(&tm);
+
+        /* Package as msgpack */
+        flb_ml_append(ml, mst, FLB_ML_TYPE_TEXT, &tm, r->buf, len);
+    }
+
+    if (ml) {
+        flb_ml_destroy(ml);
+    }
+
+    flb_config_exit(config);
+}
+
+static void test_mode_cri()
+{
+    int i;
+    int len;
+    int entries;
+    struct record_check *r;
+    struct flb_config *config;
+    struct flb_time tm;
+    struct flb_ml *ml;
+    struct flb_ml_stream *mst;
+    struct expected_result res = {0};
+
+    /* Expected results context */
+    res.key = "log";
+    res.out_records = cri_output;
+
+    /*
+     * Retrieve a docker parser. We test the multiline mode in two separate ways:
+     *
+     * 1. packed JSON as Fluent Bit msgpack format
+     * 2. raw text with a parser context
+     */
+    config = flb_config_init();
+
+    /* Create docker multiline mode */
+    ml = flb_ml_mode_cri(config, 1000);
+    TEST_CHECK(ml != NULL);
+
+    mst = flb_ml_stream_create(ml, "cri", flush_callback, (void *) &res);
+    TEST_CHECK(mst != NULL);
+
+    entries = sizeof(cri_input) / sizeof(struct record_check);
+    for (i = 0; i < entries; i++) {
+        r = &cri_input[i];
+        len = strlen(r->buf);
+        flb_time_get(&tm);
+
+        /* Package as msgpack */
+        flb_ml_append(ml, mst, FLB_ML_TYPE_TEXT, &tm, r->buf, len);
+    }
+
+    if (ml) {
+        flb_ml_destroy(ml);
+    }
+
+    flb_config_exit(config);
+}
+
+static void test_mode_java()
+{
+    int i;
+    int len;
+    int entries;
+    struct record_check *r;
+    struct flb_config *config;
+    struct flb_time tm;
+    struct flb_ml *ml;
+    struct flb_ml_stream *mst;
+    struct expected_result res = {0};
+
+    /* Expected results context */
+    res.key = "log";
+    res.out_records = java_output;
+
+    config = flb_config_init();
+
+    /* Create docker multiline mode */
+    ml = flb_ml_mode_java(config, 1000, NULL);
+    TEST_CHECK(ml != NULL);
+
+    mst = flb_ml_stream_create(ml, "java", flush_callback, (void *) &res);
+    TEST_CHECK(mst != NULL);
+
+
+    entries = sizeof(java_input) / sizeof(struct record_check);
+    for (i = 0; i < entries; i++) {
+        r = &java_input[i];
+        len = strlen(r->buf);
+
+        /* Package as msgpack */
+        flb_time_get(&tm);
+        flb_ml_append(ml, mst, FLB_ML_TYPE_TEXT, &tm, r->buf, len);
+    }
+
+    if (ml) {
+        flb_ml_destroy(ml);
+    }
+
+    flb_config_exit(config);
+}
+
+static void test_mode_python()
+{
+    int i;
+    int len;
+    int ret;
+    int entries;
+    size_t off = 0;
+    msgpack_packer mp_pck;
+    msgpack_sbuffer mp_sbuf;
+    msgpack_unpacked result;
+    msgpack_object root;
+    msgpack_object *map;
+    struct record_check *r;
+    struct flb_config *config;
+    struct flb_time tm;
+    struct flb_ml *ml;
+    struct flb_ml_stream *mst;
+    struct expected_result res = {0};
+
+    /* Expected results context */
+    res.key = "log";
+    res.out_records = python_output;
+
+    /* initialize buffers */
+    msgpack_sbuffer_init(&mp_sbuf);
+    msgpack_packer_init(&mp_pck, &mp_sbuf, msgpack_sbuffer_write);
+
+    /*
+     * Retrieve a docker parser. We test the multiline mode in two separate ways:
+     *
+     * 1. packed JSON as Fluent Bit msgpack format
+     * 2. raw text with a parser context
+     */
+    config = flb_config_init();
+
+    /* Create docker multiline mode */
+    ml = flb_ml_mode_python(config, 1000, "log");
+    TEST_CHECK(ml != NULL);
+
+    mst = flb_ml_stream_create(ml, "python", flush_callback, (void *) &res);
+    TEST_CHECK(mst != NULL);
+
+    flb_time_get(&tm);
+
+    printf("\n");
+    entries = sizeof(python_input) / sizeof(struct record_check);
+    for (i = 0; i < entries; i++) {
+        r = &python_input[i];
+        len = strlen(r->buf);
+
+        msgpack_sbuffer_init(&mp_sbuf);
+        msgpack_packer_init(&mp_pck, &mp_sbuf, msgpack_sbuffer_write);
+
+        /* Package raw text as a msgpack record */
+        msgpack_pack_array(&mp_pck, 2);
+        flb_time_append_to_msgpack(&tm, &mp_pck, 0);
+
+        msgpack_pack_map(&mp_pck, 1);
+        msgpack_pack_str(&mp_pck, 3);
+        msgpack_pack_str_body(&mp_pck, "log", 3);
+        msgpack_pack_str(&mp_pck, len);
+        msgpack_pack_str_body(&mp_pck, r->buf, len);
+
+        /* Unpack and lookup the content map */
+        msgpack_unpacked_init(&result);
+        off = 0;
+        ret = msgpack_unpack_next(&result, mp_sbuf.data, mp_sbuf.size, &off);
+        TEST_CHECK(ret == MSGPACK_UNPACK_SUCCESS);
+
+        root = result.data;
+        map = &root.via.array.ptr[1];
+
+        /* Package as msgpack */
+        flb_ml_append_object(ml, mst, &tm, map);
+        msgpack_unpacked_destroy(&result);
+        msgpack_sbuffer_destroy(&mp_sbuf);
+    }
+
+    if (ml) {
+        flb_ml_destroy(ml);
+    }
+
+    flb_config_exit(config);
+}
+
+static void test_mode_elastic()
+{
+    int i;
+    int len;
+    int ret;
+    int entries;
+    size_t off = 0;
+    msgpack_packer mp_pck;
+    msgpack_sbuffer mp_sbuf;
+    msgpack_unpacked result;
+    msgpack_object root;
+    msgpack_object *map;
+    struct record_check *r;
+    struct flb_config *config;
+    struct flb_time tm;
+    struct flb_ml *ml;
+    struct flb_ml_stream *mst;
+    struct expected_result res = {0};
+
+    /* Expected results context */
+    res.key = "log";
+    res.out_records = elastic_output;
+
+    /*
+     * Retrieve a docker parser. We test the multiline mode in two separate ways:
+     *
+     * 1. packed JSON as Fluent Bit msgpack format
+     * 2. raw text with a parser context
+     */
+    config = flb_config_init();
+
+    ml = flb_ml_create(config,
+                       FLB_ML_REGEX,         /* type      */
+                       NULL,                 /* match_str */
+                       FLB_FALSE,            /* negate    */
+                       1000,                 /* flush_ms  */
+                       "log",                /* key_content */
+                       NULL,                 /* key_pattern */
+                       NULL);                /* parser */
+
+    ret = flb_ml_rule_create(ml, "start_state", "/^\\[/", "elastic_cont", NULL);
+    if (ret != 0) {
+        fprintf(stderr, "error creating rule 1");
+    }
+
+    ret = flb_ml_rule_create(ml, "elastic_cont", "/^\\s+/", "elastic_cont", NULL);
+    if (ret != 0) {
+        fprintf(stderr, "error creating rule 2");
+    }
+
+    mst = flb_ml_stream_create(ml, "elastic", flush_callback, (void *) &res);
+    TEST_CHECK(mst != NULL);
+
+    ret = flb_ml_init(ml);
+    if (ret != 0) {
+        fprintf(stderr, "error initializing multiline\n");
+        flb_ml_destroy(ml);
+        return;
+    }
+
+    printf("\n");
+    entries = sizeof(elastic_input) / sizeof(struct record_check);
+    for (i = 0; i < entries; i++) {
+        r = &elastic_input[i];
+        len = strlen(r->buf);
+
+        /* initialize buffers */
+        msgpack_sbuffer_init(&mp_sbuf);
+        msgpack_packer_init(&mp_pck, &mp_sbuf, msgpack_sbuffer_write);
+
+        /* Package raw text as a msgpack record */
+        msgpack_pack_array(&mp_pck, 2);
+
+        flb_time_get(&tm);
+        flb_time_append_to_msgpack(&tm, &mp_pck, 0);
+
+        msgpack_pack_map(&mp_pck, 1);
+        msgpack_pack_str(&mp_pck, 3);
+        msgpack_pack_str_body(&mp_pck, "log", 3);
+        msgpack_pack_str(&mp_pck, len);
+        msgpack_pack_str_body(&mp_pck, r->buf, len);
+
+        /* Unpack and lookup the content map */
+        msgpack_unpacked_init(&result);
+        off = 0;
+        ret = msgpack_unpack_next(&result, mp_sbuf.data, mp_sbuf.size, &off);
+        TEST_CHECK(ret == MSGPACK_UNPACK_SUCCESS);
+
+        root = result.data;
+        map = &root.via.array.ptr[1];
+
+        /* Package as msgpack */
+        ret = flb_ml_append_object(ml, mst, &tm, map);
+        msgpack_unpacked_destroy(&result);
+        msgpack_sbuffer_destroy(&mp_sbuf);
+    }
+
+    if (ml) {
+        flb_ml_destroy(ml);
+    }
+
+    flb_config_exit(config);
+}
+
+static void test_mode_go()
+{
+    int i;
+    int len;
+    int entries;
+    struct record_check *r;
+    struct flb_config *config;
+    struct flb_time tm;
+    struct flb_ml *ml;
+    struct flb_ml_stream *mst;
+    struct expected_result res = {0};
+
+    /* Expected results context */
+    res.key = "log";
+    res.out_records = go_output;
+
+    config = flb_config_init();
+
+    /* Create docker multiline mode */
+    ml = flb_ml_mode_go(config, 1000, NULL);
+    TEST_CHECK(ml != NULL);
+
+    mst = flb_ml_stream_create(ml, "go", flush_callback, (void *) &res);
+    TEST_CHECK(mst != NULL);
+
+    entries = sizeof(go_input) / sizeof(struct record_check);
+    for (i = 0; i < entries; i++) {
+        r = &go_input[i];
+        len = strlen(r->buf);
+
+        /* Package as msgpack */
+        flb_time_get(&tm);
+        flb_ml_append(ml, mst, FLB_ML_TYPE_TEXT, &tm, r->buf, len);
+    }
+
+    if (ml) {
+        flb_ml_destroy(ml);
+    }
+
+    flb_config_exit(config);
+}
+
+TEST_LIST = {
+    { "mode_docker",  test_mode_docker},
+    { "mode_cri",     test_mode_cri},
+    { "mode_java",    test_mode_java},
+    { "mode_python",  test_mode_python},
+    { "mode_elastic", test_mode_elastic},
+    { "mode_go",      test_mode_go},
+    { 0 }
+};