[SalesForce] Sporadic Bulk API error with PKChunking – Failed to create batch, since the Job is not Open

Sometime when retrieving the batch results for a Bulk API query job submitted with PK chunking enabled, I get the following error message:

PKChunking failed. InvalidJobState : Failed to create batch, since the
Job is not Open. Current job state is 'Closed'

I have used Bulk queries without PK chunking many times and have not seen that error. Even with PK chunking enabled, most of the time it works correctly i.e. all batches are submitted and completed.

This is what I'm doing to submit the job:

create a job with PK Chunking header using the instructions in the Bulk API documentation
submit a batch containing the query
after getting server response from the batch submit, close the job immediately

Am I missing any step e.g. do I need to do anything like check batch status before closing the job? Is it even necessary to close Bulk jobs if they are query jobs?

Best Answer

Very simple solution to this - don't close the job until the query has completed. When I was running jobs without PK Chunking enabled I was always closing the job immediately after submitting jobs, and that works fine, but when PK Chunking is enabled it seems that more time is needed for the queries to submit.

So now I'm

creating a job with PK Chunking header
submit a batch containing the query
polling for job to complete
downloading results
closing the job

Related Solutions

[SalesForce] Cannot close job with Salesforce Bulk API

It turns out that you have to remove all whitespace from the request, so if you pass this, it will work:

<?xml version="1.0" encoding="UTF-8"?><jobInfo xlmns="http://www.force.com/2009/06/asyncapi/dataload"><state>Closed</state></jobInfo>

I had the idea from a developer forum post on a different API where one of the Salesforce people mentioned that adding extra whitespace to the xml could cause a request to fail (and they said they would fix it).

[SalesForce] Bulk Query with PK Chunking enabled sometimes does not return all the batches

I believe that I have found the solution. Close the bulk job at the end of the processing.

Keep a count of how many batches are extracted, for example in the variable "finalBatchCount" Now once all the batches are read, just before closing the Bulk Query Job, call the "bulkConnection.getBatchInfoList(job.getId()).getBatchInfo().length" again, and if this is less then "finalBatchCount", then get the records from the rest of the batches. Once all the batches are read, close the Bulk Query Job

Looping this way will make sure that all the batches have been read. Attached is the code for reference

import java.io.*;
import java.util.*;
import com.sforce.async.*;
import com.sforce.soap.partner.PartnerConnection;
import com.sforce.ws.ConnectionException;
import com.sforce.ws.ConnectorConfig;

public class BulkQueryDirect {


public static void main(String[] args)
  throws AsyncApiException, ConnectionException, IOException {
    BulkQueryDirect exampleQuery = new BulkQueryDirect();
    System.setProperty("https.protocols", "TLSv1.2");
    BulkConnection bulkConnection = exampleQuery.login();
    bulkConnection.addHeader("Sforce-Enable-PKChunking","chunkSize=200");
    exampleQuery.doBulkQuery(bulkConnection);
}


public BulkConnection login() {

      String userName = "abc@abc.com";
      String passWord = "passwordSecuritytoken";
      String url = "https://login.salesforce.com/services/Soap/u/32.0";
      BulkConnection _bulkConnection = null;
      try {

            ConnectorConfig partnerConfig = new ConnectorConfig();
            partnerConfig.setUsername(userName);
            partnerConfig.setPassword(passWord);
            partnerConfig.setAuthEndpoint(url);
            new PartnerConnection(partnerConfig);
            ConnectorConfig config = new ConnectorConfig();
            config.setSessionId(partnerConfig.getSessionId());
            String soapEndpoint = partnerConfig.getServiceEndpoint();
            String apiVersion = "32.0";
            String restEndpoint = soapEndpoint.substring(0, soapEndpoint.indexOf("Soap/"))
                + "async/" + apiVersion;
            config.setRestEndpoint(restEndpoint);
            config.setCompression(true);
            config.setTraceMessage(false);
            _bulkConnection = new BulkConnection(config);
      } catch (AsyncApiException aae) {
        aae.printStackTrace();
      } catch (ConnectionException ce) {
        ce.printStackTrace();
      }
      return _bulkConnection;
    }



public void doBulkQuery(BulkConnection bulkConnection) {

      try {

        int finalBatchCount=0;

        JobInfo job = new JobInfo();
        job.setObject("tmpTest__c");


        job.setOperation(OperationEnum.query);
        job.setConcurrencyMode(ConcurrencyMode.Parallel);
        job.setContentType(ContentType.CSV);

        job = bulkConnection.createJob(job);
        assert job.getId() != null;
        System.out.println("Job id is " + job.getId());

        job = bulkConnection.getJobStatus(job.getId());

        String query = "SELECT Name, Id FROM tmpTest__c";
        ByteArrayInputStream bout = new ByteArrayInputStream(query.getBytes()); 
        bulkConnection.createBatchFromStream(job, bout);

        BatchInfo[] bListInfo = bulkConnection.getBatchInfoList(job.getId()).getBatchInfo();

        while(bListInfo.length < 2)
            bListInfo = bulkConnection.getBatchInfoList(job.getId()).getBatchInfo();

        System.out.println("Numbe of batches is " + bListInfo.length);
        int numberOfBatchesForQueryExtract = 0;
        int numberOfRecordsExtracted = 0;

        for(int ib=1; ib < bListInfo.length; ib++)
        {   
            finalBatchCount++;
            BatchInfo info = bListInfo[ib];
            numberOfBatchesForQueryExtract++;

            String[] queryResults = null;

            for(int i=0; i<10000; i++) 
            {
                info = bulkConnection.getBatchInfo(job.getId(), info.getId());

                if (info.getState() == BatchStateEnum.Completed) 
                {
                    QueryResultList list = bulkConnection.getQueryResultList(job.getId(),info.getId());             
                    queryResults = list.getResult();
                    break;
                } 
                else if (info.getState() == BatchStateEnum.Failed) 
                {
                    System.out.println("-------------- failed ----------" + info);                  
                    break;
                } 
                else 
                {
                    System.out.println("-------------- waiting ----------"  + info);               
                }
            }

            System.out.println("Batch " + ib + " data is");
            if (queryResults != null) 
            {
                for (String resultId : queryResults) 
                {
                    InputStream resultStream = bulkConnection.getQueryResultStream(job.getId(), info.getId(), resultId);
                    BufferedReader lineReader = new BufferedReader(new InputStreamReader(resultStream,"UTF8"));
                    String lineString = null;
                    while((lineString = lineReader.readLine()) != null) 
                    {
                        System.out.println("lineString : " + lineString);
                        numberOfRecordsExtracted++;
                    }
                }
            }               
        }



        while(finalBatchCount < (bulkConnection.getBatchInfoList(job.getId()).getBatchInfo().length -1))
        {
            bListInfo = bulkConnection.getBatchInfoList(job.getId()).getBatchInfo();

            for(int ib=finalBatchCount; ib < bListInfo.length; ib++)
            {   
                finalBatchCount++;
                BatchInfo info = bListInfo[ib];
                numberOfBatchesForQueryExtract++;

                String[] queryResults = null;

                for(int i=0; i<10000; i++) 
                {
                    info = bulkConnection.getBatchInfo(job.getId(), info.getId());

                    if (info.getState() == BatchStateEnum.Completed) 
                    {
                        QueryResultList list = bulkConnection.getQueryResultList(job.getId(),info.getId());             
                        queryResults = list.getResult();
                        break;
                    } 
                    else if (info.getState() == BatchStateEnum.Failed) 
                    {
                        System.out.println("-------------- failed ----------" + info);                  
                        break;
                    } 
                    else 
                    {
                        System.out.println("-------------- waiting ----------"  + info);               
                    }
                }

                System.out.println("Batch " + ib + " data is");
                if (queryResults != null) 
                {
                    for (String resultId : queryResults) 
                    {
                        InputStream resultStream = bulkConnection.getQueryResultStream(job.getId(), info.getId(), resultId);
                        BufferedReader lineReader = new BufferedReader(new InputStreamReader(resultStream,"UTF8"));
                        String lineString = null;
                        while((lineString = lineReader.readLine()) != null) 
                        {
                            System.out.println("lineString : " + lineString);
                            numberOfRecordsExtracted++;
                        }
                    }
                } 
            }
        }
        bulkConnection.closeJob(job.getId());
        System.out.println("Number of batches created with chunkSize 2000 is " + numberOfBatchesForQueryExtract);
        System.out.println("Number of total records extracted is " + numberOfRecordsExtracted);
        System.out.println("Final Total Batch Count is " + finalBatchCount);
        System.out.println("BatchList length is " + bulkConnection.getBatchInfoList(job.getId()).getBatchInfo().length);
        } catch (AsyncApiException aae) {
            aae.printStackTrace();
        } catch (UnsupportedEncodingException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }         
}

public String convertInputStreamToString(InputStream inputStream)
{
    StringBuilder result = new StringBuilder();
    String line; 
    String newLine = System.getProperty("line.separator");
    boolean flag = false;
    try {
    BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream,"UTF8"));

        while ((line = reader.readLine()) != null) {
            result.append(flag? newLine: "").append(line);
            flag = true;
        }
    } catch (IOException e) {
        e.printStackTrace();
    }
    return result.toString();
}


public static int countLines(String filename)  
{
    InputStream is = null;
    try {
        is = new BufferedInputStream(new FileInputStream(filename));

        byte[] c = new byte[1024];
        int count = 0;
        int readChars = 0;
        boolean empty = true;
        while ((readChars = is.read(c)) != -1) {
            empty = false;
            for (int i = 0; i < readChars; ++i) {
                if (c[i] == '\n') {
                    ++count;
                }
            }
        }
        return (count == 0 && !empty) ? 1 : count;
    } catch (IOException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    } finally {
        try {
            is.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    return 0;
}    
}

Best Answer

Related Solutions

[SalesForce] Cannot close job with Salesforce Bulk API

[SalesForce] Bulk Query with PK Chunking enabled sometimes does not return all the batches

Related Topic