Rest api Read Timed Out for large dataset

Question

I have a rest request (user initiated via a lwc button) to an Azure Function that gets sent a masterId and followerId which contacts are assigned to. The Azure function uses a SOSL query. It will send a response to Salesforce to process a batch class which will merge and then update contact records.

The request is successful if the number of contact records is below 8k but will fail with a 'System Exception: Read Timed Out' when sending more than 8k contact records.

Unfortunately, I can't move the contacts temporarily to a different masterId or followerId in order to lower the number of contacts to below the 8K threshold for each request.

Since more than 8K records exceeds the 2 minute request time limit, how would I limit the number of contact records that are sent with each request?

Any guidance and examples would be appreciated. Thanks!

This is the HTTP request:

public static Boolean merge(String masterId, String followerId){
String isRunningInSandbox = String.valueOf(Utils.runningInASandbox()).substring(0,1).toUpperCase() + String.valueOf(Utils.runningInASandbox()).substring(1,String.valueOf(Utils.runningInASandbox()).length());      
JSONGenerator gen = JSON.createGenerator(true);    
gen.writeStartObject();
gen.writeStringField('is_sandbox', isRunningInSandbox);      
gen.writeStringField('master_territory',masterId);
gen.writeStringField('follower_territory',followerId);
gen.writeEndObject();
String body = gen.getAsString();
String endpoint = 'https://myAzureFunction.azurewebsites.net/api/myAzureFunction?code=Kfhauf62894jhfjrfhjfhs81162Test==';
HttpRequest req = new HttpRequest();
req.setEndpoint(endpoint);
req.setMethod('POST');
req.setbody(body);
req.setHeader('x-functions-key', 'pppq/8shgt/UQHUskfh239vnjewh21POPTEST==');
Http http = new Http();
req.setTimeout(120000);
HTTPResponse response;
  if (!Test.isRunningTest()){
    response = http.send(req);
    if (response.getStatusCode() == 500) {
      return False;
    }
  }       
return True;
}

import logging

import azure.functions as func

from simple_salesforce import SalesforceAPI
from functools import total_ordering
from json import JSONEncoder

import datetime
import json
import math
import os

@total_ordering
class TerritoryMergeContact(JSONEncoder):

    accountsProtectedMap = {'A Account’: '0013600000OXfZSAA1',
                            ‘B Account’: '0013600000OXeNsAAL',
                            'C Account’: '0013600000OXdolAAD',
                            ‘D Account': '0013600000OXdp0AAD',
                            'E Account': '0013600000OXdonAAD',
                            ‘F Account': '0013600000xTmVlAAK',
                            ‘G Account': '0013600000OXeOAAA1',
                            ‘H Account': '0013600000OXeHdAAL',
                            ‘I Account': '0013600000OXe1XAAT',
                            ‘J Account': '0013600000OXeOMAA1',
                            ‘K Account': '0013600000ut9g2AAA',
                            ‘L Account': '0013600000OXegzAAD',
                            ‘M Account': '0013600000OXdpGAAT',
                            ’N Account': '0013600000OXeHaAAL',
                            ‘O Account': '0013600000OXdpJAAT',
                            ‘P Account': '0013600000OXdoUAAT',
                            ‘Q Account': '0013600000OXdpQAAT',
                            ‘R Account': '0013600000OXeLCAA1',
                            ’S Account': '0013600000OXdpkAAD'}

    def __init__(self, c):
        self.contact = c
        self.isGrouped = False
        self.keyA = self.findKey(c['LastName'], c['CRD__c'])
        self.keyB = self.findKey(c['LastName'], c['NPN__c'])
        self.keyC = self.findKey(c['FirstName'], c['LastName'], c['Email'])
        self.keyC2 = self.findKey(c['FirstName'], c['LastName'], c['Other_Email__c'])
        self.keyC3 = self.findKey(c['FirstName'], c['LastName'], c['Personal_Email__c'])
        self.keyD = self.findKey(c['FirstName'], c['LastName'], c['Phone'])
        self.keyD2 = self.findKey(c['FirstName'], c['LastName'], c['OtherPhone'])
        self.keyD3 = self.findKey(c['FirstName'], c['LastName'], c['MobilePhone'])
        self.keyE = self.findKey(c['CRD__c'])
        self.keyF = self.findKey(c['NPN__c'])
        self.keyG = self.findKey(c['FirstName'], c['LastName'],c['MailingCity'], c['MailingState'], c['MailingPostalCode'])
        self.level = self.getLevel(c)

        self.noMergeBD = False
        self.isNonMergeTable = False
        self.mergeRule = None
        self.isMaster = False
        if (c['Broker_Dealer__c'] in self.accountsProtectedMap.values()):
            #             print(c['LastName'], c['Broker_Dealer__c'])
            self.noMergeBD = True
            self.isNonMergeTable = True
            self.mergeRule = 'No-Merge BD'

    def findKey(self, *args):
        #         print(args)
        key = ''
        for s in args:
            if s != None:
                key = key + s
            else:
                return None
        return key

    def getLevel(self, c):
        level = 15
        if c['Retention_Stage__c']!=None and (c['Retention_Stage__c'][:1]=='3'):
            level = 1
        elif c['Retention_Stage__c']!=None and (c['Retention_Stage__c'][:1]=='2'):
            level = 2
        elif c['Retention_Stage__c']!=None and (c['Retention_Stage__c'][:1]=='1'):
            level = 3
        elif c[‘Company_Contact_ID__c']!=None and c['Contact_Stage__c']!=None and (c['Contact_Stage__c'][:1]=='5'):
            level = 4
        elif c['Retention_Stage__c']!=None and (c['Retention_Stage__c'][:1]=='4'):
            level = 5
        elif c['Retention_Stage__c']!=None and (c['Retention_Stage__c'][:1]=='5'):
            level = 6
        elif c['Contact_Stage__c']!=None and (c['Contact_Stage__c'][:1]=='5'):
            level = 7
        elif c['Contact_Stage__c']!=None and (c['Contact_Stage__c'][:1]=='4'):
            level = 8
        elif c['Contact_Stage__c']!=None and (c['Contact_Stage__c'][:1]=='3'):
            level = 9
        elif c['List_Source__c']!=None:
            level = 10
        elif c['Contact_Stage__c']!=None and (c['Contact_Stage__c'][:1]=='2'):
            level = 11
        elif c['Conference_Tracking__c']!=None:
            level = 12
        elif c['CreatedDate'][:4] == datetime.datetime.now().year:
            level = 13
        elif c['CreatedDate'][:4] == datetime.datetime.now().year-1:
            level = 14
        return level

    def __lt__(self, other):
        if int(self.level) - other.level < 0:
            return True
        elif int(self.level) - other.level > 0:
            return False
        else:
            if 'List_Source__c' in self.contact and 'List_Source__c' not in other.contact:
                return True
            elif 'List_Source__c' in other.contact and 'List_Source__c' not in self.contact:
                return False
            else:
                if 'Conference_Tracking__c' in self.contact and 'Conference_Tracking__c' not in other.contact:
                    return True
                elif 'Conference_Tracking__c' in other.contact and 'Conference_Tracking__c' not in self.contact:
                    return False
                else:
                    return (self.contact['CreatedDate'] >= other.contact['CreatedDate'])

    def default(self, o):
        return o.__dict__

    def reprJSON(self):
        return self.__dict__


class ContactGroupWrapper(JSONEncoder):
    def __init__(self, tmcList, groupingKey):
        self.tmcList = tmcList
        self.groupingKey = groupingKey

    def default(self, o):
        return o.__dict__

    def reprJSON(self):
        return dict(tmcList=self.tmcList, groupingKey=self.groupingKey)


class ComplexEncoder(json.JSONEncoder):
    def default(self, obj):
        if hasattr(obj, 'reprJSON'):
            return obj.reprJSON()
        else:
            return json.JSONEncoder.default(self, obj)


def run(username, password, token, is_sandbox, master_id, follower_id):

    sf = SalesforceAPI(username, password, token, is_sandbox)    
    qry = 'SELECT Id, Name, Broker_Dealer__c, Contact_Stage__c, Retention_Stage__c, Conference_Tracking__c, CreatedDate, RecordType.Name, Company_Contact_ID__c, List_Source__c, Phone, OtherPhone, MobilePhone, CRD__c, FirstName, LastName, Email, Other_Email__c, Personal_Email__c, NPN__c, Territory__r.Name, MailingCity, MailingState, MailingPostalCode FROM Contact WHERE Territory__c IN (\'%s\', \'%s\') AND Contact_Stage__c!=\'%s\'' % (master_id, follower_id, 'Do Not Contact') 
    # print(qry)
    contacts = sf.query_all(qry)
    # print('RECORDS:', len(contacts['records']))

    territoryMasterName = contacts['records'][0]['Territory__r']['Name']
    tmcList = []
    for c in contacts['records']:
        tmc = TerritoryMergeContact(c)
        tmcList.append(tmc)

    groupNumber = 1
    groupingMap = {}  # Id, Integer - Id of Contact, groupingKey
    cgwMap = {}       # Integer, List<ContactGroupWrapper>

    for key in ['keyA', 'keyB', 'keyC', 'keyC2', 'keyC3', 'keyD', 'keyD2', 'keyD3', 'keyE', 'keyF', 'keyG']:
        for i in range(0, len(tmcList)-1):
            sourceContact = tmcList[i]
            sourceKey = getattr(sourceContact, key)
            if sourceKey == None:
                continue
            for j in range(i+1, len(tmcList)):
                # Contact we want to match against
                targetContact = tmcList[j]
                targetKey = getattr(targetContact, key)
                if targetKey == None or targetContact.isGrouped == True:
                    continue
                if targetKey == sourceKey:
                    if sourceContact.contact['Id'] in groupingMap.keys():
                        groupingMap[targetContact.contact['Id']] = groupingMap[sourceContact.contact['Id']]
                        cgwMap[groupingMap[sourceContact.contact['Id']]].tmcList.append(targetContact)
                    elif targetContact.contact['Id'] in groupingMap.keys():
                        groupingMap[sourceContact.contact['Id']] = groupingMap[targetContact.contact['Id']]
                        cgwMap[groupingMap[targetContact.contact['Id']]].tmcList.append(sourceContact)
                    else:
                        groupingMap[sourceContact.contact['Id']] = groupNumber
                        groupingMap[targetContact.contact['Id']] = groupNumber
                        cgwMap[groupNumber] = ContactGroupWrapper([sourceContact, targetContact], int(groupNumber))
                    groupNumber += 1
                    sourceContact.isGrouped = True
                    targetContact.isGrouped = True

    cgwList = list(cgwMap.values())
    i = 0
    contactsToUpdate = []
    while i < len(cgwList):
        cgw = cgwList[i]
        numContracted = 0
        isMultipleContracted = False
        for tmc in cgw.tmcList:
            if tmc.level <= 6:
                numContracted += 1
                if numContracted == 2:
                    isMultipleContracted = True
                    break
        if isMultipleContracted == True:
            for tmc in cgw.tmcList:
                tmc.isNonMergeTable = True
                tmc.mergeRule = 'Multiple Contracted Producers'
                tmc.contact['Unique_BD_Rep_ID__c'] = str(
                    format(cgw.groupingKey, '05d'))+'/'+str(tmc.level)+'/'+tmc.mergeRule
                contactsToUpdate.append(tmc.contact)
            del cgwList[i]
            continue
        else:
            # Update no merge db unique bd rep id
            isNoMerge = False
            for tmc in cgw.tmcList:
                if tmc.isNonMergeTable == True and tmc.noMergeBD == True:
                    isNoMerge = True
                    break
            if isNoMerge == True:
                for tmc in cgw.tmcList:
                    tmc.mergeRule = 'No-Merge BD'
                    tmc.contact['Unique_BD_Rep_ID__c'] = str(
                        format(cgw.groupingKey, '05d'))+'/'+str(tmc.level)+'/'+tmc.mergeRule
                    contactsToUpdate.append(tmc.contact)
                del cgwList[i]
                continue
        i += 1
    for tmc in tmcList:
        if tmc.isNonMergeTable == False and tmc.noMergeBD == True:
            tmc.mergeRule = 'Step 15'
            tmc.contact['Unique_BD_Rep_ID__c'] = '/00000/' + \
                str(tmc.level)+'/'+tmc.mergeRule
            contactsToUpdate.append(tmc.contact)

    for cgw in cgwList:
        cgw.tmcList.sort()
        cgw.tmcList[0].isMaster = True
        cgw.tmcList[0].mergeRule = 'Master Merged'
        for j in range(1, len(cgw.tmcList)):
            cgw.tmcList[j].isMaster = False

    # print(len(cgwList))

    x = 0
    x_step = 500

    total_batches = math.ceil(len(cgwList)/x_step)
    n = 1
    while x < len(cgwList):
        if x+x_step <= len(cgwList):
            x2 = x+x_step
        else:
            x2 = len(cgwList)
        isFinal = (n == max(total_batches-1, 1))
        payload = {
            "contactsToUpdate": [],
            "territoryMasterName": territoryMasterName,
            "cgwListString": json.dumps([x.reprJSON() for x in cgwList[x:x2]], cls=ComplexEncoder),
            "isFinal": isFinal
        }
        result = sf.apexecute('Territory/Merge', method='POST', data=payload)
        # print(result)
        x = x2
        n += 1
    y = 0
    y_step = 400
    while y < len(contactsToUpdate):
        if y+y_step <= len(contactsToUpdate):
            y2 = y+y_step
        else:
            y2 = len(contactsToUpdate)
        payload = {
            "contactsToUpdate": contactsToUpdate[y:y2],
            "territoryMasterName": territoryMasterName,
            "cgwListString": ''
        }
        result = sf.apexecute('Territory/UpdateContacts', method='POST', data=payload)
        # print(result)
        y = y2
    return str(len(cgwList))


def main(req: func.HttpRequest) -> func.HttpResponse:
    logging.info('Python HTTP trigger function processed a request.')

    req_body = req.get_json()
    is_sandbox = req_body.get('is_sandbox')
    if is_sandbox == 'True':
        username = os.environ['SFDC_UAT_USERNAME']
        password = os.environ['SFDC_UAT_PASSWORD']
        token = os.environ['SFDC_UAT_TOKEN']
    else:
        username = os.environ['SFDC_PROD_USERNAME']
        password = os.environ['SFDC_PROD_PASSWORD']
        token = os.environ['SFDC_PROD_TOKEN']
    master_territory = req_body.get('master_territory')
    follower_territory = req_body.get('follower_territory')
    try:
        result = run(username, password, token, is_sandbox, master_territory, follower_territory)
        return func.HttpResponse(result)
    except:
        return func.HttpResponse(
            "Please pass is_sandbox, master_territory, follower_territory. Also check authentication.",
            status_code=400
        )

Answer 1

There's a lot going on here, so I don't think I can share a comprehensive or conclusive answer, but I think the key is that run() has to complete before your Azure function returns a status:

result = run(username, password, token, is_sandbox, master_territory, follower_territory)
return func.HttpResponse(result)

Then, in run(), you perform a SOQL query that (as you say) returns > 8,000 records:

qry = 'SELECT Id, Name, Broker_Dealer__c, Contact_Stage__c, Retention_Stage__c, Conference_Tracking__c, CreatedDate, RecordType.Name, Company_Contact_ID__c, List_Source__c, Phone, OtherPhone, MobilePhone, CRD__c, FirstName, LastName, Email, Other_Email__c, Personal_Email__c, NPN__c, Territory__r.Name, MailingCity, MailingState, MailingPostalCode FROM Contact WHERE Territory__c IN (\'%s\', \'%s\') AND Contact_Stage__c!=\'%s\'' % (master_id, follower_id, 'Do Not Contact') 
# print(qry)
contacts = sf.query_all(qry)

You perform complex processing on the returned result, including a triply-nested for loop that I suspect is at least O(n^2). Then, you walk through the list of items to process several more times, ultimately invoking two Apex classes via REST, synchronously:

    result = sf.apexecute('Territory/Merge', method='POST', data=payload) 
    result = sf.apexecute('Territory/UpdateContacts', method='POST', data=payload)

The whole time, Salesforce is waiting for your function to return an HTTP status. It's just too much work to do in a single synchronous callout, at that record volume.

I am not an Azure expert, but at the broadest level I see a few options for slimming this down:

• Profile the entire process and see if you can identify the primary time costs. If those Apex web services, for example, are absurdly slow, can you rebuild their logic in Python to gain time savings and call the native REST API directly? Maybe not enough to cover 8,000 records in a single invocation but likely a worthwhile effort.
• Add a LIMIT clause to your query. Experimentation should help determine what's a "safe" threshold for processing in a single request. Return a status code to Salesforce to advise the caller that partial processing has been completed and to send another request to process more records.

  Since you're doing merge operations, it's not clear that this would yield correct results, or that the process can meaningfully be chunked in this way.

• Rebuild your off-platform process to return a success immediately upon receipt of a valid request. Handle the processing asynchronously, and call in to Salesforce to report any errors. (These could be surfaced to users in a variety of ways depending on what the UI is for this whole process - potentially, for example, you could monitor an Error__c object with the Streaming API from a Lightning component or Visualforce page, while a user is monitoring job progress).
• Rebuild your off-platform process to use a regular scheduled job so that your one-off requests are lower in scale.

You also asked about using the Bulk API. I don't think that will avail you here, since you're not directly calling Salesforce's REST API at all (just your own Apex web services). The Bulk API is also inherently asynchronous and can take quite a while to complete, so if anything it's likely to make your current problem both worse and less predictable.