public class BulkQueryPKChunking {


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


public BulkConnection login() {

      String userName = "deepnaik@ibm.com";
      String passWord = "DN2006ascPBmF0sqBPkVVCBp1ZmOhxhXQF";
      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 {
        JobInfo job = new JobInfo();
        job.setObject("Account");

        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 Account";
        ByteArrayInputStream bout = new ByteArrayInputStream(query.getBytes());
        bulkConnection.createBatchFromStream(job, bout);

        BatchInfo[] bListInfo = bulkConnection.getBatchInfoList(job.getId()).getBatchInfo();
        System.out.println("Numbe of batches is " + bListInfo.length);
        int numberOfBatchesForQueryExtract = 0;

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

            String[] queryResults = null;

            for(int i=0; i<10000; i++)
            {
                Thread.sleep(30000); //30 sec
                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);
                }
            }

            if (queryResults != null)
            {
                for (String resultId : queryResults)
                {
                    InputStream resultStream = bulkConnection.getQueryResultStream(job.getId(), info.getId(), resultId);
                    String fileNameCreated = createFileName(job.getId(), info.getId(),convertInputStreamToString(resultStream));
                    //System.out.println("Data extracted successfully into CVS file");
                    System.out.println("Filename " + fileNameCreated + ", No of rows in the file is " + countLines(fileNameCreated));
                }
            }
        }
        System.out.println("Number of batches created with chunkSize 2000 is" + numberOfBatchesForQueryExtract);
        } catch (AsyncApiException aae) {
            aae.printStackTrace();
        } catch (InterruptedException ie) {
            ie.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 String createFileName(String jobId, String batchId, String dataAsString)
{
    String namePrefix = "SFBulkExtract";
    String nameSuffix = ".csv";
    String fileSeperator = "/";
    String fFileName = namePrefix + "_" + jobId + "_" + batchId +  nameSuffix;
    createFile(dataAsString,fFileName);
    return fFileName;
}

//Create the CVS file
private void createFile(String dataAsString, String fFileName)
{
    String fFileNameString = fFileName;
    OutputStream outFile = null;
    try {
            File oFile = new File(fFileNameString);
            if (oFile.exists())
            {
                boolean deleteFileSucceeded = oFile.delete();
                if (! deleteFileSucceeded)
                {
                    throw new IOException("Unable to delete file");
                }
            }
            outFile = new BufferedOutputStream(new FileOutputStream(fFileNameString));
            outFile.write(dataAsString.getBytes());
            outFile.flush();
        }catch(FileNotFoundException ex){
            ex.printStackTrace();
        }catch(IOException ex){
            ex.printStackTrace();
        }finally {
            if(outFile != null)
                try {
                    outFile.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
        }
}

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;
}