Fix the issue of ignoring callback calls for removed keys.

common/configdb.h

@@ -105,8 +105,11 @@ class ConfigDBConnector_Native : public SonicV2Connector_Native
                     try:
                         (table, row) = key.split(self.TABLE_NAME_SEPARATOR, 1)
                         if table in self.handlers:
-                            client = self.get_redis_client(self.db_name)
-                            data = self.raw_to_typed(client.hgetall(key))
+                            if item['data'] == 'del':
+                                data = None
+                            else:
+                                client = self.get_redis_client(self.db_name)
+                                data = self.raw_to_typed(client.hgetall(key))
                             if table in init_data and row in init_data[table]:
                                 cache_hit = init_data[table][row] == data
                                 del init_data[table][row]

tests/test_redis_ut.py

@@ -634,29 +634,40 @@ def thread_coming_entry():
 def test_ConfigDBInit():
     table_name_1 = 'TEST_TABLE_1'
     table_name_2 = 'TEST_TABLE_2'
+    table_name_3 = 'TEST_TABLE_3'
     test_key = 'key1'
     test_data = {'field1': 'value1'}
-    test_data_update = {'field1': 'value2'}
+
+    queue = multiprocessing.Queue()
 
     manager = multiprocessing.Manager()
     ret_data = manager.dict()
 
-    def test_handler(table, key, data, ret):
-        ret[table] = {key: data}
+    def test_handler(table, key, data, ret, q=None):
+        if data is None:
+            ret[table] = {k: v for k, v in ret[table].items() if k != key}
+        else:
+            ret[table] = {key: data}
+
+        if q:
+            q.put(ret[table])
 
-    def test_init_handler(data, ret):
+    def test_init_handler(data, ret, queue):
         ret.update(data)
+        queue.put(ret)
 
-    def thread_listen(ret):
+    def thread_listen(ret, queue):
         config_db = ConfigDBConnector()
         config_db.connect(wait_for_init=False)
 
         config_db.subscribe(table_name_1, lambda table, key, data: test_handler(table, key, data, ret),
                             fire_init_data=False)
         config_db.subscribe(table_name_2, lambda table, key, data: test_handler(table, key, data, ret),
                             fire_init_data=True)
+        config_db.subscribe(table_name_3, lambda table, key, data: test_handler(table, key, data, ret, queue),
+                            fire_init_data=False)
 
-        config_db.listen(init_data_handler=lambda data: test_init_handler(data, ret))
+        config_db.listen(init_data_handler=lambda data: test_init_handler(data, ret, queue))
 
     config_db = ConfigDBConnector()
     config_db.connect(wait_for_init=False)
@@ -666,14 +677,26 @@ def thread_listen(ret):
     # Init table data
     config_db.set_entry(table_name_1, test_key, test_data)
     config_db.set_entry(table_name_2, test_key, test_data)
+    config_db.set_entry(table_name_3, test_key, {})
 
-    thread = multiprocessing.Process(target=thread_listen, args=(ret_data,))
+    thread = multiprocessing.Process(target=thread_listen, args=(ret_data, queue))
     thread.start()
-    time.sleep(5)
-    thread.terminate()
 
-    assert ret_data[table_name_1] == {test_key: test_data}
-    assert ret_data[table_name_2] == {test_key: test_data}
+    init_data = queue.get(5)
+
+    # Verify that all tables initialized correctly
+    assert init_data[table_name_1] == {test_key: test_data}
+    assert init_data[table_name_2] == {test_key: test_data}
+    assert init_data[table_name_3] == {test_key: {}}
+
+    # Remove the entry (with no attributes) from the table.
+    # Verify that the update is received and a callback is called
+    config_db.set_entry(table_name_3, test_key, None)
+
+    table_3_data = queue.get(5)
+    assert test_key not in table_3_data
+
+    thread.terminate()
 
 
 def test_DBConnectFailure():