diff --git a/orchagent/Makefile.am b/orchagent/Makefile.am
index 429d82e5fe7e..79cfe99f6d58 100644
--- a/orchagent/Makefile.am
+++ b/orchagent/Makefile.am
@@ -1,4 +1,4 @@
-INCLUDES = -I $(top_srcdir) -I $(top_srcdir)/warmrestart -I flex_counter
+INCLUDES = -I $(top_srcdir) -I $(top_srcdir)/warmrestart -I flex_counter -I debug_counter
 
 CFLAGS_SAI = -I /usr/include/sai
 
@@ -54,9 +54,11 @@ orchagent_SOURCES = \
             watermarkorch.cpp \
             policerorch.cpp \
             sfloworch.cpp \
-            chassisorch.cpp
+            chassisorch.cpp \
+            debugcounterorch.cpp
 
 orchagent_SOURCES += flex_counter/flex_counter_manager.cpp flex_counter/flex_counter_stat_manager.cpp
+orchagent_SOURCES += debug_counter/debug_counter.cpp debug_counter/drop_counter.cpp
 
 orchagent_CFLAGS = $(DBGFLAGS) $(AM_CFLAGS) $(CFLAGS_COMMON) $(CFLAGS_SAI)
 orchagent_CPPFLAGS = $(DBGFLAGS) $(AM_CFLAGS) $(CFLAGS_COMMON) $(CFLAGS_SAI)
diff --git a/orchagent/debug_counter/debug_counter.cpp b/orchagent/debug_counter/debug_counter.cpp
new file mode 100644
index 000000000000..8e01052e6e2a
--- /dev/null
+++ b/orchagent/debug_counter/debug_counter.cpp
@@ -0,0 +1,112 @@
+#include "debug_counter.h"
+#include "drop_counter.h"
+
+#include <unordered_set>
+#include <unordered_map>
+#include <string>
+#include <memory>
+#include <vector>
+#include "rediscommand.h"
+#include <exception>
+#include "logger.h"
+
+using std::runtime_error;
+using std::string;
+using std::unique_ptr;
+using std::unordered_map;
+using std::unordered_set;
+using std::vector;
+using swss::FieldValueTuple;
+
+extern sai_object_id_t gSwitchId;
+extern sai_debug_counter_api_t *sai_debug_counter_api;
+
+// Set of supported attributes to support easy look-up.
+const unordered_set<string> DebugCounter::supported_debug_counter_attributes =
+{
+    COUNTER_ALIAS,
+    COUNTER_TYPE,
+    COUNTER_DESCRIPTION,
+    COUNTER_GROUP
+};
+
+const std::unordered_map<std::string, sai_debug_counter_type_t> DebugCounter::debug_counter_type_lookup =
+{
+    { PORT_INGRESS_DROPS, SAI_DEBUG_COUNTER_TYPE_PORT_IN_DROP_REASONS },
+    { PORT_EGRESS_DROPS,  SAI_DEBUG_COUNTER_TYPE_PORT_OUT_DROP_REASONS },
+    { SWITCH_INGRESS_DROPS, SAI_DEBUG_COUNTER_TYPE_SWITCH_IN_DROP_REASONS },
+    { SWITCH_EGRESS_DROPS, SAI_DEBUG_COUNTER_TYPE_SWITCH_OUT_DROP_REASONS }
+};
+
+// It is expected that derived types populate any relevant fields and
+// initialize the counter in the SAI.
+//
+// If counter_type is not a member of debug_counter_type_lookup then this
+// constructor will throw a runtime error.
+DebugCounter::DebugCounter(
+        const string& counter_name,
+        const string& counter_type)
+    : name(counter_name)
+{
+    SWSS_LOG_ENTER();
+
+    auto counter_type_it = debug_counter_type_lookup.find(counter_type);
+    if (counter_type_it == debug_counter_type_lookup.end()) {
+        SWSS_LOG_ERROR("Failed to initialize debug counter of type '%s'",
+                counter_type.c_str());
+        throw runtime_error("Failed to initialize debug counter");
+    }
+    type = counter_type_it->first;
+}
+
+// It is expected that derived types delete the counter from the SAI.
+DebugCounter::~DebugCounter()
+{
+    SWSS_LOG_ENTER();
+}
+
+void DebugCounter::serializeDebugCounterType(sai_attribute_t& type_attribute)
+{
+    SWSS_LOG_ENTER();
+
+    sai_debug_counter_type_t sai_counter_type = debug_counter_type_lookup.at(type);
+    type_attribute.id = SAI_DEBUG_COUNTER_ATTR_TYPE;
+    type_attribute.value.s32 = sai_counter_type;
+
+    SWSS_LOG_DEBUG("Serializing debug counter of type '%s'", type.c_str());
+}
+
+// addDebugCounterToSAI creates a new debug counter object in the SAI given a list of debug counter attributes.
+//
+// If the SAI returns an error then this method will throw a runtime error.
+//
+// Behavior is undefined if num_attributes is not equal to the number of
+// attributes in debug_counter_attributes.
+void DebugCounter::addDebugCounterToSAI(const int num_attributes, const sai_attribute_t *debug_counter_attributes)
+{
+    SWSS_LOG_ENTER();
+
+    SWSS_LOG_DEBUG("Adding debug counter '%s' to SAI", name.c_str());
+    sai_object_id_t debug_counter_id;
+    if (sai_debug_counter_api->create_debug_counter(&debug_counter_id,
+                                                    gSwitchId,
+                                                    num_attributes,
+                                                    debug_counter_attributes) != SAI_STATUS_SUCCESS) {
+        SWSS_LOG_ERROR("Failed to create debug counter '%s'", name.c_str());
+        throw std::runtime_error("Failed to create debug counter");
+    }
+
+    SWSS_LOG_DEBUG("Created debug counter '%s' with OID=%lu", name.c_str(), debug_counter_id);
+    counter_id = debug_counter_id;
+}
+
+void DebugCounter::removeDebugCounterFromSAI()
+{
+    SWSS_LOG_ENTER();
+
+    SWSS_LOG_DEBUG("Removing debug counter '%s' from SAI", name.c_str());
+    if (sai_debug_counter_api->remove_debug_counter(counter_id) != SAI_STATUS_SUCCESS) {
+        SWSS_LOG_ERROR("Failed to remove debug counter '%s'", name.c_str());
+        throw std::runtime_error("Failed to remove debug counter");
+    }
+}
diff --git a/orchagent/debug_counter/debug_counter.h b/orchagent/debug_counter/debug_counter.h
new file mode 100644
index 000000000000..27dec20920d5
--- /dev/null
+++ b/orchagent/debug_counter/debug_counter.h
@@ -0,0 +1,64 @@
+#ifndef SWSS_UTIL_DEBUG_COUNTER_H_
+#define SWSS_UTIL_DEBUG_COUNTER_H_
+
+#include <string>
+#include <unordered_map>
+#include <unordered_set>
+
+extern "C" {
+#include "sai.h"
+}
+
+// Supported debug counter attributes.
+#define COUNTER_ALIAS       "alias"
+#define COUNTER_TYPE        "type"
+#define COUNTER_DESCRIPTION "desc"
+#define COUNTER_GROUP       "group"
+
+// Supported debug counter types.
+#define PORT_INGRESS_DROPS   "PORT_INGRESS_DROPS"
+#define PORT_EGRESS_DROPS    "PORT_EGRESS_DROPS"
+#define SWITCH_INGRESS_DROPS "SWITCH_INGRESS_DROPS"
+#define SWITCH_EGRESS_DROPS  "SWITCH_EGRESS_DROPS"
+
+// DebugCounter represents a SAI debug counter object.
+class DebugCounter
+{
+    public:
+        DebugCounter(const std::string& counter_name, const std::string& counter_type) noexcept(false);
+        DebugCounter(const DebugCounter&) = delete;
+        DebugCounter& operator=(const DebugCounter&) = delete;
+        virtual ~DebugCounter();
+
+        std::string getCounterName() const { return name; }
+        std::string getCounterType() const { return type; }
+
+        virtual std::string getDebugCounterSAIStat() const noexcept(false) = 0;
+
+        static const std::unordered_set<std::string>& getSupportedDebugCounterAttributes()
+        {
+            return supported_debug_counter_attributes;
+        }
+
+        // TODO: We should try to neatly abstract this like we've done for the isValid methods in DropCounter.
+        static const std::unordered_map<std::string, sai_debug_counter_type_t>& getDebugCounterTypeLookup()
+        {
+            return debug_counter_type_lookup;
+        }
+
+    protected:
+        // These methods are intended to help with initialization. Dervied types will most likely
+        // need to define additional helper methods to serialize additional fields (see DropCounter for example).
+        void serializeDebugCounterType(sai_attribute_t& type_attribute);
+        void addDebugCounterToSAI(int num_attrs, const sai_attribute_t *counter_attrs) noexcept(false);
+        void removeDebugCounterFromSAI() noexcept(false);
+
+        std::string name;
+        std::string type;
+        sai_object_id_t counter_id = 0;
+
+        static const std::unordered_set<std::string> supported_debug_counter_attributes;
+        static const std::unordered_map<std::string, sai_debug_counter_type_t> debug_counter_type_lookup;
+};
+
+#endif // _SWSS_UTIL_DEBUG_COUNTER_H_
diff --git a/orchagent/debug_counter/drop_counter.cpp b/orchagent/debug_counter/drop_counter.cpp
new file mode 100644
index 000000000000..aab9c4b54719
--- /dev/null
+++ b/orchagent/debug_counter/drop_counter.cpp
@@ -0,0 +1,372 @@
+#include "drop_counter.h"
+
+#include "logger.h"
+#include "sai_serialize.h"
+
+using std::runtime_error;
+using std::string;
+using std::unordered_map;
+using std::unordered_set;
+using std::vector;
+
+extern sai_object_id_t gSwitchId;
+extern sai_debug_counter_api_t *sai_debug_counter_api;
+
+const unordered_map<string, sai_in_drop_reason_t> DropCounter::ingress_drop_reason_lookup =
+{
+    { L2_ANY,               SAI_IN_DROP_REASON_L2_ANY },
+    { SMAC_MULTICAST,       SAI_IN_DROP_REASON_SMAC_MULTICAST },
+    { SMAC_EQUALS_DMAC,     SAI_IN_DROP_REASON_SMAC_EQUALS_DMAC },
+    { DMAC_RESERVED,        SAI_IN_DROP_REASON_DMAC_RESERVED },
+    { VLAN_TAG_NOT_ALLOWED, SAI_IN_DROP_REASON_VLAN_TAG_NOT_ALLOWED },
+    { INGRESS_VLAN_FILTER,  SAI_IN_DROP_REASON_INGRESS_VLAN_FILTER },
+    { INGRESS_STP_FILTER,   SAI_IN_DROP_REASON_INGRESS_STP_FILTER },
+    { FDB_UC_DISCARD,       SAI_IN_DROP_REASON_FDB_UC_DISCARD },
+    { FDB_MC_DISCARD,       SAI_IN_DROP_REASON_FDB_MC_DISCARD },
+    { L2_LOOPBACK_FILTER,   SAI_IN_DROP_REASON_L2_LOOPBACK_FILTER },
+    { EXCEEDS_L2_MTU,       SAI_IN_DROP_REASON_EXCEEDS_L2_MTU },
+    { L3_ANY,               SAI_IN_DROP_REASON_L3_ANY },
+    { EXCEEDS_L3_MTU,       SAI_IN_DROP_REASON_EXCEEDS_L3_MTU },
+    { TTL,                  SAI_IN_DROP_REASON_TTL },
+    { L3_LOOPBACK_FILTER,   SAI_IN_DROP_REASON_L3_LOOPBACK_FILTER },
+    { NON_ROUTABLE,         SAI_IN_DROP_REASON_NON_ROUTABLE },
+    { NO_L3_HEADER,         SAI_IN_DROP_REASON_NO_L3_HEADER },
+    { IP_HEADER_ERROR,      SAI_IN_DROP_REASON_IP_HEADER_ERROR },
+    { UC_DIP_MC_DMAC,       SAI_IN_DROP_REASON_UC_DIP_MC_DMAC },
+    { DIP_LOOPBACK,         SAI_IN_DROP_REASON_DIP_LOOPBACK },
+    { SIP_LOOPBACK,         SAI_IN_DROP_REASON_SIP_LOOPBACK },
+    { SIP_MC,               SAI_IN_DROP_REASON_SIP_MC },
+    { SIP_CLASS_E,          SAI_IN_DROP_REASON_SIP_CLASS_E },
+    { SIP_UNSPECIFIED,      SAI_IN_DROP_REASON_SIP_UNSPECIFIED },
+    { MC_DMAC_MISMATCH,     SAI_IN_DROP_REASON_MC_DMAC_MISMATCH },
+    { SIP_EQUALS_DIP,       SAI_IN_DROP_REASON_SIP_EQUALS_DIP },
+    { SIP_BC,               SAI_IN_DROP_REASON_SIP_BC },
+    { DIP_LOCAL,            SAI_IN_DROP_REASON_DIP_LOCAL },
+    { DIP_LINK_LOCAL,       SAI_IN_DROP_REASON_DIP_LINK_LOCAL },
+    { SIP_LINK_LOCAL,       SAI_IN_DROP_REASON_SIP_LINK_LOCAL },
+    { IPV6_MC_SCOPE0,       SAI_IN_DROP_REASON_IPV6_MC_SCOPE0 },
+    { IPV6_MC_SCOPE1,       SAI_IN_DROP_REASON_IPV6_MC_SCOPE1 },
+    { IRIF_DISABLED,        SAI_IN_DROP_REASON_IRIF_DISABLED },
+    { ERIF_DISABLED,        SAI_IN_DROP_REASON_ERIF_DISABLED },
+    { LPM4_MISS,            SAI_IN_DROP_REASON_LPM4_MISS },
+    { LPM6_MISS,            SAI_IN_DROP_REASON_LPM6_MISS },
+    { BLACKHOLE_ROUTE,      SAI_IN_DROP_REASON_BLACKHOLE_ROUTE },
+    { BLACKHOLE_ARP,        SAI_IN_DROP_REASON_BLACKHOLE_ARP },
+    { UNRESOLVED_NEXT_HOP,  SAI_IN_DROP_REASON_UNRESOLVED_NEXT_HOP },
+    { L3_EGRESS_LINK_DOWN,  SAI_IN_DROP_REASON_L3_EGRESS_LINK_DOWN },
+    { DECAP_ERROR,          SAI_IN_DROP_REASON_DECAP_ERROR },
+    { ACL_ANY,              SAI_IN_DROP_REASON_ACL_ANY},
+    { ACL_INGRESS_PORT,     SAI_IN_DROP_REASON_ACL_INGRESS_PORT },
+    { ACL_INGRESS_LAG,      SAI_IN_DROP_REASON_ACL_INGRESS_LAG },
+    { ACL_INGRESS_VLAN,     SAI_IN_DROP_REASON_ACL_INGRESS_VLAN },
+    { ACL_INGRESS_RIF,      SAI_IN_DROP_REASON_ACL_INGRESS_RIF },
+    { ACL_INGRESS_SWITCH,   SAI_IN_DROP_REASON_ACL_INGRESS_SWITCH },
+    { ACL_EGRESS_PORT,      SAI_IN_DROP_REASON_ACL_EGRESS_PORT },
+    { ACL_EGRESS_LAG,       SAI_IN_DROP_REASON_ACL_EGRESS_LAG },
+    { ACL_EGRESS_VLAN,      SAI_IN_DROP_REASON_ACL_EGRESS_VLAN },
+    { ACL_EGRESS_RIF,       SAI_IN_DROP_REASON_ACL_EGRESS_RIF },
+    { ACL_EGRESS_SWITCH,    SAI_IN_DROP_REASON_ACL_EGRESS_SWITCH }
+};
+
+const unordered_map<string, sai_out_drop_reason_t> DropCounter::egress_drop_reason_lookup =
+{
+    { L2_ANY,              SAI_OUT_DROP_REASON_L2_ANY },
+    { EGRESS_VLAN_FILTER,  SAI_OUT_DROP_REASON_EGRESS_VLAN_FILTER },
+    { L3_ANY,              SAI_OUT_DROP_REASON_L3_ANY },
+    { L3_EGRESS_LINK_DOWN, SAI_OUT_DROP_REASON_L3_EGRESS_LINK_DOWN },
+};
+
+// Need to allocate enough space for the SAI to report the drop reasons, 100
+// gives us plenty of space for both ingress and egress drop reasons.
+const uint32_t maxDropReasons = 100;
+
+// If initialization fails, this constructor will throw a runtime error.
+DropCounter::DropCounter(const string& counter_name, const string& counter_type, const unordered_set<string>& drop_reasons)
+        : DebugCounter(counter_name, counter_type), drop_reasons(drop_reasons)
+{
+    SWSS_LOG_ENTER();
+    initializeDropCounterInSAI();
+}
+
+DropCounter::~DropCounter()
+{
+    SWSS_LOG_ENTER();
+    try
+    {
+        DebugCounter::removeDebugCounterFromSAI();
+    }
+    catch (const std::runtime_error& e)
+    {
+        SWSS_LOG_ERROR("Failed to remove drop counter '%s' from SAI", name.c_str());
+    }
+}
+
+// If we are unable to query the SAI or the type of counter is not supported
+// then this method throws a runtime error.
+std::string DropCounter::getDebugCounterSAIStat() const
+{
+    SWSS_LOG_ENTER();
+
+    sai_attribute_t index_attribute;
+    index_attribute.id = SAI_DEBUG_COUNTER_ATTR_INDEX;
+    if (sai_debug_counter_api->get_debug_counter_attribute(counter_id, 1, &index_attribute) != SAI_STATUS_SUCCESS)
+    {
+        SWSS_LOG_ERROR("Failed to get stat for debug counter '%s'", name.c_str());
+        throw runtime_error("Failed to get debug counter stat");
+    }
+
+    auto index = index_attribute.value.u32;
+    if (type == PORT_INGRESS_DROPS)
+    {
+        return sai_serialize_port_stat(static_cast<sai_port_stat_t>(SAI_PORT_STAT_IN_DROP_REASON_RANGE_BASE + index));
+    }
+    else if (type == PORT_EGRESS_DROPS)
+    {
+        return sai_serialize_port_stat(static_cast<sai_port_stat_t>(SAI_PORT_STAT_OUT_DROP_REASON_RANGE_BASE + index));
+    }
+    else if (type == SWITCH_INGRESS_DROPS)
+    {
+        return sai_serialize_switch_stat(static_cast<sai_switch_stat_t>(SAI_SWITCH_STAT_IN_DROP_REASON_RANGE_BASE + index));
+    }
+    else if (type == SWITCH_EGRESS_DROPS)
+    {
+        return sai_serialize_switch_stat(static_cast<sai_switch_stat_t>(SAI_SWITCH_STAT_OUT_DROP_REASON_RANGE_BASE + index));
+    }
+    else
+    {
+        SWSS_LOG_ERROR("No stat found for debug counter '%s' of type '%s'", name.c_str(), type.c_str());
+        throw runtime_error("No stat found for debug counter");
+    }
+}
+
+// If the drop reason is already present on this counter, this method has no
+// effect.
+//
+// If the update fails, this method throws a runtime error.
+void DropCounter::addDropReason(const std::string& drop_reason)
+{
+    SWSS_LOG_ENTER();
+
+    if (drop_reasons.find(drop_reason) != drop_reasons.end())
+    {
+        SWSS_LOG_DEBUG("Drop reason '%s' already present on '%s'", drop_reason.c_str(), name.c_str());
+        return;
+    }
+
+    try
+    {
+        drop_reasons.emplace(drop_reason);
+        updateDropReasonsInSAI();
+    }
+    catch (const std::runtime_error& e)
+    {
+        drop_reasons.erase(drop_reason);
+        throw;
+    }
+}
+
+// If the drop reason is not present on this counter, this method has no
+// effect.
+//
+// If the update fails, this method throws a runtime error.
+void DropCounter::removeDropReason(const std::string& drop_reason)
+{
+    SWSS_LOG_ENTER();
+
+    auto drop_reason_it = drop_reasons.find(drop_reason);
+    if (drop_reason_it == drop_reasons.end())
+    {
+        SWSS_LOG_DEBUG("Drop reason '%s' not present on '%s'", drop_reason.c_str(), name.c_str());
+        return;
+    }
+
+    try
+    {
+        drop_reasons.erase(drop_reason_it);
+        updateDropReasonsInSAI();
+    }
+    catch (const std::runtime_error& e)
+    {
+        drop_reasons.emplace(drop_reason);
+        throw e;
+    }
+}
+
+bool DropCounter::isIngressDropReasonValid(const std::string& drop_reason)
+{
+    return ingress_drop_reason_lookup.find(drop_reason) != ingress_drop_reason_lookup.end();
+}
+
+bool DropCounter::isEgressDropReasonValid(const std::string& drop_reason)
+{
+    return egress_drop_reason_lookup.find(drop_reason) != egress_drop_reason_lookup.end();
+}
+
+// If initialization fails for any reason, this method throws a runtime error.
+void DropCounter::initializeDropCounterInSAI()
+{
+    sai_attribute_t debug_counter_attributes[2];
+    vector<int32_t> drop_reason_list(drop_reasons.size());
+    DebugCounter::serializeDebugCounterType(debug_counter_attributes[0]);
+    DropCounter::serializeDropReasons(static_cast<uint32_t>(drop_reasons.size()), drop_reason_list.data(), debug_counter_attributes + 1);
+    DebugCounter::addDebugCounterToSAI(2, debug_counter_attributes);
+}
+
+// serializeDropReasons takes the list of drop reasons associated with this
+// counter and stores them in a SAI readable format in drop_reason_attribute.
+//
+// This method assumes that drop_reason_list points to a region in memory with
+// enough space for drop_reason_count drop reasons to be stored.
+//
+// If any of the provided drop reasons (or their serialization) is undefined,
+// then this method throws a runtime error.
+void DropCounter::serializeDropReasons(uint32_t drop_reason_count, int32_t *drop_reason_list, sai_attribute_t *drop_reason_attribute)
+{
+    SWSS_LOG_ENTER();
+
+    if (type == PORT_INGRESS_DROPS || type == SWITCH_INGRESS_DROPS)
+    {
+        drop_reason_attribute->id = SAI_DEBUG_COUNTER_ATTR_IN_DROP_REASON_LIST;
+        drop_reason_attribute->value.s32list.count = drop_reason_count;
+        drop_reason_attribute->value.s32list.list = drop_reason_list;
+
+        int index = 0;
+        for (auto drop_reason: drop_reasons)
+        {
+            auto reason_it = ingress_drop_reason_lookup.find(drop_reason);
+            if (reason_it == ingress_drop_reason_lookup.end())
+            {
+                SWSS_LOG_ERROR("Ingress drop reason '%s' not found", drop_reason.c_str());
+                throw runtime_error("Ingress drop reason not found");
+            }
+
+            drop_reason_list[index++] = static_cast<int32_t>(reason_it->second);
+        }
+    }
+    else if (type == PORT_EGRESS_DROPS || type == SWITCH_EGRESS_DROPS)
+    {
+            drop_reason_attribute->id = SAI_DEBUG_COUNTER_ATTR_OUT_DROP_REASON_LIST;
+            drop_reason_attribute->value.s32list.count = drop_reason_count;
+            drop_reason_attribute->value.s32list.list = drop_reason_list;
+
+            int index = 0;
+            for (auto drop_reason: drop_reasons)
+            {
+                auto reason_it = egress_drop_reason_lookup.find(drop_reason);
+                if (reason_it == egress_drop_reason_lookup.end())
+                {
+                    SWSS_LOG_ERROR("Egress drop reason '%s' not found", drop_reason.c_str());
+                    throw runtime_error("Egress drop reason not found");
+                }
+
+                drop_reason_list[index++] = static_cast<int32_t>(reason_it->second);
+            }
+    }
+    else
+    {
+        SWSS_LOG_ERROR("Serialization undefined for drop counter type '%s'", type.c_str());
+        throw runtime_error("Failed to serialize drop counter attributes");
+    }
+}
+
+// If the SAI update fails, this method throws a runtime error.
+void DropCounter::updateDropReasonsInSAI()
+{
+    SWSS_LOG_ENTER();
+
+    sai_attribute_t updated_drop_reasons;
+    vector<int32_t> drop_reason_list(drop_reasons.size());
+    serializeDropReasons(static_cast<uint32_t>(drop_reasons.size()), drop_reason_list.data(), &updated_drop_reasons);
+    if (sai_debug_counter_api->set_debug_counter_attribute(counter_id, &updated_drop_reasons) != SAI_STATUS_SUCCESS)
+    {
+        SWSS_LOG_ERROR("Could not update drop reasons for drop counter '%s'", name.c_str());
+        throw runtime_error("Could not update drop reason list");
+    }
+}
+
+// Multiple calls to this function are guaranteed to return the same reasons
+// (assuming the device has not been rebooted between calls). The order of
+// the list of reasons is not guaranteed to be the same between calls.
+//
+// If the device does not support querying drop reasons, this method will
+// return an empty list.
+vector<string> DropCounter::getSupportedDropReasons(sai_debug_counter_attr_t drop_reason_type)
+{
+    sai_s32_list_t drop_reason_list;
+    int32_t        supported_reasons[maxDropReasons];
+    drop_reason_list.count = maxDropReasons;
+    drop_reason_list.list = supported_reasons;
+
+    if (sai_query_attribute_enum_values_capability(gSwitchId,
+                                                   SAI_OBJECT_TYPE_DEBUG_COUNTER,
+                                                   drop_reason_type,
+                                                   &drop_reason_list) != SAI_STATUS_SUCCESS)
+    {
+        SWSS_LOG_NOTICE("This device does not support querying drop reasons");
+        return {};
+    }
+
+    vector<string> supported_drop_reasons;
+    for (uint32_t i = 0; i < drop_reason_list.count; i++)
+    {
+        string drop_reason;
+        if (drop_reason_type == SAI_DEBUG_COUNTER_ATTR_IN_DROP_REASON_LIST)
+        {
+            drop_reason = sai_serialize_ingress_drop_reason(static_cast<sai_in_drop_reason_t>(drop_reason_list.list[i]));
+        }
+        else
+        {
+            drop_reason = sai_serialize_egress_drop_reason(static_cast<sai_out_drop_reason_t>(drop_reason_list.list[i]));
+        }
+
+        supported_drop_reasons.push_back(drop_reason);
+    }
+
+    return supported_drop_reasons;
+}
+
+// serializeSupportedDropReasons takes a list of drop reasons and returns that
+// list as a string.
+//
+// e.g. { "SMAC_EQUALS_DMAC", "INGRESS_VLAN_FILTER" } -> "["SMAC_EQUALS_DMAC","INGRESS_VLAN_FILTER"]"
+// e.g. { } -> "[]"
+string DropCounter::serializeSupportedDropReasons(vector<string> drop_reasons)
+{
+    if (drop_reasons.size() == 0)
+    {
+        return "[]";
+    }
+
+    string supported_drop_reasons;
+    for (auto const &drop_reason : drop_reasons)
+    {
+        supported_drop_reasons += ',';
+        supported_drop_reasons += drop_reason;
+    }
+
+    supported_drop_reasons[0] = '[';
+    return supported_drop_reasons + ']';
+}
+
+// It is not guaranteed that the amount of available counters will change only
+// if counters are added or removed. Depending on the platform, debug counters
+// may share hardware resources with other ASIC objects in which case this
+// amount may change due to resource allocation in other parts of the system.
+//
+// If the device does not support querying counter availability, this method
+// will return 0.
+uint64_t DropCounter::getSupportedDebugCounterAmounts(sai_debug_counter_type_t counter_type)
+{
+    sai_attribute_t attr;
+    uint64_t count;
+
+    attr.id = SAI_DEBUG_COUNTER_ATTR_TYPE;
+    attr.value.s32 = counter_type;
+    if (sai_object_type_get_availability(gSwitchId, SAI_OBJECT_TYPE_DEBUG_COUNTER, 1, &attr, &count) != SAI_STATUS_SUCCESS)
+    {
+        SWSS_LOG_NOTICE("This device does not support querying the number of drop counters");
+        return 0;
+    }
+
+    return count;
+}
diff --git a/orchagent/debug_counter/drop_counter.h b/orchagent/debug_counter/drop_counter.h
new file mode 100644
index 000000000000..8ae34e3d4b69
--- /dev/null
+++ b/orchagent/debug_counter/drop_counter.h
@@ -0,0 +1,54 @@
+#ifndef SWSS_UTIL_DROP_COUNTER_H_
+#define SWSS_UTIL_DROP_COUNTER_H_
+
+#include <string>
+#include <vector>
+#include <unordered_set>
+#include <unordered_map>
+#include "debug_counter.h"
+#include "drop_reasons.h"
+
+extern "C" {
+#include "sai.h"
+}
+
+// DropCounter represents a SAI debug counter object that track packet drops.
+class DropCounter : public DebugCounter
+{
+    public:
+        DropCounter(const std::string& counter_name,
+                    const std::string& counter_type,
+                    const std::unordered_set<std::string>& drop_reasons) noexcept(false);
+        DropCounter(const DropCounter&) = delete;
+        DropCounter& operator=(const DropCounter&) = delete;
+        virtual ~DropCounter();
+
+        const std::unordered_set<std::string>& getDropReasons() const { return drop_reasons; }
+
+        virtual std::string getDebugCounterSAIStat() const noexcept(false);
+
+        void addDropReason(const std::string& drop_reason) noexcept(false);
+        void removeDropReason(const std::string& drop_reason) noexcept(false);
+
+        static bool isIngressDropReasonValid(const std::string& drop_reason);
+        static bool isEgressDropReasonValid(const std::string& drop_reason);
+
+        static std::vector<std::string> getSupportedDropReasons(sai_debug_counter_attr_t drop_reason_type);
+        static std::string serializeSupportedDropReasons(std::vector<std::string> drop_reasons);
+        static uint64_t getSupportedDebugCounterAmounts(sai_debug_counter_type_t counter_type);
+
+    private:
+        void initializeDropCounterInSAI() noexcept(false);
+        void serializeDropReasons(
+                uint32_t drop_reason_count,
+                int32_t *drop_reason_list,
+                sai_attribute_t *drop_reason_attribute) noexcept(false);
+        void updateDropReasonsInSAI() noexcept(false);
+
+        std::unordered_set<std::string> drop_reasons;
+
+        static const std::unordered_map<std::string, sai_in_drop_reason_t> ingress_drop_reason_lookup;
+        static const std::unordered_map<std::string, sai_out_drop_reason_t> egress_drop_reason_lookup;
+};
+
+#endif // SWSS_UTIL_DROP_COUNTER_H_
diff --git a/orchagent/debug_counter/drop_reasons.h b/orchagent/debug_counter/drop_reasons.h
new file mode 100644
index 000000000000..08265f21e4cb
--- /dev/null
+++ b/orchagent/debug_counter/drop_reasons.h
@@ -0,0 +1,65 @@
+#ifndef DROP_REASONS_H
+#define DROP_REASONS_H
+
+// L2 Drop Reasons
+#define L2_ANY               "L2_ANY"
+#define SMAC_MULTICAST       "SMAC_MULTICAST"
+#define SMAC_EQUALS_DMAC     "SMAC_EQUALS_DMAC"
+#define DMAC_RESERVED        "DMAC_RESERVED"
+#define VLAN_TAG_NOT_ALLOWED "VLAN_TAG_NOT_ALLOWED"
+#define INGRESS_VLAN_FILTER  "INGRESS_VLAN_FILTER"
+#define EGRESS_VLAN_FILTER   "EGRESS_VLAN_FILTER"
+#define INGRESS_STP_FILTER   "INGRESS_STP_FILTER"
+#define FDB_UC_DISCARD       "FDB_UC_DISCARD"
+#define FDB_MC_DISCARD       "FDB_MC_DISCARD"
+#define L2_LOOPBACK_FILTER   "L2_LOOPBACK_FILTER"
+#define EXCEEDS_L2_MTU       "EXCEEDS_L2_MTU"
+
+// L3 Drop Reasons
+#define L3_ANY              "L3_ANY"
+#define EXCEEDS_L3_MTU      "EXCEEDS_L3_MTU"
+#define TTL                 "TTL"
+#define L3_LOOPBACK_FILTER  "L3_LOOPBACK_FILTER"
+#define NON_ROUTABLE        "NON_ROUTABLE"
+#define NO_L3_HEADER        "NO_L3_HEADER"
+#define IP_HEADER_ERROR     "IP_HEADER_ERROR"
+#define UC_DIP_MC_DMAC      "UC_DIP_MC_DMAC"
+#define DIP_LOOPBACK        "DIP_LOOPBACK"
+#define SIP_LOOPBACK        "SIP_LOOPBACK"
+#define SIP_MC              "SIP_MC"
+#define SIP_CLASS_E         "SIP_CLASS_E"
+#define SIP_UNSPECIFIED     "SIP_UNSPECIFIED"
+#define MC_DMAC_MISMATCH    "MC_DMAC_MISMATCH"
+#define SIP_EQUALS_DIP      "SIP_EQUALS_DIP"
+#define SIP_BC              "SIP_BC"
+#define DIP_LOCAL           "DIP_LOCAL"
+#define DIP_LINK_LOCAL      "DIP_LINK_LOCAL"
+#define SIP_LINK_LOCAL      "SIP_LINK_LOCAL"
+#define IPV6_MC_SCOPE0      "IPV6_MC_SCOPE0"
+#define IPV6_MC_SCOPE1      "IPV6_MC_SCOPE1"
+#define IRIF_DISABLED       "IRIF_DISABLED"
+#define ERIF_DISABLED       "ERIF_DISABLED"
+#define LPM4_MISS           "LPM4_MISS"
+#define LPM6_MISS           "LPM6_MISS"
+#define BLACKHOLE_ROUTE     "BLACKHOLE_ROUTE"
+#define BLACKHOLE_ARP       "BLACKHOLE_ARP"
+#define UNRESOLVED_NEXT_HOP "UNRESOLVED_NEXT_HOP"
+#define L3_EGRESS_LINK_DOWN "L3_EGRESS_LINK_DOWN"
+
+// Tunnel Drop Reasons
+#define DECAP_ERROR         "DECAP_ERROR"
+
+// ACL Drop Reasons
+#define ACL_ANY             "ACL_ANY"
+#define ACL_INGRESS_PORT    "ACL_INGRESS_PORT"
+#define ACL_INGRESS_LAG     "ACL_INGRESS_LAG"
+#define ACL_INGRESS_VLAN    "ACL_INGRESS_VLAN"
+#define ACL_INGRESS_RIF     "ACL_INGRESS_RIF"
+#define ACL_INGRESS_SWITCH  "ACL_INGRESS_SWITCH"
+#define ACL_EGRESS_PORT     "ACL_EGRESS_PORT"
+#define ACL_EGRESS_LAG      "ACL_EGRESS_LAG"
+#define ACL_EGRESS_VLAN     "ACL_EGRESS_VLAN"
+#define ACL_EGRESS_RIF      "ACL_EGRESS_RIF"
+#define ACL_EGRESS_SWITCH   "ACL_EGRESS_SWITCH"
+
+#endif
diff --git a/orchagent/debugcounterorch.cpp b/orchagent/debugcounterorch.cpp
new file mode 100644
index 000000000000..8c989e6aa025
--- /dev/null
+++ b/orchagent/debugcounterorch.cpp
@@ -0,0 +1,588 @@
+#include "debugcounterorch.h"
+#include "portsorch.h"
+#include "rediscommand.h"
+#include "sai_serialize.h"
+#include "schema.h"
+#include "drop_counter.h"
+#include <memory>
+
+using std::string;
+using std::unordered_map;
+using std::unordered_set;
+using std::vector;
+
+extern sai_object_id_t gSwitchId;
+extern PortsOrch *gPortsOrch;
+
+static const unordered_map<string, CounterType> flex_counter_type_lookup = {
+    { PORT_INGRESS_DROPS, CounterType::PORT_DEBUG },
+    { PORT_EGRESS_DROPS, CounterType::PORT_DEBUG },
+    { SWITCH_INGRESS_DROPS, CounterType::SWITCH_DEBUG },
+    { SWITCH_EGRESS_DROPS, CounterType::SWITCH_DEBUG },
+};
+
+// Initializing DebugCounterOrch creates a group entry in FLEX_COUNTER_DB, so this
+// object should only be initialized once.
+DebugCounterOrch::DebugCounterOrch(DBConnector *db, const vector<string>& table_names, int poll_interval) :
+        Orch(db, table_names),
+        flex_counter_manager(DEBUG_COUNTER_FLEX_COUNTER_GROUP, StatsMode::READ, poll_interval),
+        m_stateDb(new DBConnector(STATE_DB, DBConnector::DEFAULT_UNIXSOCKET, 0)),
+        m_debugCapabilitiesTable(new Table(m_stateDb.get(), STATE_DEBUG_COUNTER_CAPABILITIES_NAME)),
+        m_countersDb(new DBConnector(COUNTERS_DB, DBConnector::DEFAULT_UNIXSOCKET, 0)),
+        m_counterNameToPortStatMap(new Table(m_countersDb.get(), COUNTERS_DEBUG_NAME_PORT_STAT_MAP)),
+        m_counterNameToSwitchStatMap(new Table(m_countersDb.get(), COUNTERS_DEBUG_NAME_SWITCH_STAT_MAP))
+{
+    SWSS_LOG_ENTER();
+    publishDropCounterCapabilities();
+    flex_counter_manager.enableFlexCounterGroup();
+}
+
+DebugCounterOrch::~DebugCounterOrch(void)
+{
+    SWSS_LOG_ENTER();
+}
+
+// doTask processes updates from the consumer and modifies the state of the
+// following components:
+//     1) The ASIC, by creating, modifying, and deleting debug counters
+//     2) syncd, by creating, modifying, and deleting flex counters to
+//        keep track of the debug counters
+//
+// Updates can fail due to the following:
+//    1) Malformed requests: if the update contains an unknown or unsupported
+//       counter type or drop reason then the update will fail
+//    2) SAI failures: if the SAI returns an error for any reason then the
+//       update will fail
+//  It is guaranteed that failed updates will not modify the state of the
+//  system.
+//
+// In addition, updates are idempotent - repeating the same request any number
+// of times will always result in the same external behavior.
+void DebugCounterOrch::doTask(Consumer& consumer)
+{
+    SWSS_LOG_ENTER();
+
+    // Currently we depend on 1) the switch and 2) the ports being ready
+    // before we can set up the counters. If debug counters for other
+    // object types are added we may need to update this dependency.
+    if (!gPortsOrch->allPortsReady())
+    {
+        return;
+    }
+
+    auto it = consumer.m_toSync.begin();
+    while (it != consumer.m_toSync.end())
+    {
+        KeyOpFieldsValuesTuple t = it->second;
+        string key = kfvKey(t);
+        string op = kfvOp(t);
+        std::vector<FieldValueTuple> values = kfvFieldsValues(t);
+
+        auto table_name = consumer.getTableName();
+        task_process_status task_status = task_process_status::task_ignore;
+        if (table_name == CFG_DEBUG_COUNTER_TABLE_NAME)
+        {
+            if (op == SET_COMMAND)
+            {
+                try
+                {
+                    task_status = installDebugCounter(key, values);
+                }
+                catch (const std::runtime_error& e)
+                {
+                    SWSS_LOG_ERROR("Failed to create debug counter '%s'", key.c_str());
+                    task_status = task_process_status::task_failed;
+                }
+            }
+            else if (op == DEL_COMMAND)
+            {
+                try
+                {
+                    task_status = uninstallDebugCounter(key);
+                }
+                catch (const std::runtime_error& e)
+                {
+                    SWSS_LOG_ERROR("Failed to delete debug counter '%s'", key.c_str());
+                    task_status = task_process_status::task_failed;
+                }
+            }
+            else
+            {
+                SWSS_LOG_ERROR("Unknown operation type %s\n", op.c_str());
+            }
+        }
+        else if (table_name == CFG_DEBUG_COUNTER_DROP_REASON_TABLE_NAME)
+        {
+            string counter_name, drop_reason;
+            parseDropReasonUpdate(key, '|', &counter_name, &drop_reason);
+
+            if (op == SET_COMMAND)
+            {
+                try
+                {
+                    task_status = addDropReason(counter_name, drop_reason);
+                }
+                catch (const std::runtime_error& e)
+                {
+                    SWSS_LOG_ERROR("Failed to add drop reason '%s' to counter '%s'", drop_reason.c_str(), counter_name.c_str());
+                    task_status = task_process_status::task_failed;
+                }
+            }
+            else if (op == DEL_COMMAND)
+            {
+                try
+                {
+                    task_status = removeDropReason(counter_name, drop_reason);
+                }
+                catch (const std::runtime_error& e)
+                {
+                    SWSS_LOG_ERROR("Failed to remove drop reason '%s' from counter '%s'", drop_reason.c_str(), counter_name.c_str());
+                    task_status = task_process_status::task_failed;
+                }
+            }
+            else
+            {
+                SWSS_LOG_ERROR("Unknown operation type %s\n", op.c_str());
+            }
+        }
+        else
+        {
+            SWSS_LOG_ERROR("Received update from unknown table '%s'", table_name.c_str());
+        }
+
+        switch (task_status)
+        {
+            case task_process_status::task_success:
+                consumer.m_toSync.erase(it++);
+                break;
+            case task_process_status::task_ignore:
+                SWSS_LOG_WARN("Debug counters '%s' task ignored", op.c_str());
+                consumer.m_toSync.erase(it++);
+                break;
+            case task_process_status::task_need_retry:
+                SWSS_LOG_NOTICE("Failed to process debug counters '%s' task, retrying", op.c_str());
+                ++it;
+                break;
+            case task_process_status::task_failed:
+                SWSS_LOG_ERROR("Failed to process debug counters '%s' task, error(s) occured during execution", op.c_str());
+                consumer.m_toSync.erase(it++);
+                break;
+            default:
+                SWSS_LOG_ERROR("Invalid task status %d", task_status);
+                consumer.m_toSync.erase(it++);
+                break;
+        }
+    }
+}
+
+// Debug Capability Reporting Functions START HERE -------------------------------------------------
+
+// publishDropCounterCapabilities queries the SAI for available drop counter
+// capabilities on this device and publishes the information to the
+// DROP_COUNTER_CAPABILITIES table in STATE_DB.
+void DebugCounterOrch::publishDropCounterCapabilities()
+{
+    string supported_ingress_drop_reasons = DropCounter::serializeSupportedDropReasons(
+            DropCounter::getSupportedDropReasons(SAI_DEBUG_COUNTER_ATTR_IN_DROP_REASON_LIST));
+    string supported_egress_drop_reasons  = DropCounter::serializeSupportedDropReasons(
+            DropCounter::getSupportedDropReasons(SAI_DEBUG_COUNTER_ATTR_OUT_DROP_REASON_LIST));
+
+    for (auto const &counter_type : DebugCounter::getDebugCounterTypeLookup())
+    {
+        string num_counters = std::to_string(DropCounter::getSupportedDebugCounterAmounts(counter_type.second));
+
+        string drop_reasons;
+        if (counter_type.first == PORT_INGRESS_DROPS || counter_type.first == SWITCH_INGRESS_DROPS)
+        {
+            drop_reasons = supported_ingress_drop_reasons;
+        }
+        else
+        {
+            drop_reasons = supported_egress_drop_reasons;
+        }
+
+        // Only include available capabilities in State DB
+        if (num_counters != "0" && !drop_reasons.empty())
+        {
+            vector<FieldValueTuple> fieldValues;
+            fieldValues.push_back(FieldValueTuple("count", num_counters));
+            fieldValues.push_back(FieldValueTuple("reasons", drop_reasons));
+
+            SWSS_LOG_DEBUG("Setting '%s' capabilities to count='%s', reasons='%s'", counter_type.first.c_str(), num_counters.c_str(), drop_reasons.c_str());
+            m_debugCapabilitiesTable->set(counter_type.first, fieldValues);
+        }
+    }
+}
+
+// doTask Handler Functions START HERE -------------------------------------------------------------
+
+// Note that this function cannot be used to re-initialize a counter. To create a counter
+// with the same name but different attributes (e.g. type, drop reasons, etc.) you will need
+// to delete the original counter first or use a function like addDropReason, if available.
+task_process_status DebugCounterOrch::installDebugCounter(const string& counter_name, const vector<FieldValueTuple>& attributes)
+{
+    SWSS_LOG_ENTER();
+
+    if (debug_counters.find(counter_name) != debug_counters.end())
+    {
+        SWSS_LOG_DEBUG("Debug counter '%s' already exists", counter_name.c_str());
+        return task_process_status::task_success;
+    }
+
+    // Note: this method currently assumes that all counters are drop counters.
+    // If you are adding support for a non-drop counter than it may make sense
+    // to either: a) dispatch to different handlers in doTask or b) dispatch to
+    // different helper methods in this method.
+
+    string counter_type = getDebugCounterType(attributes);
+    addFreeCounter(counter_name, counter_type);
+    reconcileFreeDropCounters(counter_name);
+
+    SWSS_LOG_NOTICE("Succesfully created drop counter %s", counter_name.c_str());
+    return task_process_status::task_success;
+}
+
+task_process_status DebugCounterOrch::uninstallDebugCounter(const string& counter_name)
+{
+    SWSS_LOG_ENTER();
+
+    auto it = debug_counters.find(counter_name);
+    if (it == debug_counters.end())
+    {
+        if (free_drop_counters.find(counter_name) != free_drop_counters.end())
+        {
+            deleteFreeCounter(counter_name);
+        }
+        else
+        {
+            SWSS_LOG_ERROR("Debug counter %s does not exist", counter_name.c_str());
+        }
+
+        return task_process_status::task_ignore;
+    }
+
+    DebugCounter *counter = it->second.get();
+    string counter_type = counter->getCounterType();
+    string counter_stat = counter->getDebugCounterSAIStat();
+
+    debug_counters.erase(it);
+    uninstallDebugFlexCounters(counter_type, counter_stat);
+
+    if (counter_type == PORT_INGRESS_DROPS || counter_type == PORT_EGRESS_DROPS)
+    {
+        m_counterNameToPortStatMap->hdel("", counter_name);
+    }
+    else
+    {
+        m_counterNameToSwitchStatMap->hdel("", counter_name);
+    }
+
+    SWSS_LOG_NOTICE("Succesfully deleted drop counter %s", counter_name.c_str());
+    return task_process_status::task_success;
+}
+
+task_process_status DebugCounterOrch::addDropReason(const string& counter_name, const string& drop_reason)
+{
+    SWSS_LOG_ENTER();
+
+    if (!isDropReasonValid(drop_reason))
+    {
+        return task_failed;
+    }
+
+    auto it = debug_counters.find(counter_name);
+    if (it == debug_counters.end())
+    {
+        // In order to gracefully handle the case where the drop reason updates
+        // are received before the create counter update, we keep track of reasons
+        // we've seen in the free_drop_reasons table.
+        addFreeDropReason(counter_name, drop_reason);
+        reconcileFreeDropCounters(counter_name);
+
+        SWSS_LOG_NOTICE("Succesfully created drop counter %s", counter_name.c_str());
+        return task_process_status::task_success;
+    }
+
+    DropCounter *counter = dynamic_cast<DropCounter*>(it->second.get());
+    counter->addDropReason(drop_reason);
+
+    SWSS_LOG_NOTICE("Added drop reason %s to drop counter %s", drop_reason.c_str(), counter_name.c_str());
+    return task_process_status::task_success;
+}
+
+// A drop counter must always contain at least one drop reason, so this function
+// will do nothing if you attempt to remove the last drop reason.
+task_process_status DebugCounterOrch::removeDropReason(const string& counter_name, const string& drop_reason)
+{
+    SWSS_LOG_ENTER();
+
+    if (!isDropReasonValid(drop_reason))
+    {
+        return task_failed;
+    }
+
+    auto it = debug_counters.find(counter_name);
+    if (it == debug_counters.end())
+    {
+        deleteFreeDropReason(counter_name, drop_reason);
+        return task_success;
+    }
+
+    DropCounter *counter = dynamic_cast<DropCounter*>(it->second.get());
+    const unordered_set<string>& drop_reasons = counter->getDropReasons();
+
+    if (drop_reasons.size() <= 1)
+    {
+        SWSS_LOG_WARN("Attempted to remove all drop reasons from counter '%s'", counter_name.c_str());
+        return task_ignore;
+    }
+
+    counter->removeDropReason(drop_reason);
+
+    SWSS_LOG_NOTICE("Removed drop reason %s from drop counter %s", drop_reason.c_str(), counter_name.c_str());
+    return task_success;
+}
+
+// Free Table Management Functions START HERE ------------------------------------------------------
+
+// Note that entries will remain in the table until at least one drop reason is added to the counter.
+void DebugCounterOrch::addFreeCounter(const string& counter_name, const string& counter_type)
+{
+    SWSS_LOG_ENTER();
+
+    if (free_drop_counters.find(counter_name) != free_drop_counters.end())
+    {
+        SWSS_LOG_DEBUG("Debug counter '%s' is in free counter table", counter_name.c_str());
+        return;
+    }
+
+    SWSS_LOG_DEBUG("Adding debug counter '%s' to free counter table", counter_name.c_str());
+    free_drop_counters.emplace(counter_name, counter_type);
+}
+
+void DebugCounterOrch::deleteFreeCounter(const string& counter_name)
+{
+    SWSS_LOG_ENTER();
+
+    if (free_drop_counters.find(counter_name) == free_drop_counters.end())
+    {
+        SWSS_LOG_ERROR("Debug counter %s does not exist", counter_name.c_str());
+        return;
+    }
+
+    SWSS_LOG_DEBUG("Removing debug counter '%s' from free counter table", counter_name.c_str());
+    free_drop_counters.erase(counter_name);
+}
+
+// Note that entries will remain in the table until a drop counter is added.
+void DebugCounterOrch::addFreeDropReason(const string& counter_name, const string& drop_reason)
+{
+    SWSS_LOG_ENTER();
+
+    auto reasons_it = free_drop_reasons.find(counter_name);
+
+    if (reasons_it == free_drop_reasons.end())
+    {
+        SWSS_LOG_DEBUG("Creating free drop reason table for counter '%s'", counter_name.c_str());
+        unordered_set<string> new_reasons = { drop_reason };
+        free_drop_reasons.emplace(make_pair(counter_name, new_reasons));
+    }
+    else
+    {
+        SWSS_LOG_DEBUG("Adding additional drop reasons to free drop reason table for counter '%s'", counter_name.c_str());
+        reasons_it->second.emplace(drop_reason);
+    }
+}
+
+void DebugCounterOrch::deleteFreeDropReason(const string& counter_name, const string& drop_reason)
+{
+    SWSS_LOG_ENTER();
+
+    auto reasons_it = free_drop_reasons.find(counter_name);
+
+    if (reasons_it == free_drop_reasons.end()) {
+        SWSS_LOG_DEBUG("Attempted to remove drop reason '%s' from counter '%s' that does not exist", drop_reason.c_str(), counter_name.c_str());
+        return;
+    }
+
+    SWSS_LOG_DEBUG("Removing free drop reason from counter '%s'", counter_name.c_str());
+    reasons_it->second.erase(drop_reason);
+
+    if (reasons_it->second.empty()) {
+        free_drop_reasons.erase(reasons_it);
+    }
+}
+
+void DebugCounterOrch::reconcileFreeDropCounters(const string& counter_name)
+{
+    SWSS_LOG_ENTER();
+
+    auto counter_it = free_drop_counters.find(counter_name);
+    auto reasons_it = free_drop_reasons.find(counter_name);
+
+    if (counter_it != free_drop_counters.end() && reasons_it != free_drop_reasons.end())
+    {
+        SWSS_LOG_DEBUG("Found counter '%s' and drop reasons, creating the counter", counter_name.c_str());
+        createDropCounter(counter_name, counter_it->second, reasons_it->second);
+        free_drop_counters.erase(counter_it);
+        free_drop_reasons.erase(reasons_it);
+        SWSS_LOG_NOTICE("Succesfully created new drop counter %s", counter_name.c_str());
+    }
+}
+
+// Flex Counter Management Functions START HERE ----------------------------------------------------
+
+CounterType DebugCounterOrch::getFlexCounterType(const string& counter_type)
+{
+    SWSS_LOG_ENTER();
+
+    auto flex_counter_type_it = flex_counter_type_lookup.find(counter_type);
+    if (flex_counter_type_it == flex_counter_type_lookup.end())
+    {
+        SWSS_LOG_ERROR("Flex counter type '%s' not found", counter_type.c_str());
+        throw runtime_error("Flex counter type not found");
+    }
+    return flex_counter_type_it->second;
+}
+
+void DebugCounterOrch::installDebugFlexCounters(const string& counter_type,
+                                                const string& counter_stat)
+{
+    SWSS_LOG_ENTER();
+    CounterType flex_counter_type = getFlexCounterType(counter_type);
+
+    if (flex_counter_type == CounterType::SWITCH_DEBUG)
+    {
+        flex_counter_manager.addFlexCounterStat(gSwitchId, flex_counter_type, counter_stat);
+    }
+    else if (flex_counter_type == CounterType::PORT_DEBUG)
+    {
+        for (auto const &curr : gPortsOrch->getAllPorts())
+        {
+            if (curr.second.m_type != Port::Type::PHY)
+            {
+                continue;
+            }
+
+            flex_counter_manager.addFlexCounterStat(
+                    curr.second.m_port_id,
+                    flex_counter_type,
+                    counter_stat);
+        }
+    }
+}
+
+void DebugCounterOrch::uninstallDebugFlexCounters(const string& counter_type,
+                                                  const string& counter_stat)
+{
+    SWSS_LOG_ENTER();
+    CounterType flex_counter_type = getFlexCounterType(counter_type);
+
+    if (flex_counter_type == CounterType::SWITCH_DEBUG)
+    {
+        flex_counter_manager.removeFlexCounterStat(gSwitchId, flex_counter_type, counter_stat);
+    }
+    else if (flex_counter_type == CounterType::PORT_DEBUG)
+    {
+        for (auto const &curr : gPortsOrch->getAllPorts())
+        {
+            if (curr.second.m_type != Port::Type::PHY)
+            {
+                continue;
+            }
+
+            flex_counter_manager.removeFlexCounterStat(
+                curr.second.m_port_id,
+                flex_counter_type,
+                counter_stat);
+        }
+    }
+}
+
+// Debug Counter Initialization Helper Functions START HERE ----------------------------------------
+
+// NOTE: At this point COUNTER_TYPE is the only field from CONFIG_DB that we care about. In the future
+// if other fields become relevant it may make sense to extend this method and return a struct with the
+// relevant fields.
+std::string DebugCounterOrch::getDebugCounterType(const vector<FieldValueTuple>& values) const
+{
+    SWSS_LOG_ENTER();
+
+    std::string counter_type;
+    for (auto attr : values)
+    {
+        std::string attr_name = fvField(attr);
+        auto supported_debug_counter_attributes = DebugCounter::getSupportedDebugCounterAttributes();
+        auto attr_name_it = supported_debug_counter_attributes.find(attr_name);
+        if (attr_name_it == supported_debug_counter_attributes.end())
+        {
+            SWSS_LOG_ERROR("Unknown debug counter attribute '%s'", attr_name.c_str());
+            continue;
+        }
+
+        std::string attr_value = fvValue(attr);
+        if (attr_name == "type")
+        {
+            auto debug_counter_type_lookup = DebugCounter::getDebugCounterTypeLookup();
+            auto counter_type_it = debug_counter_type_lookup.find(attr_value);
+            if (counter_type_it == debug_counter_type_lookup.end())
+            {
+                SWSS_LOG_ERROR("Debug counter type '%s' does not exist", attr_value.c_str());
+                throw std::runtime_error("Failed to initialize debug counter");
+            }
+
+            counter_type = counter_type_it->first;
+        }
+    }
+
+    return counter_type;
+}
+
+// createDropCounter creates a new drop counter in the SAI and installs a
+// flex counter to poll the counter data.
+//
+// If SAI initialization fails or flex counter installation fails then this
+// method will throw an exception.
+void DebugCounterOrch::createDropCounter(const string& counter_name, const string& counter_type, const unordered_set<string>& drop_reasons)
+{
+    auto counter = std::unique_ptr<DropCounter>(new DropCounter(counter_name, counter_type, drop_reasons));
+    std::string counter_stat = counter->getDebugCounterSAIStat();
+    debug_counters.emplace(counter_name, std::move(counter));
+    installDebugFlexCounters(counter_type, counter_stat);
+
+    if (counter_type == PORT_INGRESS_DROPS || counter_type == PORT_EGRESS_DROPS)
+    {
+        m_counterNameToPortStatMap->set("", { FieldValueTuple(counter_name, counter_stat) });
+    }
+    else
+    {
+        m_counterNameToSwitchStatMap->set("", { FieldValueTuple(counter_name, counter_stat) });
+    }
+}
+
+// Debug Counter Configuration Helper Functions START HERE -----------------------------------------
+
+// parseDropReasonUpdate takes a key from CONFIG_DB and returns the 1) the counter name being targeted and
+// 2) the drop reason to be added or removed via output parameters.
+void DebugCounterOrch::parseDropReasonUpdate(const string& key, const char delimeter, string *counter_name, string *drop_reason) const
+{
+    size_t counter_end = key.find(delimeter);
+    *counter_name = key.substr(0, counter_end);
+    SWSS_LOG_DEBUG("DEBUG_COUNTER COUNTER NAME = %s (%d, %zd)", counter_name->c_str(), 0, counter_end);
+    *drop_reason = key.substr(counter_end + 1);
+    SWSS_LOG_DEBUG("DEBUG_COUNTER RULE NAME = %s (%zd, %zd)", drop_reason->c_str(), counter_end + 1, key.length());
+}
+
+bool DebugCounterOrch::isDropReasonValid(const string& drop_reason) const
+{
+    SWSS_LOG_ENTER();
+
+    if (!DropCounter::isIngressDropReasonValid(drop_reason) &&
+        !DropCounter::isEgressDropReasonValid(drop_reason))
+    {
+        SWSS_LOG_ERROR("Drop reason %s not found", drop_reason.c_str());
+        return false;
+    }
+
+    return true;
+}
\ No newline at end of file
diff --git a/orchagent/debugcounterorch.h b/orchagent/debugcounterorch.h
new file mode 100644
index 000000000000..49ca64b7f73f
--- /dev/null
+++ b/orchagent/debugcounterorch.h
@@ -0,0 +1,97 @@
+#ifndef DEBUG_COUNTER_ORCH_H
+#define DEBUG_COUNTER_ORCH_H
+
+#include <string>
+#include <vector>
+#include <unordered_map>
+#include <unordered_set>
+
+#include "orch.h"
+#include "flex_counter_stat_manager.h"
+#include "debug_counter.h"
+#include "drop_counter.h"
+
+extern "C" {
+#include "sai.h"
+}
+
+#define DEBUG_COUNTER_FLEX_COUNTER_GROUP "DEBUG_COUNTER"
+
+// DebugCounterOrch is an orchestrator for managing debug counters. It handles
+// the creation, deletion, and modification of debug counters.
+class DebugCounterOrch: public Orch
+{
+public:
+    DebugCounterOrch(swss::DBConnector *db, const std::vector<std::string>& table_names, int poll_interval);
+    virtual ~DebugCounterOrch(void);
+
+    void doTask(Consumer& consumer);
+
+private:
+    // Debug Capability Reporting Functions
+    void publishDropCounterCapabilities();
+
+    // doTask Handler Functions
+    task_process_status installDebugCounter(const std::string& counter_name, const std::vector<swss::FieldValueTuple>& attributes);
+    task_process_status uninstallDebugCounter(const std::string& counter_name);
+    task_process_status addDropReason(const std::string& counter_name, const std::string& drop_reason);
+    task_process_status removeDropReason(const std::string& counter_name, const std::string& drop_reason);
+
+    // Free Table Management Functions
+    void addFreeCounter(const std::string& counter_name, const std::string& counter_type);
+    void deleteFreeCounter(const std::string& counter_name);
+    void addFreeDropReason(const std::string& counter_name, const std::string& drop_reason);
+    void deleteFreeDropReason(const std::string& counter_name, const std::string& drop_reason);
+    void reconcileFreeDropCounters(const std::string& counter_name);
+
+    // Flex Counter Management Functions
+    CounterType getFlexCounterType(const std::string& counter_type) noexcept(false);
+    void installDebugFlexCounters(
+            const std::string& counter_type,
+            const std::string& counter_stat);
+    void uninstallDebugFlexCounters(
+            const std::string& counter_type,
+            const std::string& counter_stat);
+
+    // Debug Counter Initialization Helper Functions
+    std::string getDebugCounterType(
+            const std::vector<swss::FieldValueTuple>& values) const noexcept(false);
+    void createDropCounter(
+            const std::string& counter_name,
+            const std::string& counter_type,
+            const std::unordered_set<std::string>& drop_reasons) noexcept(false);
+
+    // Debug Counter Configuration Helper Functions
+    void parseDropReasonUpdate(
+            const std::string& key,
+            const char delimeter,
+            std::string *counter_name,
+            std::string *drop_reason) const;
+    bool isDropReasonValid(const std::string& drop_reason) const;
+
+    // Data Members
+    std::shared_ptr<swss::DBConnector> m_stateDb = nullptr;
+    std::shared_ptr<swss::Table> m_debugCapabilitiesTable = nullptr;
+
+    std::shared_ptr<swss::DBConnector> m_countersDb = nullptr;
+    std::shared_ptr<swss::Table> m_counterNameToPortStatMap = nullptr;
+    std::shared_ptr<swss::Table> m_counterNameToSwitchStatMap = nullptr;
+
+    FlexCounterStatManager flex_counter_manager;
+
+    std::unordered_map<std::string, std::unique_ptr<DebugCounter>> debug_counters;
+
+    // free_drop_counters are drop counters that have been created by a user
+    // that do not have any drop reasons associated with them yet. Because
+    // we cannot create a drop counter without any drop reasons, we keep track
+    // of these counters in this table.
+    std::unordered_map<std::string, std::string> free_drop_counters;
+
+    // free_drop_reasons are drop reasons that have been added by a user
+    // that do not have a counter associated with them yet. Because we
+    // cannot add drop reasons to a counter that doesn't exist yet,
+    // we keep track of the reasons in this table.
+    std::unordered_map<std::string, std::unordered_set<std::string>> free_drop_reasons;
+};
+
+#endif
diff --git a/orchagent/flexcounterorch.cpp b/orchagent/flexcounterorch.cpp
index 8493f6897a34..c94c1b189116 100644
--- a/orchagent/flexcounterorch.cpp
+++ b/orchagent/flexcounterorch.cpp
@@ -7,6 +7,8 @@
 #include "sai_serialize.h"
 #include "pfcwdorch.h"
 #include "bufferorch.h"
+#include "flexcounterorch.h"
+#include "debugcounterorch.h"
 
 extern sai_port_api_t *sai_port_api;
 
@@ -25,6 +27,7 @@ unordered_map<string, string> flexCounterGroupMap =
     {"PG_WATERMARK", PG_WATERMARK_STAT_COUNTER_FLEX_COUNTER_GROUP},
     {BUFFER_POOL_WATERMARK_KEY, BUFFER_POOL_WATERMARK_STAT_COUNTER_FLEX_COUNTER_GROUP},
     {"RIF", RIF_STAT_COUNTER_FLEX_COUNTER_GROUP},
+    {"DEBUG_COUNTER", DEBUG_COUNTER_FLEX_COUNTER_GROUP},
 };
 
 
diff --git a/orchagent/orchdaemon.cpp b/orchagent/orchdaemon.cpp
index ada19d1494db..756b8a35eb3d 100644
--- a/orchagent/orchdaemon.cpp
+++ b/orchagent/orchdaemon.cpp
@@ -200,6 +200,13 @@ bool OrchDaemon::init()
     };
     SflowOrch *sflow_orch = new SflowOrch(m_applDb,  sflow_tables);
 
+    vector<string> debug_counter_tables = {
+        CFG_DEBUG_COUNTER_TABLE_NAME,
+        CFG_DEBUG_COUNTER_DROP_REASON_TABLE_NAME
+    };
+
+    DebugCounterOrch *debug_counter_orch = new DebugCounterOrch(m_configDb, debug_counter_tables, 1000);
+
     /*
      * The order of the orch list is important for state restore of warm start and
      * the queued processing in m_toSync map after gPortsOrch->allPortsReady() is set.
@@ -208,8 +215,7 @@ bool OrchDaemon::init()
      * when iterating ConsumerMap.
      * That is ensured implicitly by the order of map key, "LAG_TABLE" is smaller than "VLAN_TABLE" in lexicographic order.
      */
-    m_orchList = { gSwitchOrch, gCrmOrch, gBufferOrch, gPortsOrch, gIntfsOrch, gNeighOrch, gRouteOrch, copp_orch, tunnel_decap_orch, qos_orch, wm_orch, policer_orch, sflow_orch};
-
+    m_orchList = { gSwitchOrch, gCrmOrch, gBufferOrch, gPortsOrch, gIntfsOrch, gNeighOrch, gRouteOrch, copp_orch, tunnel_decap_orch, qos_orch, wm_orch, policer_orch, sflow_orch, debug_counter_orch};
 
     bool initialize_dtel = false;
     if (platform == BFN_PLATFORM_SUBSTRING || platform == VS_PLATFORM_SUBSTRING)
diff --git a/orchagent/orchdaemon.h b/orchagent/orchdaemon.h
index bfa52fdef818..5e9c9b914afb 100644
--- a/orchagent/orchdaemon.h
+++ b/orchagent/orchdaemon.h
@@ -28,6 +28,7 @@
 #include "watermarkorch.h"
 #include "policerorch.h"
 #include "sfloworch.h"
+#include "debugcounterorch.h"
 #include "directory.h"
 
 using namespace swss;
diff --git a/orchagent/saihelper.cpp b/orchagent/saihelper.cpp
index dbadc7e28830..9fbe44917960 100644
--- a/orchagent/saihelper.cpp
+++ b/orchagent/saihelper.cpp
@@ -42,6 +42,7 @@ sai_fdb_api_t*              sai_fdb_api;
 sai_dtel_api_t*             sai_dtel_api;
 sai_bmtor_api_t*            sai_bmtor_api;
 sai_samplepacket_api_t*     sai_samplepacket_api;
+sai_debug_counter_api_t*    sai_debug_counter_api;
 
 extern sai_object_id_t gSwitchId;
 extern bool gSairedisRecord;
@@ -132,6 +133,7 @@ void initSaiApi()
     sai_api_query(SAI_API_DTEL,                 (void **)&sai_dtel_api);
     sai_api_query((sai_api_t)SAI_API_BMTOR,     (void **)&sai_bmtor_api);
     sai_api_query(SAI_API_SAMPLEPACKET,         (void **)&sai_samplepacket_api);
+    sai_api_query(SAI_API_DEBUG_COUNTER,        (void **)&sai_debug_counter_api);
 
     sai_log_set(SAI_API_SWITCH,                 SAI_LOG_LEVEL_NOTICE);
     sai_log_set(SAI_API_BRIDGE,                 SAI_LOG_LEVEL_NOTICE);
@@ -159,6 +161,7 @@ void initSaiApi()
     sai_log_set(SAI_API_DTEL,                   SAI_LOG_LEVEL_NOTICE);
     sai_log_set((sai_api_t)SAI_API_BMTOR,       SAI_LOG_LEVEL_NOTICE);
     sai_log_set(SAI_API_SAMPLEPACKET,           SAI_LOG_LEVEL_NOTICE);
+    sai_log_set(SAI_API_DEBUG_COUNTER,          SAI_LOG_LEVEL_NOTICE);
 }
 
 void initSaiRedis(const string &record_location)
diff --git a/tests/mock_tests/Makefile.am b/tests/mock_tests/Makefile.am
index 247a33bf1adf..73574f5b6297 100644
--- a/tests/mock_tests/Makefile.am
+++ b/tests/mock_tests/Makefile.am
@@ -1,3 +1,8 @@
+FLEX_CTR_DIR = $(top_srcdir)/orchagent/flex_counter
+DEBUG_CTR_DIR = $(top_srcdir)/orchagent/debug_counter
+
+INCLUDES = -I $(FLEX_CTR_DIR) -I $(DEBUG_CTR_DIR)
+
 CFLAGS_SAI = -I /usr/include/sai
 
 TESTS = tests
@@ -54,7 +59,11 @@ tests_SOURCES = aclorch_ut.cpp \
                 $(top_srcdir)/orchagent/flexcounterorch.cpp \
                 $(top_srcdir)/orchagent/watermarkorch.cpp \
                 $(top_srcdir)/orchagent/chassisorch.cpp \
-                $(top_srcdir)/orchagent/sfloworch.cpp
+                $(top_srcdir)/orchagent/sfloworch.cpp \
+                $(top_srcdir)/orchagent/debugcounterorch.cpp
+
+tests_SOURCES += $(FLEX_CTR_DIR)/flex_counter_manager.cpp $(FLEX_CTR_DIR)/flex_counter_stat_manager.cpp
+tests_SOURCES += $(DEBUG_CTR_DIR)/debug_counter.cpp $(DEBUG_CTR_DIR)/drop_counter.cpp
 
 tests_CFLAGS = $(DBGFLAGS) $(AM_CFLAGS) $(CFLAGS_COMMON) $(CFLAGS_GTEST) $(CFLAGS_SAI)
 tests_CPPFLAGS = $(DBGFLAGS) $(AM_CFLAGS) $(CFLAGS_COMMON) $(CFLAGS_GTEST) $(CFLAGS_SAI) -I$(top_srcdir)/orchagent
diff --git a/tests/test_drop_counters.py b/tests/test_drop_counters.py
new file mode 100644
index 000000000000..473c20885f9f
--- /dev/null
+++ b/tests/test_drop_counters.py
@@ -0,0 +1,705 @@
+from swsscommon import swsscommon
+
+import time
+
+# Supported drop counters
+PORT_INGRESS_DROPS = 'PORT_INGRESS_DROPS'
+PORT_EGRESS_DROPS = 'PORT_EGRESS_DROPS'
+SWITCH_INGRESS_DROPS = 'SWITCH_INGRESS_DROPS'
+SWITCH_EGRESS_DROPS = 'SWITCH_EGRESS_DROPS'
+
+# Debug Counter Table
+DEBUG_COUNTER_TABLE = 'DEBUG_COUNTER'
+DROP_REASON_TABLE =   'DEBUG_COUNTER_DROP_REASON'
+
+# Debug Counter Capability Table
+CAPABILITIES_TABLE = 'DEBUG_COUNTER_CAPABILITIES'
+CAP_COUNT = 'count'
+CAP_REASONS = 'reasons'
+SUPPORTED_COUNTER_CAPABILITIES = [CAP_COUNT, CAP_REASONS]
+INGRESS_COUNTER_TYPES = [PORT_INGRESS_DROPS, SWITCH_INGRESS_DROPS]
+EGRESS_COUNTER_TYPES = [PORT_EGRESS_DROPS, SWITCH_EGRESS_DROPS]
+SUPPORTED_COUNTER_TYPES = INGRESS_COUNTER_TYPES + EGRESS_COUNTER_TYPES
+
+# Debug Counter Flex Counter Group
+FLEX_COUNTER_GROUP_TABLE = 'FLEX_COUNTER_GROUP_TABLE'
+DEBUG_COUNTER_FLEX_GROUP = 'DEBUG_COUNTER'
+FLEX_STATS_MODE_FIELD = 'STATS_MODE'
+FLEX_POLL_INTERVAL_FIELD = 'POLL_INTERVAL'
+FLEX_STATUS_FIELD = 'FLEX_COUNTER_STATUS'
+FLEX_STATUS_ENABLE = 'enable'
+EXPECTED_FLEX_STATS_MODE = 'STATS_MODE_READ'
+EXPECTED_POLL_INTERVAL_THRESHOLD = 0
+EXPECTED_FLEX_GROUP_FIELDS = [FLEX_STATS_MODE_FIELD, FLEX_POLL_INTERVAL_FIELD, FLEX_STATUS_FIELD]
+
+# Debug Counter Flex Counters
+FLEX_COUNTER_TABLE = 'FLEX_COUNTER_TABLE:DEBUG_COUNTER'
+PORT_DEBUG_COUNTER_LIST = 'PORT_DEBUG_COUNTER_ID_LIST'
+SWITCH_DEBUG_COUNTER_LIST = 'SWITCH_DEBUG_COUNTER_ID_LIST'
+PORT_STAT_BASE = 'SAI_PORT_STAT_IN_DROP_REASON_RANGE_BASE'
+PORT_STAT_INDEX_1 = 'SAI_PORT_STAT_IN_CONFIGURED_DROP_REASONS_1_DROPPED_PKTS'
+SWITCH_STAT_BASE = 'SAI_SWITCH_STAT_IN_DROP_REASON_RANGE_BASE'
+SWITCH_STAT_INDEX_1 = 'SAI_SWITCH_STAT_IN_CONFIGURED_DROP_REASONS_1_DROPPED_PKTS'
+
+# ASIC DB Fields
+ASIC_STATE_TABLE = 'ASIC_STATE:SAI_OBJECT_TYPE_DEBUG_COUNTER'
+ASIC_COUNTER_TYPE_FIELD = 'SAI_DEBUG_COUNTER_ATTR_TYPE'
+ASIC_COUNTER_INGRESS_REASON_LIST_FIELD = 'SAI_DEBUG_COUNTER_ATTR_IN_DROP_REASON_LIST'
+ASIC_COUNTER_EGRESS_REASON_LIST_FIELD = 'SAI_DEBUG_COUNTER_ATTR_OUT_DROP_REASON_LIST'
+ASIC_COUNTER_PORT_IN_TYPE = 'SAI_DEBUG_COUNTER_TYPE_PORT_IN_DROP_REASONS'
+ASIC_COUNTER_PORT_OUT_TYPE = 'SAI_DEBUG_COUNTER_TYPE_PORT_OUT_DROP_REASONS'
+ASIC_COUNTER_SWITCH_IN_TYPE = 'SAI_DEBUG_COUNTER_TYPE_SWITCH_IN_DROP_REASONS'
+ASIC_COUNTER_SWITCH_OUT_TYPE = 'SAI_DEBUG_COUNTER_TYPE_SWITCH_OUT_DROP_REASONS'
+EXPECTED_ASIC_FIELDS = [ASIC_COUNTER_TYPE_FIELD, ASIC_COUNTER_INGRESS_REASON_LIST_FIELD, ASIC_COUNTER_EGRESS_REASON_LIST_FIELD]
+EXPECTED_NUM_ASIC_FIELDS = 2
+
+# FIXME: It is really annoying to have to re-run tests due to inconsistent timing, should
+# implement some sort of polling interface for checking ASIC/flex counter tables after
+# applying changes to config DB
+class TestDropCounters(object):
+    def setup_db(self, dvs):
+        self.asic_db = swsscommon.DBConnector(1, dvs.redis_sock, 0)
+        self.config_db = swsscommon.DBConnector(4, dvs.redis_sock, 0)
+        self.flex_db = swsscommon.DBConnector(5, dvs.redis_sock, 0)
+        self.state_db = swsscommon.DBConnector(6, dvs.redis_sock, 0)
+
+    def genericGetAndAssert(self, table, key):
+        status, fields = table.get(key)
+        assert status
+        return fields
+
+    def checkFlexCounterGroup(self):
+        flex_group_table = swsscommon.Table(self.flex_db, FLEX_COUNTER_GROUP_TABLE)
+        status, group_vars = flex_group_table.get(DEBUG_COUNTER_FLEX_GROUP)
+        assert status
+        assert len(group_vars) == len(EXPECTED_FLEX_GROUP_FIELDS)
+
+        for var in group_vars:
+            assert len(var) == 2
+            group_field = var[0]
+            group_contents = var[1]
+
+            assert group_field in EXPECTED_FLEX_GROUP_FIELDS
+
+            if group_field == FLEX_STATS_MODE_FIELD:
+                assert group_contents == EXPECTED_FLEX_STATS_MODE
+            elif group_field == FLEX_POLL_INTERVAL_FIELD:
+                assert int(group_contents) > EXPECTED_POLL_INTERVAL_THRESHOLD
+            elif group_field == FLEX_STATUS_FIELD:
+                assert group_contents == FLEX_STATUS_ENABLE
+            else:
+                assert False
+
+    def checkFlexState(self, stats, counter_list_type):
+        flex_counter_table = swsscommon.Table(self.flex_db, FLEX_COUNTER_TABLE)
+
+        for oid in flex_counter_table.getKeys():
+            attributes = self.genericGetAndAssert(flex_counter_table, oid)
+            assert len(attributes) == 1
+            field, tracked_stats = attributes[0]
+            assert field == counter_list_type
+            for stat in stats:
+                assert stat in tracked_stats
+
+    def checkASICState(self, counter, counter_type, reasons):
+        asic_state_table = swsscommon.Table(self.asic_db, ASIC_STATE_TABLE)
+        asic_counter_params = self.genericGetAndAssert(asic_state_table, counter)
+        if len(asic_counter_params) != EXPECTED_NUM_ASIC_FIELDS:
+            return False
+
+        for param in asic_counter_params:
+            param_name = param[0]
+            param_contents = param[1]
+
+            if param_name not in EXPECTED_ASIC_FIELDS:
+                return False
+
+            if param_name == ASIC_COUNTER_TYPE_FIELD:
+                if param_contents != counter_type:
+                    return False
+            elif param_name == ASIC_COUNTER_INGRESS_REASON_LIST_FIELD:
+                if int(param_contents[0]) != len(reasons):
+                    return False
+
+                for reason in reasons:
+                    if reason not in param_contents:
+                        return False
+            else:
+                return False
+
+        return True
+
+    def create_drop_counter(self, name, counter_type):
+        debug_counter_table = swsscommon.Table(self.config_db, DEBUG_COUNTER_TABLE)
+        counter_metadata = swsscommon.FieldValuePairs([('type', counter_type)])
+        debug_counter_table.set(name, counter_metadata)
+
+    def add_drop_reason(self, name, drop_reason):
+        drop_reason_table = swsscommon.Table(self.config_db, '{}|{}'.format(DROP_REASON_TABLE, name))
+        drop_reason_entry = swsscommon.FieldValuePairs([('NULL', 'NULL')])
+        drop_reason_table.set(drop_reason, drop_reason_entry)
+
+    def remove_drop_reason(self, name, drop_reason):
+        drop_reason_table = swsscommon.Table(self.config_db, '{}|{}'.format(DROP_REASON_TABLE, name))
+        drop_reason_table._del(drop_reason)
+
+    def delete_drop_counter(self, name):
+        debug_counter_table = swsscommon.Table(self.config_db, DEBUG_COUNTER_TABLE)
+        debug_counter_table._del(name)
+
+    def test_deviceCapabilitiesTablePopulated(self, dvs, testlog):
+        """
+            This test verifies that DebugCounterOrch has succesfully queried
+            the capabilities for this device and populated state DB with the
+            correct values.
+        """
+        self.setup_db(dvs)
+
+        # Check that the capabilities table 1) exists and 2) has been populated
+        # for each type of counter
+        capabilities_table = swsscommon.Table(self.state_db, CAPABILITIES_TABLE)
+        counter_types = capabilities_table.getKeys()
+        assert len(counter_types) == len(SUPPORTED_COUNTER_TYPES)
+
+        # Check that the data for each counter type is consistent
+        for counter_type in SUPPORTED_COUNTER_TYPES:
+            assert counter_type in counter_types
+
+            # By definiton, each capability entry should contain exactly the same fields
+            counter_capabilities = self.genericGetAndAssert(capabilities_table, counter_type)
+            assert len(counter_capabilities) == len(SUPPORTED_COUNTER_CAPABILITIES)
+
+            # Check that the values of each field actually match the
+            # capabilities currently defined in the virtual switch
+            for capability in counter_capabilities:
+                assert len(capability) == 2
+                capability_name = capability[0]
+                capability_contents = capability[1]
+
+                assert capability_name in SUPPORTED_COUNTER_CAPABILITIES
+
+                if capability_name == CAP_COUNT:
+                    assert int(capability_contents) == 3
+                elif capability_name == CAP_REASONS and counter_type in INGRESS_COUNTER_TYPES:
+                    assert len(capability_contents.split(',')) == 3
+                elif capability_name == CAP_REASONS and counter_type in EGRESS_COUNTER_TYPES:
+                    assert len(capability_contents.split(',')) == 2
+                else:
+                    assert False
+
+    def test_flexCounterGroupInitialized(self, dvs, testlog):
+        """
+            This test verifies that DebugCounterOrch has succesfully
+            setup a flex counter group for the drop counters.
+        """
+        self.setup_db(dvs)
+        self.checkFlexCounterGroup()
+
+    def test_createAndRemoveDropCounterBasic(self, dvs, testlog):
+        """
+            This test verifies that a drop counter can succesfully be
+            created and deleted.
+        """
+        self.setup_db(dvs)
+
+        asic_state_table = swsscommon.Table(self.asic_db, ASIC_STATE_TABLE)
+        flex_counter_table = swsscommon.Table(self.flex_db, FLEX_COUNTER_TABLE)
+
+        name = 'TEST'
+        reason = 'L3_ANY'
+
+        self.create_drop_counter(name, PORT_INGRESS_DROPS)
+
+        # Because no reasons have been added to the counter yet, nothing should
+        # be put in ASIC DB and the flex counters should not start polling yet.
+        assert len(asic_state_table.getKeys()) == 0
+        assert len(flex_counter_table.getKeys()) == 0
+
+        self.add_drop_reason(name, reason)
+        time.sleep(3)
+
+        # Verify that the flex counters have been created to poll the new
+        # counter.
+        self.checkFlexState([PORT_STAT_BASE], PORT_DEBUG_COUNTER_LIST)
+
+        # Verify that the drop counter has been added to ASIC DB with the
+        # correct reason added.
+        asic_keys = asic_state_table.getKeys()
+        assert len(asic_keys) == 1
+        assert self.checkASICState(asic_keys[0], ASIC_COUNTER_PORT_IN_TYPE, [reason])
+
+        self.delete_drop_counter(name)
+        time.sleep(3)
+
+        # Verify that the counter has been removed from ASIC DB and the flex
+        # counters have been torn down.
+        assert len(asic_state_table.getKeys()) == 0
+        assert len(flex_counter_table.getKeys()) == 0
+
+        # Cleanup for the next test.
+        self.remove_drop_reason(name, reason)
+
+    def test_createAndRemoveDropCounterReversed(self, dvs, testlog):
+        """
+            This test verifies that a drop counter can succesfully be created
+            and deleted when the drop reasons are added before the counter is
+            created.
+        """
+        self.setup_db(dvs)
+
+        asic_state_table = swsscommon.Table(self.asic_db, ASIC_STATE_TABLE)
+        flex_counter_table = swsscommon.Table(self.flex_db, FLEX_COUNTER_TABLE)
+
+        name = 'TEST'
+        reason = 'L3_ANY'
+
+        self.add_drop_reason(name, reason)
+
+        # Because the actual counter has not been created yet, nothing should
+        # be put in ASIC DB and the flex counters should not start polling yet.
+        assert len(asic_state_table.getKeys()) == 0
+        assert len(flex_counter_table.getKeys()) == 0
+
+        self.create_drop_counter(name, PORT_INGRESS_DROPS)
+        time.sleep(3)
+
+        # Verify that the flex counters have been created to poll the new
+        # counter.
+        self.checkFlexState([PORT_STAT_BASE], PORT_DEBUG_COUNTER_LIST)
+
+        # Verify that the drop counter has been added to ASIC DB with the
+        # correct reason added.
+        asic_keys = asic_state_table.getKeys()
+        assert len(asic_keys) == 1
+        assert self.checkASICState(asic_keys[0], ASIC_COUNTER_PORT_IN_TYPE, [reason])
+
+        self.delete_drop_counter(name)
+        time.sleep(3)
+
+        # Verify that the counter has been removed from ASIC DB and the flex
+        # counters have been torn down.
+        assert len(asic_state_table.getKeys()) == 0
+        assert len(flex_counter_table.getKeys()) == 0
+
+        # Cleanup for the next test.
+        self.remove_drop_reason(name, reason)
+
+    def test_createCounterWithInvalidCounterType(self, dvs, testlog):
+        """
+            This test verifies that the state of the system is unaffected
+            when an invalid counter type is passed to CONFIG DB.
+        """
+        self.setup_db(dvs)
+
+        asic_state_table = swsscommon.Table(self.asic_db, ASIC_STATE_TABLE)
+        flex_counter_table = swsscommon.Table(self.flex_db, FLEX_COUNTER_TABLE)
+
+        name = 'BAD_CTR'
+        reason = 'L3_ANY'
+
+        self.create_drop_counter(name, 'THIS_IS_DEFINITELY_NOT_A_VALID_COUNTER_TYPE')
+        self.add_drop_reason(name, reason)
+        time.sleep(3)
+
+        # Verify that nothing has been added to ASIC DB and no flex counters
+        # were created.
+        assert len(asic_state_table.getKeys()) == 0
+        assert len(flex_counter_table.getKeys()) == 0
+
+        # Cleanup for the next test.
+        self.delete_drop_counter(name)
+        self.remove_drop_reason(name, reason)
+
+    def test_createCounterWithInvalidDropReason(self, dvs, testlog):
+        """
+            This test verifies that the state of the system is unaffected
+            when an invalid drop reason is passed to CONFIG DB.
+        """
+        self.setup_db(dvs)
+
+        asic_state_table = swsscommon.Table(self.asic_db, ASIC_STATE_TABLE)
+        flex_counter_table = swsscommon.Table(self.flex_db, FLEX_COUNTER_TABLE)
+
+        name = 'BAD_CTR'
+        reason = 'THIS_IS_DEFINITELY_NOT_A_VALID_DROP_REASON'
+
+        self.create_drop_counter(name, SWITCH_INGRESS_DROPS)
+        self.add_drop_reason(name, reason)
+        time.sleep(3)
+
+        # Verify that nothing has been added to ASIC DB and no flex counters
+        # were created.
+        assert len(asic_state_table.getKeys()) == 0
+        assert len(flex_counter_table.getKeys()) == 0
+
+        # Cleanup for the next test.
+        self.delete_drop_counter(name)
+        self.remove_drop_reason(name, reason)
+
+    def test_addReasonToInitializedCounter(self, dvs, testlog):
+        """
+            This test verifies that a drop reason can be added to a counter
+            that has already been initialized.
+        """
+        self.setup_db(dvs)
+
+        asic_state_table = swsscommon.Table(self.asic_db, ASIC_STATE_TABLE)
+        flex_counter_table = swsscommon.Table(self.flex_db, FLEX_COUNTER_TABLE)
+
+        name = 'ADD_TEST'
+        reason1 = 'L3_ANY'
+
+        self.create_drop_counter(name, SWITCH_INGRESS_DROPS)
+        self.add_drop_reason(name, reason1)
+        time.sleep(3)
+
+        # Verify that a counter has been created. We will verify the state of
+        # the counter in the next step.
+        assert len(asic_state_table.getKeys()) == 1
+        self.checkFlexState([SWITCH_STAT_BASE], SWITCH_DEBUG_COUNTER_LIST)
+
+        reason2 = 'ACL_ANY'
+        self.add_drop_reason(name, reason2)
+        time.sleep(3)
+
+        # Verify that the drop counter has been added to ASIC DB, including the
+        # reason that was added.
+        asic_keys = asic_state_table.getKeys()
+        assert len(asic_keys) == 1
+        assert self.checkASICState(asic_keys[0], ASIC_COUNTER_SWITCH_IN_TYPE, [reason1, reason2])
+
+        # Cleanup for the next test.
+        self.delete_drop_counter(name)
+        self.remove_drop_reason(name, reason1)
+        self.remove_drop_reason(name, reason2)
+
+    def test_removeReasonFromInitializedCounter(self, dvs, testlog):
+        """
+            This test verifies that a drop reason can be removed from a counter
+            that has already been initialized without deleting the counter.
+        """
+        self.setup_db(dvs)
+
+        asic_state_table = swsscommon.Table(self.asic_db, ASIC_STATE_TABLE)
+        flex_counter_table = swsscommon.Table(self.flex_db, FLEX_COUNTER_TABLE)
+
+        name = 'ADD_TEST'
+        reason1 = 'L3_ANY'
+        reason2 = 'ACL_ANY'
+
+        self.create_drop_counter(name, SWITCH_INGRESS_DROPS)
+        self.add_drop_reason(name, reason1)
+        self.add_drop_reason(name, reason2)
+        time.sleep(3)
+
+        # Verify that a counter has been created. We will verify the state of
+        # the counter in the next step.
+        assert len(asic_state_table.getKeys()) == 1
+        self.checkFlexState([SWITCH_STAT_BASE], SWITCH_DEBUG_COUNTER_LIST)
+
+        self.remove_drop_reason(name, reason2)
+        time.sleep(3)
+
+        # Verify that the drop counter has been added to ASIC DB, excluding the
+        # reason that was removed.
+        asic_keys = asic_state_table.getKeys()
+        assert len(asic_keys) == 1
+        assert self.checkASICState(asic_keys[0], ASIC_COUNTER_SWITCH_IN_TYPE, [reason1])
+
+        # Cleanup for the next test.
+        self.delete_drop_counter(name)
+        self.remove_drop_reason(name, reason1)
+
+    def test_removeAllDropReasons(self, dvs, testlog):
+        """
+            This test verifies that it is not possible to remove all drop
+            reasons from a drop counter.
+        """
+        self.setup_db(dvs)
+
+        asic_state_table = swsscommon.Table(self.asic_db, ASIC_STATE_TABLE)
+        flex_counter_table = swsscommon.Table(self.flex_db, FLEX_COUNTER_TABLE)
+
+        name = 'ADD_TEST'
+        reason1 = 'L3_ANY'
+
+        self.create_drop_counter(name, SWITCH_INGRESS_DROPS)
+        self.add_drop_reason(name, reason1)
+        time.sleep(3)
+
+        # Verify that a counter has been created. We will verify the state of
+        # the counter in the next step.
+        assert len(asic_state_table.getKeys()) == 1
+        self.checkFlexState([SWITCH_STAT_BASE], SWITCH_DEBUG_COUNTER_LIST)
+
+        self.remove_drop_reason(name, reason1)
+        time.sleep(3)
+
+        # Verify that the drop counter has been added to ASIC DB, including the
+        # last reason that we attempted to remove.
+        asic_keys = asic_state_table.getKeys()
+        assert len(asic_keys) == 1
+        assert self.checkASICState(asic_keys[0], ASIC_COUNTER_SWITCH_IN_TYPE, [reason1])
+
+        # Cleanup for the next test.
+        self.delete_drop_counter(name)
+
+    def test_addDropReasonMultipleTimes(self, dvs, testlog):
+        """
+            This test verifies that the same drop reason can be added multiple
+            times without affecting the system.
+        """
+        self.setup_db(dvs)
+
+        asic_state_table = swsscommon.Table(self.asic_db, ASIC_STATE_TABLE)
+        flex_counter_table = swsscommon.Table(self.flex_db, FLEX_COUNTER_TABLE)
+
+        name = 'ADD_TEST'
+        reason1 = 'L3_ANY'
+
+        self.create_drop_counter(name, SWITCH_INGRESS_DROPS)
+        self.add_drop_reason(name, reason1)
+        time.sleep(3)
+
+        # Verify that a counter has been created. We will verify the state of
+        # the counter in the next step.
+        assert len(asic_state_table.getKeys()) == 1
+        self.checkFlexState([SWITCH_STAT_BASE], SWITCH_DEBUG_COUNTER_LIST)
+
+        reason2 = 'ACL_ANY'
+        self.add_drop_reason(name, reason2)
+        time.sleep(3)
+
+        # Verify that the drop counter has been added to ASIC DB, including the
+        # reason that was added.
+        asic_keys = asic_state_table.getKeys()
+        assert len(asic_keys) == 1
+        assert self.checkASICState(asic_keys[0], ASIC_COUNTER_SWITCH_IN_TYPE, [reason1, reason2])
+
+        self.add_drop_reason(name, reason2)
+        time.sleep(3)
+
+        # Verify that the ASIC state is the same as before adding the redundant
+        # drop reason.
+        asic_keys = asic_state_table.getKeys()
+        assert len(asic_keys) == 1
+        assert self.checkASICState(asic_keys[0], ASIC_COUNTER_SWITCH_IN_TYPE, [reason1, reason2])
+
+        # Cleanup for the next test.
+        self.delete_drop_counter(name)
+        self.remove_drop_reason(name, reason1)
+        self.remove_drop_reason(name, reason2)
+
+    def test_addInvalidDropReason(self, dvs, testlog):
+        """
+            This test verifies that adding a drop reason to a counter that is
+            not recognized will not affect the system.
+        """
+        self.setup_db(dvs)
+
+        asic_state_table = swsscommon.Table(self.asic_db, ASIC_STATE_TABLE)
+        flex_counter_table = swsscommon.Table(self.flex_db, FLEX_COUNTER_TABLE)
+
+        name = 'ADD_TEST'
+        reason1 = 'L3_ANY'
+
+        self.create_drop_counter(name, SWITCH_INGRESS_DROPS)
+        self.add_drop_reason(name, reason1)
+        time.sleep(3)
+
+        # Verify that a counter has been created. We will verify the state of
+        # the counter in the next step.
+        assert len(asic_state_table.getKeys()) == 1
+        self.checkFlexState([SWITCH_STAT_BASE], SWITCH_DEBUG_COUNTER_LIST)
+
+        reason2 = 'ACL_ANY'
+        self.add_drop_reason(name, reason2)
+        time.sleep(3)
+
+        # Verify that the drop counter has been added to ASIC DB, including the
+        # reason that was added.
+        asic_keys = asic_state_table.getKeys()
+        assert len(asic_keys) == 1
+        assert self.checkASICState(asic_keys[0], ASIC_COUNTER_SWITCH_IN_TYPE, [reason1, reason2])
+
+        dummy_reason = 'ZOBOOMBAFOO'
+        self.add_drop_reason(name, dummy_reason)
+        time.sleep(3)
+
+        # Verify that the ASIC state is the same as before adding the invalid
+        # drop reason.
+        asic_keys = asic_state_table.getKeys()
+        assert len(asic_keys) == 1
+        assert self.checkASICState(asic_keys[0], ASIC_COUNTER_SWITCH_IN_TYPE, [reason1, reason2])
+
+        # Cleanup for the next test.
+        self.delete_drop_counter(name)
+        self.remove_drop_reason(name, reason1)
+        self.remove_drop_reason(name, reason2)
+
+    def test_removeDropReasonMultipleTimes(self, dvs, testlog):
+        """
+            This test verifies that removing a drop reason multiple times will
+            not affect the system.
+        """
+        self.setup_db(dvs)
+
+        asic_state_table = swsscommon.Table(self.asic_db, ASIC_STATE_TABLE)
+        flex_counter_table = swsscommon.Table(self.flex_db, FLEX_COUNTER_TABLE)
+
+        name = 'ADD_TEST'
+        reason1 = 'L3_ANY'
+        reason2 = 'ACL_ANY'
+
+        self.create_drop_counter(name, SWITCH_INGRESS_DROPS)
+        self.add_drop_reason(name, reason1)
+        self.add_drop_reason(name, reason2)
+        time.sleep(3)
+
+        # Verify that a counter has been created. We will verify the state of
+        # the counter in the next step.
+        assert len(asic_state_table.getKeys()) == 1
+        self.checkFlexState([SWITCH_STAT_BASE], SWITCH_DEBUG_COUNTER_LIST)
+
+        self.remove_drop_reason(name, reason2)
+        time.sleep(3)
+
+        # Verify that the drop counter has been added to ASIC DB, excluding the
+        # reason that was removed.
+        asic_keys = asic_state_table.getKeys()
+        assert len(asic_keys) == 1
+        assert self.checkASICState(asic_keys[0], ASIC_COUNTER_SWITCH_IN_TYPE, [reason1])
+
+        self.remove_drop_reason(name, reason2)
+        time.sleep(3)
+
+        # Verify that the ASIC state is the same as before the redundant
+        # remove operation.
+        asic_keys = asic_state_table.getKeys()
+        assert len(asic_keys) == 1
+        assert self.checkASICState(asic_keys[0], ASIC_COUNTER_SWITCH_IN_TYPE, [reason1])
+
+        # Cleanup for the next test.
+        self.delete_drop_counter(name)
+        self.remove_drop_reason(name, reason1)
+
+    def test_removeNonexistentDropReason(self, dvs, testlog):
+        """
+            This test verifies that removing a drop reason that does not exist
+            on the device will not affect the system.
+        """
+        self.setup_db(dvs)
+
+        asic_state_table = swsscommon.Table(self.asic_db, ASIC_STATE_TABLE)
+        flex_counter_table = swsscommon.Table(self.flex_db, FLEX_COUNTER_TABLE)
+
+        name = 'ADD_TEST'
+        reason1 = 'L3_ANY'
+        reason2 = 'ACL_ANY'
+
+        self.create_drop_counter(name, SWITCH_INGRESS_DROPS)
+        self.add_drop_reason(name, reason1)
+        time.sleep(3)
+
+        # Verify the counter has been created and is in the correct state.
+        asic_keys = asic_state_table.getKeys()
+        assert len(asic_keys) == 1
+        assert self.checkASICState(asic_keys[0], ASIC_COUNTER_SWITCH_IN_TYPE, [reason1])
+
+        self.remove_drop_reason(name, reason2)
+        time.sleep(3)
+
+        # Verify that the ASIC state is unchanged after the nonexistent remove.
+        asic_keys = asic_state_table.getKeys()
+        assert len(asic_keys) == 1
+        assert self.checkASICState(asic_keys[0], ASIC_COUNTER_SWITCH_IN_TYPE, [reason1])
+
+        # Cleanup for the next test.
+        self.delete_drop_counter(name)
+        self.remove_drop_reason(name, reason1)
+
+    def test_removeInvalidDropReason(self, dvs, testlog):
+        """
+            This test verifies that removing a drop reason that is not recognized
+            will not affect the system.
+        """
+        self.setup_db(dvs)
+
+        asic_state_table = swsscommon.Table(self.asic_db, ASIC_STATE_TABLE)
+        flex_counter_table = swsscommon.Table(self.flex_db, FLEX_COUNTER_TABLE)
+
+        name = 'ADD_TEST'
+        reason1 = 'L3_ANY'
+        bogus_reason = 'LIVE_LAUGH_LOVE'
+
+        self.create_drop_counter(name, SWITCH_INGRESS_DROPS)
+        self.add_drop_reason(name, reason1)
+        time.sleep(3)
+
+        # Verify the counter has been created and is in the correct state.
+        asic_keys = asic_state_table.getKeys()
+        assert len(asic_keys) == 1
+        assert self.checkASICState(asic_keys[0], ASIC_COUNTER_SWITCH_IN_TYPE, [reason1])
+
+        self.remove_drop_reason(name, bogus_reason)
+        time.sleep(3)
+
+        # Verify that the ASIC state is unchanged after the bad remove.
+        asic_keys = asic_state_table.getKeys()
+        assert len(asic_keys) == 1
+        assert self.checkASICState(asic_keys[0], ASIC_COUNTER_SWITCH_IN_TYPE, [reason1])
+
+        # Cleanup for the next test.
+        self.delete_drop_counter(name)
+        self.remove_drop_reason(name, reason1)
+
+    def test_createAndDeleteMultipleCounters(self, dvs, testlog):
+        """
+            This test verifies that creating and deleting multiple drop counters
+            at the same time works correctly.
+        """
+        self.setup_db(dvs)
+
+        asic_state_table = swsscommon.Table(self.asic_db, ASIC_STATE_TABLE)
+        flex_counter_table = swsscommon.Table(self.flex_db, FLEX_COUNTER_TABLE)
+
+        name1 = 'DEBUG_0'
+        reason1 = 'L3_ANY'
+
+        name2 = 'DEBUG_1'
+        reason2 = 'ACL_ANY'
+
+        self.create_drop_counter(name1, PORT_INGRESS_DROPS)
+        self.add_drop_reason(name1, reason1)
+
+        self.create_drop_counter(name2, PORT_INGRESS_DROPS)
+        self.add_drop_reason(name2, reason2)
+
+        time.sleep(5)
+
+        # Verify that the flex counters are correctly tracking two different
+        # drop counters.
+        self.checkFlexState([PORT_STAT_BASE, PORT_STAT_INDEX_1], PORT_DEBUG_COUNTER_LIST)
+
+        # Verify that there are two entries in the ASIC DB, one for each counter.
+        asic_keys = asic_state_table.getKeys()
+        assert len(asic_keys) == 2
+        for key in asic_keys:
+            assert (self.checkASICState(key, ASIC_COUNTER_PORT_IN_TYPE, [reason1]) or self.checkASICState(key, ASIC_COUNTER_PORT_IN_TYPE, [reason2]))
+
+        self.delete_drop_counter(name2)
+        self.remove_drop_reason(name2, reason2)
+        time.sleep(3)
+
+        # Verify that the flex counters are tracking ONE drop counter after
+        # the update.
+        self.checkFlexState([PORT_STAT_BASE], PORT_DEBUG_COUNTER_LIST)
+
+        # Verify that there is ONE entry in the ASIC DB after the update.
+        asic_keys = asic_state_table.getKeys()
+        assert len(asic_keys) == 1
+        assert self.checkASICState(asic_keys[0], ASIC_COUNTER_PORT_IN_TYPE, [reason1])
+
+        # Cleanup for the next test.
+        self.delete_drop_counter(name1)
+        self.remove_drop_reason(name1, reason1)