diff --git a/lib/inc/sai_redis.h b/lib/inc/sai_redis.h
index bee7472008..bf9c60459e 100644
--- a/lib/inc/sai_redis.h
+++ b/lib/inc/sai_redis.h
@@ -45,6 +45,7 @@ void check_notifications_pointers(
 // there is something wrong and we should fail
 #define GET_RESPONSE_TIMEOUT (6*60*1000)
 
+extern std::string getSelectResultAsString(int result);
 extern void clear_local_state();
 extern void setRecording(bool record);
 extern sai_status_t setRecordingOutputDir(
diff --git a/lib/src/sai_redis_fdb.cpp b/lib/src/sai_redis_fdb.cpp
index 5c262bef31..abea6954f5 100644
--- a/lib/src/sai_redis_fdb.cpp
+++ b/lib/src/sai_redis_fdb.cpp
@@ -1,5 +1,101 @@
 #include "sai_redis.h"
 #include "meta/saiserialize.h"
+#include "meta/saiattributelist.h"
+
+sai_status_t internal_redis_flush_fdb_entries(
+        _In_ sai_object_id_t switch_id,
+        _In_ uint32_t attr_count,
+        _In_ const sai_attribute_t *attr_list)
+{
+    SWSS_LOG_ENTER();
+
+    std::vector<swss::FieldValueTuple> entry = SaiAttributeList::serialize_attr_list(
+            SAI_OBJECT_TYPE_FDB_FLUSH,
+            attr_count,
+            attr_list,
+            false);
+
+    std::string str_object_type = sai_serialize_object_type(SAI_OBJECT_TYPE_FDB_FLUSH);
+
+    std::string key = str_object_type + ":" + sai_serialize_object_id(switch_id);
+
+    SWSS_LOG_DEBUG("flush key: %s, fields: %lu", key.c_str(), entry.size());
+
+    if (g_record)
+    {
+        recordLine("f|" + key + "|" + joinFieldValues(entry));
+    }
+
+    // flush is special, it will not put data
+    // into asic view, only to message queue
+    g_asicState->set(key, entry, "flush");
+
+    // wait for response
+
+    swss::Select s;
+
+    // get consumer will be reused for flush
+
+    s.addSelectable(g_redisGetConsumer.get());
+
+    while (true)
+    {
+        SWSS_LOG_DEBUG("wait for response");
+
+        swss::Selectable *sel;
+
+        int fd;
+
+        int result = s.select(&sel, &fd, GET_RESPONSE_TIMEOUT);
+
+        if (result == swss::Select::OBJECT)
+        {
+            swss::KeyOpFieldsValuesTuple kco;
+
+            g_redisGetConsumer->pop(kco);
+
+            const std::string &op = kfvOp(kco);
+            const std::string &opkey = kfvKey(kco);
+
+            SWSS_LOG_DEBUG("response: op = %s, key = %s", opkey.c_str(), op.c_str());
+
+            if (op != "flushresponse") // ignore non response messages
+            {
+                continue;
+            }
+
+            std::string str_sai_status = opkey;
+
+            sai_status_t status;
+
+            sai_deserialize_status(str_sai_status, status);
+
+            if (g_record)
+            {
+                const std::string &str_status = kfvKey(kco);
+
+                // first serialized is status
+                recordLine("F|" + str_status);
+            }
+
+            SWSS_LOG_DEBUG("flush status: %d", status);
+
+            return status;
+        }
+
+        SWSS_LOG_ERROR("flush failed due to SELECT operation result: %s", getSelectResultAsString(result).c_str());
+        break;
+    }
+
+    if (g_record)
+    {
+        recordLine("F|SAI_STATUS_FAILURE");
+    }
+
+    SWSS_LOG_ERROR("flush failed to get response");
+
+    return SAI_STATUS_FAILURE;
+}
 
 sai_status_t redis_flush_fdb_entries(
         _In_ sai_object_id_t switch_id,
@@ -10,7 +106,11 @@ sai_status_t redis_flush_fdb_entries(
 
     SWSS_LOG_ENTER();
 
-    return SAI_STATUS_NOT_IMPLEMENTED;
+    return meta_sai_flush_fdb_entries(
+            switch_id,
+            attr_count,
+            attr_list,
+            internal_redis_flush_fdb_entries);
 }
 
 REDIS_GENERIC_QUAD_ENTRY(FDB_ENTRY,fdb_entry);
diff --git a/meta/sai_meta.cpp b/meta/sai_meta.cpp
index 6d25970ddc..3e0fce1a98 100644
--- a/meta/sai_meta.cpp
+++ b/meta/sai_meta.cpp
@@ -4737,3 +4737,154 @@ void meta_sai_on_fdb_event(
         meta_sai_on_fdb_event_single(data[i]);
     }
 }
+
+// FDB FLUSH
+
+sai_status_t meta_sai_flush_fdb_entries(
+        _In_ sai_object_id_t switch_id,
+        _In_ uint32_t attr_count,
+        _In_ const sai_attribute_t *attr_list,
+        _In_ sai_flush_fdb_entries_fn flush_fdb_entries)
+{
+    SWSS_LOG_ENTER();
+
+    if (flush_fdb_entries == NULL)
+    {
+        SWSS_LOG_ERROR("function pointer is NULL");
+
+        return SAI_STATUS_INVALID_PARAMETER;
+    }
+
+    if (attr_count > MAX_LIST_COUNT)
+    {
+        SWSS_LOG_ERROR("create attribute count is too large %u > then max list count %u", attr_count, MAX_LIST_COUNT);
+
+        return SAI_STATUS_INVALID_PARAMETER;
+    }
+
+    if (attr_count != 0 && attr_list == NULL)
+    {
+        SWSS_LOG_ERROR("attribute list is NULL");
+
+        return SAI_STATUS_INVALID_PARAMETER;
+    }
+
+    sai_object_type_t swot = sai_object_type_query(switch_id);
+
+    if (swot != SAI_OBJECT_TYPE_SWITCH)
+    {
+        SWSS_LOG_ERROR("object type for switch_id %s is %s",
+                sai_serialize_object_id(switch_id).c_str(),
+                sai_serialize_object_type(swot).c_str());
+
+        return SAI_STATUS_INVALID_PARAMETER;
+    }
+
+    if (!object_reference_exists(switch_id))
+    {
+        SWSS_LOG_ERROR("switch id %s don't exists",
+                sai_serialize_object_id(switch_id).c_str());
+
+        return SAI_STATUS_INVALID_PARAMETER;
+    }
+
+    // validate attributes
+    // - attribute list can be empty
+    // - validation is similar to "create" action but there is no
+    //   post create step and no references are updated
+    // - fdb entries are updated in fdb notification
+
+    std::unordered_map<sai_attr_id_t, const sai_attribute_t*> attrs;
+
+    SWSS_LOG_DEBUG("attr count = %u", attr_count);
+
+    for (uint32_t idx = 0; idx < attr_count; ++idx)
+    {
+        const sai_attribute_t* attr = &attr_list[idx];
+
+        auto mdp = sai_metadata_get_attr_metadata(SAI_OBJECT_TYPE_FDB_FLUSH, attr->id);
+
+        if (mdp == NULL)
+        {
+            SWSS_LOG_ERROR("unable to find attribute metadata SAI_OBJECT_TYPE_FDB_FLUSH:%d", attr->id);
+
+            return SAI_STATUS_INVALID_PARAMETER;
+        }
+
+        const sai_attribute_value_t& value = attr->value;
+
+        const sai_attr_metadata_t& md = *mdp;
+
+        META_LOG_DEBUG(md, "(fdbflush)");
+
+        if (attrs.find(attr->id) != attrs.end())
+        {
+            META_LOG_ERROR(md, "attribute id (%u) is defined on attr list multiple times", attr->id);
+
+            return SAI_STATUS_INVALID_PARAMETER;
+        }
+
+        attrs[attr->id] = attr;
+
+        if (md.flags != SAI_ATTR_FLAGS_CREATE_ONLY)
+        {
+            META_LOG_ERROR(md, "attr is expected to be marked as CREATE_ONLY");
+
+            return SAI_STATUS_INVALID_PARAMETER;
+        }
+
+        if (md.isconditional || md.validonlylength > 0)
+        {
+            META_LOG_ERROR(md, "attr should not be conditional or validonly");
+
+            return SAI_STATUS_INVALID_PARAMETER;
+        }
+
+        switch (md.attrvaluetype)
+        {
+            case SAI_ATTR_VALUE_TYPE_UINT16:
+
+                if (md.isvlan && (value.u16 >= 0xFFF || value.u16 == 0))
+                {
+                    META_LOG_ERROR(md, "is vlan id but has invalid id %u", value.u16);
+
+                    return SAI_STATUS_INVALID_PARAMETER;
+                }
+
+                break;
+
+            case SAI_ATTR_VALUE_TYPE_INT32:
+
+                if (md.isenum && !sai_metadata_is_allowed_enum_value(&md, value.s32))
+                {
+                    META_LOG_ERROR(md, "is enum, but value %d not found on allowed values list", value.s32);
+
+                    return SAI_STATUS_INVALID_PARAMETER;
+                }
+
+                break;
+
+            case SAI_ATTR_VALUE_TYPE_OBJECT_ID:
+
+                {
+                    sai_status_t status = meta_generic_validation_objlist(md, switch_id, 1, &value.oid);
+
+                    if (status != SAI_STATUS_SUCCESS)
+                    {
+                        return status;
+                    }
+
+                    break;
+                }
+
+            default:
+
+                META_LOG_THROW(md, "serialization type is not supported yet FIXME");
+        }
+    }
+
+    // there are no mandatory attributes
+    // there are no conditional attributes
+
+    return flush_fdb_entries(switch_id, attr_count, attr_list);
+}
diff --git a/meta/sai_meta.h b/meta/sai_meta.h
index 029558f3a0..0e5ce711ff 100644
--- a/meta/sai_meta.h
+++ b/meta/sai_meta.h
@@ -158,6 +158,14 @@ extern void meta_sai_on_fdb_event(
         _In_ uint32_t count,
         _In_ sai_fdb_event_notification_data_t *data);
 
+// FDB FLUSH
+
+extern sai_status_t meta_sai_flush_fdb_entries(
+        _In_ sai_object_id_t switch_id,
+        _In_ uint32_t attr_count,
+        _In_ const sai_attribute_t *attr_list,
+        _In_ sai_flush_fdb_entries_fn flush_fdb_entries);
+
 // UNIT TESTS HELPERS
 
 /**
diff --git a/syncd/syncd.cpp b/syncd/syncd.cpp
index 1ffea12efd..5714cba9df 100644
--- a/syncd/syncd.cpp
+++ b/syncd/syncd.cpp
@@ -2259,6 +2259,49 @@ sai_status_t processBulkEvent(
     return status;
 }
 
+sai_status_t processFdbFlush(
+        _In_ const swss::KeyOpFieldsValuesTuple &kco)
+{
+    SWSS_LOG_ENTER();
+
+    const std::string &key = kfvKey(kco);
+    const std::string &str_object_type = key.substr(0, key.find(":"));
+    const std::string &str_object_id = key.substr(key.find(":") + 1);
+
+    sai_object_id_t switch_vid;
+
+    sai_deserialize_object_id(str_object_id, switch_vid);
+
+    sai_object_id_t switch_rid = translate_vid_to_rid(switch_vid);
+
+    const std::vector<swss::FieldValueTuple> &values = kfvFieldsValues(kco);
+
+    for (const auto &v: values)
+    {
+        SWSS_LOG_DEBUG("attr: %s: %s", fvField(v).c_str(), fvValue(v).c_str());
+    }
+
+    SaiAttributeList list(SAI_OBJECT_TYPE_FDB_FLUSH, values, false);
+
+    /*
+     * Attribute list can't be const since we will use it to translate VID to
+     * RID inplace.
+     */
+
+    sai_attribute_t *attr_list = list.get_attr_list();
+    uint32_t attr_count = list.get_attr_count();
+
+    translate_vid_to_rid_list(SAI_OBJECT_TYPE_FDB_FLUSH, attr_count, attr_list);
+
+    sai_status_t status = sai_metadata_sai_fdb_api->flush_fdb_entries(switch_rid, attr_count, attr_list);
+
+    std::vector<swss::FieldValueTuple> en;
+
+    getResponse->set(sai_serialize_status(status), en, "flushresponse");
+
+    return status;
+}
+
 sai_status_t processEvent(
         _In_ swss::ConsumerTable &consumer)
 {
@@ -2330,6 +2373,10 @@ sai_status_t processEvent(
     {
         return processGetStatsEvent(kco);
     }
+    else if (op == "flush")
+    {
+        return processFdbFlush(kco);
+    }
     else
     {
         SWSS_LOG_THROW("api '%s' is not implemented", op.c_str());
diff --git a/vslib/src/sai_vs_fdb.cpp b/vslib/src/sai_vs_fdb.cpp
index 9c7ede5bd7..5ec7edfbe3 100644
--- a/vslib/src/sai_vs_fdb.cpp
+++ b/vslib/src/sai_vs_fdb.cpp
@@ -1,6 +1,21 @@
 #include "sai_vs.h"
 #include "sai_vs_internal.h"
 
+sai_status_t internal_vs_flush_fdb_entries(
+        _In_ sai_object_id_t switch_id,
+        _In_ uint32_t attr_count,
+        _In_ const sai_attribute_t *attr_list)
+{
+    SWSS_LOG_ENTER();
+
+    // TODO implement actual flush (9 cases) with ntf generation (queue)
+    // also update meta db here
+
+    SWSS_LOG_ERROR("not implemented");
+
+    return SAI_STATUS_NOT_IMPLEMENTED;
+}
+
 sai_status_t vs_flush_fdb_entries(
         _In_ sai_object_id_t switch_id,
         _In_ uint32_t attr_count,
@@ -10,7 +25,11 @@ sai_status_t vs_flush_fdb_entries(
 
     SWSS_LOG_ENTER();
 
-    return SAI_STATUS_NOT_IMPLEMENTED;
+    return meta_sai_flush_fdb_entries(
+            switch_id,
+            attr_count,
+            attr_list,
+            internal_vs_flush_fdb_entries);
 }
 
 VS_GENERIC_QUAD_ENTRY(FDB_ENTRY,fdb_entry);