Decrypt File onto Stream , Transform and Encrypt as a file using Spring Batch

Question

Aim - Decrypt a .pgp encrypted file , read the data as a stream , perform Transformation to vendor requirement, encrypt as stream and write to a File.

Logic - Custom Reader, Writer and Tasklet , Stored the decrypted/encrypted data onto ExecutionContext and pass to different steps.

Works for - Small file (~1MB)

Issue faced - Tried with a (~10MB - 10K records) - Reading step was successful, but when begin writing data as an Encrypted File - Memory issue - java.lang.OutOfMemoryError: Java heap space

Code snippet -

<job id="testJob" xmlns="http://www.springframework.org/schema/batch">

    <!-- Read Encrypted file and Decrypt -->
    <batch:step id="decryptFile" next="readAndWriteData">
        <batch:tasklet ref="fileDecryptionTasklet">
            <batch:listeners>
                <batch:listener ref="decryptFileListener" />
            </batch:listeners>
        </batch:tasklet>
    </batch:step>

    <!-- Read data from decryption step and write to Stream -->
    <batch:step id="readAndWriteData" next="encryptFile">
        <batch:tasklet>
            <batch:chunk reader="hrdsCustomReader" processor="Processor"
                writer="CustomWriter" commit-interval="${.ftp.comit.interval}" />
            <batch:listeners>
                <batch:listener ref="encryptFileListener" />
            </batch:listeners>
        </batch:tasklet>
    </batch:step>

    <!-- Write to vendor specific file -->
    <batch:step id="encryptFile">
        <batch:tasklet ref="fileEncryptionTasklet" />
    </batch:step>

</job>

Tasklet and custom writers code snippets -

@Override
public String read() throws Exception, UnexpectedInputException,
        ParseException {

    decryptedData = (String) stepExecution.getJobExecution()
            .getExecutionContext().get("DecryptedData");
    if (decryptedData != null)
        //logger.info("decryptedData in Custom Reader - \n" + decryptedData);

    stepExecution.getJobExecution().getExecutionContext()
            .put("DecryptedData", null);
    return decryptedData;
}

 
public void write(List items) throws Exception {
    logger.info("Begin writing data as an Encrypted File");

      Iterator itr = items.iterator();
      while(itr.hasNext()) {
             String element =  itr.next();
             lineBuffer.append(element+LINE_SEPARATOR);
          }
      ExecutionContext stepContext = this.stepExecution.getExecutionContext();
        stepContext.put("EncryptedData", lineBuffer);
}

public RepeatStatus execute(StepContribution step, ChunkContext chunk)
        throws Exception {

    InputStream inputstream = new FileInputStream(inputdirectory);

    Message encryptMessage = MessageBuilder
            .withPayload(inputstream)
            .setHeader(
                    FileEncryptionTransformer.ENCRYPTION_OPERATION_HEADER,
                    "decryptAndVerify")
            .setHeader(
                    FileEncryptionTransformer.ENCRYPTION_OPERATION_HEADER,
                    EncryptionUtil.DECRYPT_STREAM_OPERATION)
            .setHeader(FileEncryptionTransformer.SOURCE_FILE_NAME_HEADER,
                    filename).build();

    InputStream inputStream = pgptransformer
            .doTransformStream(encryptMessage);
    String strData = IOUtils.toString(inputStream, "UTF-8");
    inputstream.close();


    chunk.getStepContext().getStepExecution().getExecutionContext().put("DecryptedData", strData);

    return null;
}

public RepeatStatus execute(StepContribution step, ChunkContext chunk)
        throws Exception {


        lineBuffer = (StringBuffer) chunk.getStepContext()
                .getJobExecutionContext().get("EncryptedData");
        byte[] bytes = lineBuffer.toString().getBytes();
        InputStream inputStream = new ByteArrayInputStream(bytes);

        Message encryptMessage = MessageBuilder
                .withPayload(inputStream)
                .setHeader(PGPFileTransformer.OUTPUT_FILE_FOLDER,
                        outputdirectory)
                .setHeader(
                        FileEncryptionTransformer.ENCRYPTION_OPERATION_HEADER,
                        "signAndEncrypt")
                .setHeader(
                        FileEncryptionTransformer.ENCRYPTION_OPERATION_HEADER,
                        EncryptionUtil.ENCRYPT_STREAM_OPERATION)
                .setHeader(FileEncryptionTransformer.SOURCE_FILE_NAME_HEADER,
                        filename).build();

        pgptransformer.doTransform(encryptMessage);
        inputStream.close();


    chunk.getStepContext().getStepExecution().getExecutionContext().put("EncryptedData", null);

    return null;
}

Appreciate if somebody can help resolve the issue.

Answer 1

was able to decrypt and process 100K records in <2mins.

Logic - Process in chunks - 200 records into 1 line.

Posting the code below -

Batch Config-

<job id="aic-batch-xxx-ftp" xmlns="http://www.springframework.org/schema/batch">


    <!-- Read data , decrypt , process and write to encrypted file -->
    <batch:step id="readAndWriteData">
        <batch:tasklet>
            <batch:chunk reader="xxxCustomReader" processor="xxxFileProccessor"
                writer="xxxCustomWriter" commit-interval="${aic.batch.xxx.ftp.comit.interval}" />
        </batch:tasklet>
    </batch:step>


</job>

Reader Logic -

StringBuffer decryptedData = new StringBuffer();
    String strLine = "";

    PGPLib pgp = new PGPLib();
    KeyStore keyStore = new KeyStore("xxx.keystore", "xxx");

    long startTime = System.currentTimeMillis();
    // Read & decrypt File Line By Line
    if ((strLine = bufferedReader.readLine()) != null) {
        strLine = strLine.replace("NEW_LINE", "\r\n");
        decryptedData.append((pgp.decryptString(strLine, keyStore,
                "xxx")));
        long endTime = System.currentTimeMillis();
        logger.debug("Total time taken = " + (endTime - startTime) + " msec");
        return decryptedData;
    }
    else
        return null;

Writer Logic -

public void write(List<? extends StringBuffer> items) throws Exception {
    logger.debug("Begin writing data as an Encrypted File");

    @SuppressWarnings("unchecked")
    Iterator<StringBuffer> itr = (Iterator<StringBuffer>) items.iterator();
    while (itr.hasNext()) {
        StringBuffer element = itr.next();
        encrypt(element);
        count++;
    }
}

public void encrypt(StringBuffer element) throws PGPException, IOException {

    PGPLib pgp = new PGPLib();
    KeyStore keyStore = new KeyStore("xxx.keystore", "xxx");

    String strLine = element.toString();
    StringBuffer buffer = new StringBuffer("");
    int i = 0;
    long startTime = System.currentTimeMillis();

    if (i % 200 == 0) {
        if (i != 0) {
            String encryptString = pgp.encryptString(buffer.toString(),
                    keyStore,
                    "xxx");
            encryptString = encryptString.replace("\r\n", "NEW_LINE");
            bufferedWriter.write(encryptString);
            bufferedWriter.newLine();
        }
        buffer = new StringBuffer(strLine);
    } else {
        buffer.append("\r\n").append(strLine);
    }
    i++;
    if (buffer != null && buffer.length() > 0) {
        String encryptString = pgp.encryptString(buffer.toString(),
                keyStore, "xxx");
        encryptString = encryptString.replace("\r\n", "NEW_LINE");
        bufferedWriter.write(encryptString);
        bufferedWriter.newLine();
    }

    long endTime = System.currentTimeMillis();
    logger.debug("Total time taken = " + (endTime - startTime) + " msec");

}