ReliableCollections Service Fabric: Statemanager's GetOrAddAsync() returns invalid queue with different names

Question

I have two reliable queues and they are being accessed by two guest executables and each of them access their own. Sometimes the function I use to access them doesn't update the reliable queue object in the function and the wrong request is sent to the wrong guest executable.

What happens is that the clientId is passed by the guest executable to this function in the Get request. Let us say that there are two clientId(s) called T1 and T2.

What happens is that the guest executable (client) T2 at times gets the request that was meant for T1. Even though I tried line by line debugging the parameters passed to this function are correct. Here is my API's POST that is passed a json to be added to the queue for the clients to receive from the GET

[HttpPost("MarketInfo")]
public JObject GetMarketInfo([FromBody] JObject jObject)
{
            List<JToken> clients = jObject.GetValue("clients").ToList();
            string json;
            JObject response = new JObject();
            JArray jsonArray = new JArray();

            try
            {
                foreach (JToken client in clients)
                {
                    var id = Guid.NewGuid();
                    json = "{'name':'MarketInfo','id':'" + id.ToString() + "','mtClientId':'" + terminal["name"].ToString() + "','parameters':{'symbol':'" + terminal["symbol"].ToString() + "','property':24}}";
                    bool result = _requestsCollectionHandler.CreateRequestForClient(JObject.Parse(json));
                    JObject clientResponse = new JObject();
                    if (result==true)
                    {
                        clientResponse["name"] = client["name"].ToString();
                        clientResponse["guid"] = id.ToString();
                        jsonArray.Add(clientResponse);
                    }
                    else
                    {
                        clientResponse["name"] = terminal.Children()["name"].ToString();
                        clientResponse["guid"] = "ERROR";
                        jsonArray.Add(terminalResponse);
                    }
                }
                response["clients"] = jsonArray;
                return response;
            }
            catch (Exception e)
            {
                Debug.Write(e.Message);
                return null;
            }
}

This is the json that we pass to this API

{"clients":[{"name":"T1","symbol":"SomeInfo"},{"name":"T2","symbol":"SomeInfo"}]}

The problem is always with the clients object that is passed first. Before I explain further let me also share the code for the client's HttpGet

[HttpGet("{clientId}")]
public string Get([FromRoute] string clientId)
{
            try
            {
                string request = _requestsCollectionHandler.GetRequestJsonFromQueue(clientId);
                return request;
            }
            catch(Exception e)
            {
                return e.Message;
            }
}

This is the function that creates an object that is to be added by another function in the reliable queue

public bool CreateRequestForClient(JObject jObject)
{
    try
    {
        this._jObject = new JObject(jObject);
        CreateKey();
        AddToRequestToQueueAsync();
        return true;
    }
    catch (Exception e)
    {
        Debug.Write(e.Message);
        _exceptionMessage = e.Message;
        return false;
    }
}
    private void CreateKey()
    {
        dynamic data = JObject.Parse(_jObject.ToString(Newtonsoft.Json.Formatting.None));
        string name = data.name;
        string id = data.id;
        string clientId = data.clientId;
        _key.id = id;
        _key.name = name;
        _key.clientId = clientId;
        //key.timestamp = GetTimestamp();
        _key.timestamp = GetTimestamp();
        _key.requestJson = _jObject.ToString(Newtonsoft.Json.Formatting.None);
    }

_key is a private variable in class a custom class

This is the function in my class of request handler that adds the requests to the queue

private void AddToRequestToQueueAsync()
{
            var transaction = this._stateManager.CreateTransaction();
            CancellationToken cancellationToken
                 = new CancellationToken(false);
            try
            {
                string queue = _key.clientId;
                IReliableConcurrentQueue<TerminalResponseKey> reliableQueue =
                    _stateManager.GetOrAddAsync<IReliableConcurrentQueue<TerminalResponseKey>>(queue).Result;
                transaction = this._stateManager.CreateTransaction();
                if (reliableQueue!=null)
                {

                    long count = reliableQueue.Count;
                    reliableQueue.EnqueueAsync(transaction, _key);
                    count = reliableQueue.Count;
                    transaction.CommitAsync().Wait();
                }
                else
                {
                    transaction.Abort();
                }
            }
            catch
            {
                transaction.Abort();
                throw;
            }
}

This is function that is used by the client

public string GetRequestJsonFromQueue(string clientId)
{
string queue = clientId;
 try
            {
                IReliableConcurrentQueue<TerminalResponseKey> reliableQueue = 
                    this._stateManager.GetOrAddAsync<IReliableConcurrentQueue<TerminalResponseKey>>(queue).Result;
                if(reliableQueue != null)
                {
                    ConditionalValue<TerminalResponseKey> key = 
                        reliableQueue.TryDequeueAsync(transaction).Result;
                    if(key.HasValue)
                    {
                        string request = key.Value.requestJson;
                        transaction.CommitAsync().Wait();
                        return request;
                    }
                }
                else
                {
                    transaction.Abort();
                }
                return "NO QUEUE";
            }
            catch (Exception e)
            {
                Debug.WriteLine(e);
                transaction.Abort();
                return e.InnerException.Message;
            }
}

As far as I have found out I think my problem is in this function above. Because I don't know how the client T2 or client T1 gets another client's queue because the parameters determining the queue are their IDs and are totally unique. These Ids are also passed correctly to this:

IReliableConcurrentQueue<TerminalResponseKey> reliableQueue = 
                        this._stateManager.GetOrAddAsync<IReliableConcurrentQueue<TerminalResponseKey>>(queue).Result;

As you can see that we have queue=clientId I have tried adding proper timespans but it was of no use as there is no exception thrown for OperationTimedOut. Furthermore since I am new to ServiceFabric I maybe totally doing anything wrong.

PS: Sorry for maybe a lot of jumbled up and confused code and question AND SOME OF THE INFORMATION IS OBFUSCATED DUE TO CONFIDENTIALITY BUT NOTHING OBSTRUCTING THE UNDERSTANDING OF THIS IS HIDDEN (I Hope not an issue)

I hope this is not an issue maybe an error I am overlooking at my side

Answer 1

When you put the request in the queue, in AddToRequestToQueueAsync(), the name of the queue is set from _key.terminalId (and I don't see where you assign it), but when you read from it, in GetRequestJsonFromQueue(), the clientId is used as the queue name.