@@ -0,0 +1,1456 @@
+package com.github.sparkcaller.preprocessing;%0A%0Aimport com.github.sparkcaller.Utils;%0Aimport org.apache.spark.api.java.function.Function;%0Aimport org.broadinstitute.gatk.engine.CommandLineGATK;%0Aimport scala.Tuple2;%0A%0Aimport java.io.File;%0Aimport java.util.ArrayList;%0A%0Apublic class RealignIndels implements Function%3CTuple2%3CFile, File%3E, File%3E %7B%0A    final private String pathToReference;%0A%0A    public RealignIndels(String pathToReference) %7B%0A        this.pathToReference = pathToReference;%0A    %7D%0A    public File call(Tuple2%3CFile, File%3E indelTargetTuple) throws Exception %7B%0A        File inputBam = indelTargetTuple._1;%0A        File intelIntervals = indelTargetTuple._2;%0A%0A        ArrayList%3CString%3E alignerArgs = new ArrayList%3CString%3E();%0A        alignerArgs.add(%22-T%22);%0A        alignerArgs.add(%22IndelRealigner%22);%0A%0A        alignerArgs.add(%22-R%22);%0A        alignerArgs.add(this.pathToReference);%0A%0A        alignerArgs.add(%22-I%22);%0A        alignerArgs.add(inputBam.getPath());%0A%0A        final String ext = %22bam%22;%0A        final String newFileName = Utils.removeExtenstion(inputBam.getPath(), %22bam%22) + %22-realigned.bam%22;%0A        File outputBamFile = new File(newFileName);%0A%0A        alignerArgs.add(%22-targetIntervals%22);%0A        alignerArgs.add(intelIntervals.getPath());%0A%0A        alignerArgs.add(%22-o%22);%0A        alignerArgs.add(outputBamFile.getPath());%0A%0A        CommandLineGATK.start(new CommandLineGATK(), alignerArgs.toArray(new String%5B0%5D));%0A%0A        return outputBamFile;%0A    %7D%0A%7D%0A

Add the class which actually performs the realignment based on the intervals found by the IndelTargetCreator.