Move channel publish call into consumer_state_table_pops.lua to make them atomic.

Signed-off-by: Jipan Yang <[email protected]>

Author: jipanyang

common/consumer_state_table_pops.lua

@@ -20,4 +20,12 @@ for i = 1, n do
    -- Clean up the key in temporary state table
    redis.call('DEL', stateprefix..tablename..key)
 end
+
+-- Check whether the keyset is empty, if not, signal ourselves to process again.
+-- This is to handle the case where the number of keys in keyset is more than POP_BATCH_SIZE
+local num = redis.call('SCARD', KEYS[1])
+if num > 0 then
+   redis.call('PUBLISH', ARGV[3], 'G')
+end
+
 return ret

common/consumerstatetable.cpp

@@ -34,13 +34,6 @@ ConsumerStateTable::ConsumerStateTable(DBConnector *db, const std::string &table
         "    redis.call('PUBLISH', KEYS[1], ARGV[1])\n"
         "end\n";
     m_multiPublish = loadRedisScript(m_db, luaMultiPublish);
-
-   std::string luaScardPublish =
-        "local num = redis.call('SCARD', KEYS[2])\n"
-        "if num > 0 then\n"
-        "    redis.call('PUBLISH', KEYS[1], ARGV[1])\n"
-        "end\n";
-    m_scardPublish = loadRedisScript(m_db, luaScardPublish);
 }
 
 void ConsumerStateTable::pops(std::deque<KeyOpFieldsValuesTuple> &vkco, const std::string& /*prefix*/)
@@ -51,30 +44,19 @@ void ConsumerStateTable::pops(std::deque<KeyOpFieldsValuesTuple> &vkco, const st
 
     RedisCommand command;
     command.format(
-        "EVALSHA %s 3 %s %s: %s %d %s",
+        "EVALSHA %s 3 %s %s: %s %d %s %s",
         sha.c_str(),
         getKeySetName().c_str(),
         getTableName().c_str(),
         getDelKeySetName().c_str(),
         POP_BATCH_SIZE,
-        getStateHashPrefix().c_str());
+        getStateHashPrefix().c_str(),
+        getChannelName().c_str());
 
     RedisReply r(m_db, command);
     auto ctx0 = r.getContext();
     vkco.clear();
 
-    // Check whether the keyset is empty, if not, signal ourselves to process again.
-    // This is to handle the case where the number of keys in keyset is more than POP_BATCH_SIZE
-    // Note there is possibility of false positive since this call is not atomic with consumer_state_table_pops.lua
-    // Putting publish call inside consumer_state_table_pops.lua doesn't seem to work.
-    command.format(
-        "EVALSHA %s 2 %s %s %s",
-        m_scardPublish.c_str(),
-        getChannelName().c_str(),
-        getKeySetName().c_str(),
-        "G");
-    RedisReply r2(m_db, command);
-
     // if the set is empty, return an empty kco object
     if (ctx0->type == REDIS_REPLY_NIL)
     {

common/consumerstatetable.h

@@ -20,7 +20,6 @@ class ConsumerStateTable : public ConsumerTableBase, public TableName_KeySet
 
 private:
     std::string m_multiPublish;
-    std::string m_scardPublish;
 };
 
 }