diff --git a/CHANGES.txt b/CHANGES.txt
index 2f262b1..ed21f28 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -38,6 +38,8 @@
  
 BUG FIXES
 
+PIG-3655: BinStorage and InterStorage approach to record markers is broken (szita)
+
 PIG-5274: TestEvalPipelineLocal#testSetLocationCalledInFE is failing in spark mode after PIG-5157 (nkollar via szita)
 
 PIG-4767: Partition filter not pushed down when filter clause references variable from another load path (knoguchi)
diff --git a/src/org/apache/pig/PigConfiguration.java b/src/org/apache/pig/PigConfiguration.java
index c9a91b6..4c29da1 100644
--- a/src/org/apache/pig/PigConfiguration.java
+++ b/src/org/apache/pig/PigConfiguration.java
@@ -40,6 +40,24 @@
      */
     public static final String PIG_AUTO_LOCAL_INPUT_MAXBYTES = "pig.auto.local.input.maxbytes";
 
+
+    /**
+     * Sets the length of record markers in binary files produces by Pig between jobs
+     * The longer the byte sequence means less chance of collision with actual data,
+     * shorter sequence means less overhead
+     */
+    public static final String PIG_INTERSTORAGE_SYNCMARKER_SIZE = "pig.interstorage.syncmarker.size";
+    public static final int PIG_INTERSTORAGE_SYNCMARKER_SIZE_MAX = 16;
+    public static final int PIG_INTERSTORAGE_SYNCMARKER_SIZE_DEFAULT = 10;
+    public static final int PIG_INTERSTORAGE_SYNCMARKER_SIZE_MIN = 2;
+
+    /**
+     * Defines the interval (in bytes) when a sync marker should be written into the binary file
+     */
+    public static final String PIG_INTERSTORAGE_SYNCMARKER_INTERVAL = "pig.interstorage.syncmarker.interval";
+    public static final long PIG_INTERSTORAGE_SYNCMARKER_INTERVAL_DEFAULT = 2000;
+
+
     /**
      * Boolean value used to enable or disable fetching without a mapreduce job for DUMP. True by default
      */
diff --git a/src/org/apache/pig/impl/io/InterRecordReader.java b/src/org/apache/pig/impl/io/InterRecordReader.java
index 21c57ad..d2de6ee 100644
--- a/src/org/apache/pig/impl/io/InterRecordReader.java
+++ b/src/org/apache/pig/impl/io/InterRecordReader.java
@@ -42,16 +42,23 @@
 public class InterRecordReader extends RecordReader<Text, Tuple> {
 
   private long start;
-  private long pos;
+  private long lastDataPos;
   private long end;
   private BufferedPositionedInputStream in;
   private Tuple value = null;
-  public static final int RECORD_1 = 0x01;
-  public static final int RECORD_2 = 0x02;
-  public static final int RECORD_3 = 0x03;
   private DataInputStream inData = null;
   private static InterSedes sedes = InterSedesFactory.getInterSedesInstance();
 
+  private byte[] syncMarker;
+  private long lastSyncPos = -1;
+  private long syncMarkerInterval;
+  private long dataBytesSeen = 0;
+
+  public InterRecordReader(int syncMarkerLength, long syncMarkerInterval) {
+      this.syncMarker = new byte[syncMarkerLength];
+      this.syncMarkerInterval = syncMarkerInterval;
+  }
+
   public void initialize(InputSplit genericSplit,
                          TaskAttemptContext context) throws IOException {
     FileSplit split = (FileSplit) genericSplit;
@@ -60,63 +67,131 @@
     end = start + split.getLength();
     final Path file = split.getPath();
 
-    // open the file and seek to the start of the split
+    // open the file
     FileSystem fs = file.getFileSystem(job);
     FSDataInputStream fileIn = fs.open(split.getPath());
-    if (start != 0) {
-        fileIn.seek(start);
+
+    // read the magic byte sequence serving as record marker but only if the file is not empty
+    if (!(start == 0 && end == 0)) {
+        fileIn.readFully(0, syncMarker, 0, syncMarker.length);
     }
+
+    //seek to the start of the split
+    fileIn.seek(start);
+
     in = new BufferedPositionedInputStream(fileIn, start);
     inData = new DataInputStream(in);
   }
-  
-  public boolean nextKeyValue() throws IOException {
+
+
+    /**
+     * Skips to next sync marker
+     * @return true if marker was observed, false if EOF or EndOfSplit was reached
+     * @throws IOException
+     */
+  private boolean skipUntilMarkerOrSplitEndOrEOF() throws IOException {
       int b = 0;
-      //    skip to next record
-      while (true) {
-          if (in == null || in.getPosition() >=end) {
-              return false;
-          }
-          // check if we saw RECORD_1 in our last attempt
-          // this can happen if we have the following 
-          // sequence RECORD_1-RECORD_1-RECORD_2-RECORD_3
-          // After reading the second RECORD_1 in the above
-          // sequence, we should not look for RECORD_1 again
-          if(b != RECORD_1) {
+outer:while (b != -1) {
+          if (b != syncMarker[0]) {
+
+              //There may be a case where we read through a whole split without a marker, then we shouldn't proceed
+              // because the records are from the next split which another reader would pick up too
+              if (in.getPosition() >= end) {
+                  return false;
+              }
               b = in.read();
-              if(b != RECORD_1 && b != -1) {
+              if ((byte) b != syncMarker[0] && b != -1) {
                   continue;
               }
-              if(b == -1) return false;
+              if (b == -1) return false;
           }
+          int i = 1;
+          while (i < syncMarker.length) {
+              b = in.read();
+              if (b == -1) return false;
+              if ((byte) b != syncMarker[i]) {
+                  continue outer;
+              }
+              ++i;
+          }
+          lastSyncPos = in.getPosition();
+          return true;
+      }
+      return false;
+  }
+
+    /**
+     * Reads a sync marker
+     * @return true if sync marker was read, false if EOF reached
+     * @throws IOException thrown if neither EOF nor proper sync was found
+     */
+  private boolean readSyncFullyOrEOF() throws IOException {
+      int b = in.read();
+      if (b == -1) {
+          //EOF reached
+          return false;
+      }
+      if ((byte) b != syncMarker[0]) {
+          throw new IOException("Corrupt data file, expected sync marker at position " + in.getPosition());
+      }
+      int i = 1;
+      while (i < syncMarker.length) {
           b = in.read();
-          if(b != RECORD_2 && b != -1) {
-              continue;
+          if ((byte) b != syncMarker[i]) {
+              throw new IOException("Corrupt data file, expected sync marker at position " + in.getPosition());
           }
-          if(b == -1) return false;
-          b = in.read();
-          if(b != RECORD_3 && b != -1) {
-              continue;
+          ++i;
+      }
+      lastSyncPos = in.getPosition();
+      return true;
+
+  }
+
+  private boolean readDataOrEOF() throws IOException {
+      long preDataPos = in.getPosition();
+      int b = in.read();
+      if(!BinInterSedes.isTupleByte((byte) b) ) {
+          if (b == -1) {
+              //EOF reached
+              return false;
+          } else {
+              throw new IOException("Corrupt data file, expected tuple type byte, but seen " + b);
           }
-          if(b == -1) return false;
-          b = in.read();
-          if(!BinInterSedes.isTupleByte((byte) b) &&
-                  b != -1) {
-              continue;
-          }
-          if(b == -1) return false;
-          break;
       }
       try {
-          // if we got here, we have seen RECORD_1-RECORD_2-RECORD_3-TUPLE_MARKER
-          // sequence - lets now read the contents of the tuple 
           value =  (Tuple)sedes.readDatum(inData, (byte)b);
-          pos=in.getPosition();
+          lastDataPos = in.getPosition();
+          dataBytesSeen += (lastDataPos-preDataPos);
           return true;
       } catch (ExecException ee) {
           throw ee;
       }
+  }
 
+  public boolean nextKeyValue() throws IOException {
+
+      //No marker has been seen, look for next marker
+      if (lastSyncPos == -1) {
+          if (!skipUntilMarkerOrSplitEndOrEOF()) {
+              return false;
+          }
+      }
+
+      //If we've read more or equal amount of data than the sync interval, we expect a sync marker or EOF
+      if (dataBytesSeen >= syncMarkerInterval) {
+          boolean isEOF = !readSyncFullyOrEOF();
+          if (isEOF) {
+              return false;
+          }
+          dataBytesSeen = 0;
+          //If we've just seen a (non-first) sync marker which was completely in the next split then we need to stop
+          if (in.getPosition()-syncMarker.length >= end) {
+              return false;
+          }
+      }
+
+      //Sync marker has been seen, expect data
+      return readDataOrEOF();
   }
 
   @Override
@@ -138,7 +213,7 @@
     if (start == end) {
       return 0.0f;
     } else {
-      return Math.min(1.0f, (pos - start) / (float)(end - start));
+      return Math.min(1.0f, (lastDataPos - start) / (float)(end - start));
     }
   }
   
diff --git a/src/org/apache/pig/impl/io/InterRecordWriter.java b/src/org/apache/pig/impl/io/InterRecordWriter.java
index 5c47e7a..2c15db3 100644
--- a/src/org/apache/pig/impl/io/InterRecordWriter.java
+++ b/src/org/apache/pig/impl/io/InterRecordWriter.java
@@ -17,12 +17,16 @@
  */
 package org.apache.pig.impl.io;
 
-import java.io.DataOutputStream;
 import java.io.IOException;
+import java.rmi.server.UID;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
 
+import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.io.WritableComparable;
 import org.apache.hadoop.mapreduce.RecordWriter;
 import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.hadoop.util.Time;
 import org.apache.pig.data.InterSedes;
 import org.apache.pig.data.InterSedesFactory;
 import org.apache.pig.data.Tuple;
@@ -35,20 +39,34 @@
 public class InterRecordWriter extends
         RecordWriter<org.apache.hadoop.io.WritableComparable, Tuple> {
 
-    public static final int RECORD_1 = 0x01;
-    public static final int RECORD_2 = 0x02;
-    public static final int RECORD_3 = 0x03;
     private static InterSedes sedes = InterSedesFactory.getInterSedesInstance();
+
+    private byte[] syncMarker;
+    private long lastSyncPos = -1;
+    private long syncMarkerInterval;
     /**
      * the outputstream to write out on
      */
-    private DataOutputStream out;
+    private FSDataOutputStream out;
     
     /**
      * 
      */
-    public InterRecordWriter(DataOutputStream out) {
+    public InterRecordWriter(FSDataOutputStream out, int syncMarkerLength, long syncMarkerInterval) {
         this.out = out;
+        this.syncMarkerInterval = syncMarkerInterval;
+        syncMarker = new byte[syncMarkerLength];
+
+        try {
+            MessageDigest digester = MessageDigest.getInstance("MD5");
+            long time = Time.now();
+            digester.update((new UID()+"@"+time).getBytes());
+            byte[] generatedMarker = digester.digest();
+            System.arraycopy(generatedMarker, 0, syncMarker, 0, syncMarkerLength);
+        } catch (NoSuchAlgorithmException e) {
+            throw new RuntimeException(e);
+        }
+
     }
 
     /* (non-Javadoc)
@@ -66,10 +84,11 @@
     @Override
     public void write(WritableComparable wc, Tuple t) throws IOException,
             InterruptedException {
-        // we really only want to write the tuple (value) out here
-        out.write(RECORD_1);
-        out.write(RECORD_2);
-        out.write(RECORD_3);
+        // we really only want to write the tuple (value) out here (and a sync syncMarker before that if necessary)
+        if (lastSyncPos == -1 || out.getPos() >= (lastSyncPos + syncMarkerInterval)) {
+            out.write(syncMarker);
+            lastSyncPos = out.getPos();
+        }
         sedes.writeDatum(out, t);
         
     }
diff --git a/src/org/apache/pig/impl/io/InterStorage.java b/src/org/apache/pig/impl/io/InterStorage.java
index 4a28f95..ac2571f 100644
--- a/src/org/apache/pig/impl/io/InterStorage.java
+++ b/src/org/apache/pig/impl/io/InterStorage.java
@@ -39,6 +39,7 @@
 import org.apache.pig.FileInputLoadFunc;
 import org.apache.pig.LoadFunc;
 import org.apache.pig.LoadMetadata;
+import org.apache.pig.PigConfiguration;
 import org.apache.pig.ResourceSchema;
 import org.apache.pig.ResourceStatistics;
 import org.apache.pig.StoreFunc;
@@ -62,10 +63,10 @@
 
     private static final Log mLog = LogFactory.getLog(InterStorage.class);
     public static final String useLog = "Pig Internal storage in use";
-    
+
     private InterRecordReader recReader = null;
     private InterRecordWriter recWriter = null;
-    
+
     /**
      * Simple binary nested reader format
      */
@@ -102,7 +103,9 @@
         public RecordReader<Text, Tuple> createRecordReader(InputSplit split,
                 TaskAttemptContext context) throws IOException,
                 InterruptedException {
-            return new InterRecordReader();
+            return new InterRecordReader(retrieveMarkerLengthFromConf(context.getConfiguration()),
+                                         retrieveMarkerIntervalFromConf(context.getConfiguration())
+            );
         }
 
     }
@@ -141,7 +144,10 @@
             Path file = getDefaultWorkFile(job, "");
             FileSystem fs = file.getFileSystem(conf);
             FSDataOutputStream fileOut = fs.create(file, false);
-            return new InterRecordWriter(fileOut);
+            return new InterRecordWriter(fileOut,
+                    retrieveMarkerLengthFromConf(job.getConfiguration()),
+                    retrieveMarkerIntervalFromConf(job.getConfiguration())
+            );
         }
     }
 
@@ -208,4 +214,22 @@
     public void cleanupOnSuccess(String location, Job job) throws IOException {
         // DEFAULT: do nothing
     }
+
+    private static int retrieveMarkerLengthFromConf(Configuration conf) {
+        int requestedLength = conf.getInt(PigConfiguration.PIG_INTERSTORAGE_SYNCMARKER_SIZE,
+                PigConfiguration.PIG_INTERSTORAGE_SYNCMARKER_SIZE_DEFAULT);
+
+        if (requestedLength > PigConfiguration.PIG_INTERSTORAGE_SYNCMARKER_SIZE_MAX) {
+            requestedLength = PigConfiguration.PIG_INTERSTORAGE_SYNCMARKER_SIZE_MAX;
+        } else if (requestedLength < PigConfiguration.PIG_INTERSTORAGE_SYNCMARKER_SIZE_MIN) {
+            requestedLength = PigConfiguration.PIG_INTERSTORAGE_SYNCMARKER_SIZE_MIN;
+        }
+
+        return requestedLength;
+    }
+
+    private static long retrieveMarkerIntervalFromConf(Configuration conf) {
+        return conf.getLong(PigConfiguration.PIG_INTERSTORAGE_SYNCMARKER_INTERVAL,
+                PigConfiguration.PIG_INTERSTORAGE_SYNCMARKER_INTERVAL_DEFAULT);
+    }
 }
diff --git a/test/org/apache/pig/data/TestSchemaTuple.java b/test/org/apache/pig/data/TestSchemaTuple.java
index dac0f08..544d462 100644
--- a/test/org/apache/pig/data/TestSchemaTuple.java
+++ b/test/org/apache/pig/data/TestSchemaTuple.java
@@ -44,6 +44,7 @@
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.mapreduce.InputSplit;
 import org.apache.hadoop.mapreduce.TaskAttemptID;
 import org.apache.hadoop.mapreduce.lib.input.FileSplit;
@@ -506,30 +507,20 @@
         File temp = File.createTempFile("tmp", "tmp");
         temp.deleteOnExit();
         FileOutputStream fos = new FileOutputStream(temp);
-        DataOutputStream dos = new DataOutputStream(fos);
+        FSDataOutputStream dos = new FSDataOutputStream(fos, null);
 
-        InterRecordWriter writer = new InterRecordWriter(dos);
+        InterRecordWriter writer = new InterRecordWriter(dos,
+                PigConfiguration.PIG_INTERSTORAGE_SYNCMARKER_SIZE_DEFAULT,
+                PigConfiguration.PIG_INTERSTORAGE_SYNCMARKER_INTERVAL_DEFAULT);
 
-        // We add these lines because a part of the InterStorage logic
-        // is the ability to seek to the next Tuple based on a magic set
-        // of bytes. This emulates the random byes that will be present
-        // at the beginning of a split.
-        dos.writeByte(r.nextInt());
-        dos.writeByte(r.nextInt());
-        dos.writeByte(r.nextInt());
-        dos.writeByte(r.nextInt());
-        dos.writeByte(r.nextInt());
-        dos.writeByte(r.nextInt());
+        // This test does not cover the case of overlapping record bytes that may be present at the
+        // beginning of a split, for that see org.apache.pig.test.TestBinInterSedes#testInterStorageSyncMarker()
 
         for (int i = 0; i < sz; i++) {
             SchemaTuple<?> st = (SchemaTuple<?>)tf.newTuple();
             fillWithData(st);
             writer.write(null, st);
             written.add(st);
-
-            dos.writeByte(r.nextInt());
-            dos.writeByte(r.nextInt());
-            dos.writeByte(r.nextInt());
         }
         writer.close(null);
 
@@ -541,7 +532,8 @@
 
         InputSplit is = new FileSplit(new Path(temp.getAbsolutePath()), 0, temp.length(), null);
 
-        InterRecordReader reader = new InterRecordReader();
+        InterRecordReader reader = new InterRecordReader(PigConfiguration.PIG_INTERSTORAGE_SYNCMARKER_SIZE_DEFAULT,
+                                                         PigConfiguration.PIG_INTERSTORAGE_SYNCMARKER_INTERVAL_DEFAULT);
         reader.initialize(is, HadoopShims.createTaskAttemptContext(conf, taskId));
 
         for (int i = 0; i < sz; i++) {
diff --git a/test/org/apache/pig/test/TestBinInterSedes.java b/test/org/apache/pig/test/TestBinInterSedes.java
index b8a8ea5..65eb15e 100644
--- a/test/org/apache/pig/test/TestBinInterSedes.java
+++ b/test/org/apache/pig/test/TestBinInterSedes.java
@@ -24,11 +24,16 @@
 import java.io.DataInputStream;
 import java.io.DataOutput;
 import java.io.DataOutputStream;
+import java.io.File;
 import java.io.IOException;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.Map;
+import java.util.Properties;
 import java.util.Random;
 
+import org.apache.pig.PigConfiguration;
+import org.apache.pig.PigServer;
 import org.apache.pig.data.BagFactory;
 import org.apache.pig.data.BinInterSedes;
 import org.apache.pig.data.DataBag;
@@ -296,6 +301,142 @@
         }
     }
 
+
+    /*
+      The following tests are intended to verify the reading and writing of intermediate files of Pig (of InterStorage)
+      The test records are 11,14,22,14 bytes long.
+      Below I illustrate the splits in rows, records as [] with size and sync markers with [M]
+     */
+
+    /**
+     * One sync marker only and three splits where the records overlap the splitends.
+     * (Reader of 1st split should read every record, readers of 2nd and 3rd splits should read no records.)
+     * [M(10)] [11] [11-
+     *  -3] [ 22 ] [7-
+     *  -7]
+     *
+     * @throws Exception
+     */
+    @Test
+    public void testSyncMarkerOneMarkerAtBeginningOnly() throws Exception {
+        testInterStorageSyncMarker(32, 10, 2000L);
+    }
+
+    /**
+     * Some sync markers are positioned so that they begin at a split's end and they end in the next split's beginning.
+     * (Reader of a split has to read until the next sync marker that has all its bytes in a following split.)
+     * @throws Exception
+     */
+    @Test
+    public void testSyncMarkerOverlappingMarker() throws Exception {
+        /*
+         * [M(16)] [11] [M(16)] [5-
+         * -9] [M(16)] [ 22 ] [M(1-
+         * -15)] [14]
+         */
+        testInterStorageSyncMarker(48, 16, 10L);
+        /*
+         * [M(4)] [ 4-
+         *     -7] [1-
+         *    - 8 -
+         *   -5] [M(3-
+         * -1)] [ 7-
+         *    - 8 -
+         *   -7] [M(1-
+         * -3)] [  5-
+         *    - 8 -
+         *  -1]
+         */
+        testInterStorageSyncMarker(8, 4, 20L);
+    }
+
+    /**
+     * No illustration for this one to save characters .. Sync size is over 3 times the size of split size, this is an
+     * extremely unlikely scenario. Markers here span over 4 splits.
+     * @throws Exception
+     */
+    @Test
+    public void testSyncMarkerLongerMarkerThanSplit() throws Exception {
+        testInterStorageSyncMarker(5, 16, 20L);
+    }
+
+    /**
+     * A sync marker is positioned at exactly the end of the first split without overlapping into the next one.
+     * (Reader of the 1st split should read past it and into the 2nd split until next marker.)
+     *
+     * [M(2)] [11] [14] [M(2)]
+     * [  22  ] [M(2)] [ 5-
+     * -9]
+     * @throws Exception
+     */
+    @Test
+    public void testSyncMarkerMarkerOnSplitEnd() throws Exception {
+        testInterStorageSyncMarker(29, 2, 20L);
+    }
+
+    /**
+     * A sync marker is positioned at exactly the beginning of the 3rd split.
+     * (Reader of the 1st split should read 1st and 2nd splits fully, reader of 2nd split should read no records.)
+     *
+     * [M(3)] [11]
+     * [    14   ]
+     * [M(3) [11-
+     *   -11 ] [3-
+     *      -11 ]
+     * @throws Exception
+     */
+    @Test
+    public void testSyncMarkerMarkerOnSplitBeginning() throws Exception {
+        testInterStorageSyncMarker(14, 3, 25L);
+    }
+
+    private void testInterStorageSyncMarker(int maxSplitSize, int syncSize, long syncInterval) throws Exception {
+        PigServer pigServer = new PigServer(Util.getLocalTestMode(), new Properties());
+
+        Properties pigProperties = pigServer.getPigContext().getProperties();
+        pigProperties.setProperty("mapreduce.input.fileinputformat.split.maxsize", String.valueOf(maxSplitSize));
+        pigProperties.setProperty(PigConfiguration.PIG_INTERSTORAGE_SYNCMARKER_SIZE, String.valueOf(syncSize));
+        pigProperties.setProperty(PigConfiguration.PIG_INTERSTORAGE_SYNCMARKER_INTERVAL, String.valueOf(syncInterval));
+
+        //Without proper random record markers 0x01020327 would be identified as a marker and 0x50 as an unknown datatype
+        //ByteBuffer.wrap(new byte[]{0x01, 0x02, 0x03, 0x27, 0x50, 0x0, 0x0, 0x0}).getLong() => 72624011372134400
+
+        String[] inputData = new String[]{"apple\t1\t1","orange\t2\t2","kiwi\t16909095\t72624011372134400","orange\t4\t4"};
+        String[] expected = new String[] {"(apple,1,1)","(orange,2,2)","(kiwi,16909095,72624011372134400)","(orange,4,4)"};
+        File inputFile = Util.createInputFile("interStorageInput", "", inputData);
+        inputFile.deleteOnExit();
+
+        //Without proper random record markers 0x01020327 would be identified as a marker and although no errors are
+        // thrown the result will contain incorrect schema and values past this number
+        //ByteBuffer.wrap(new byte[]{0x01, 0x02, 0x03, 0x27, 0x01, 0x0, 0x0, 0x0}).getLong() => 72624010046734336
+
+        String[] inputData2 = new String[]{"apple\t1\t1","orange\t2\t2","kiwi\t16909095\t72624010046734336","orange\t4\t4"};
+        String[] expected2 = new String[] {"(apple,1,1)","(orange,2,2)","(kiwi,16909095,72624010046734336)","(orange,4,4)"};
+        File inputFile2 = Util.createInputFile("interStorageInput2", "", inputData2);
+        inputFile2.deleteOnExit();
+
+        File binOutputdir = new File("build/test/interStorageTest");
+        Util.deleteDirectory(binOutputdir);
+
+        String script = "A = LOAD '"+inputFile.getAbsolutePath()+"' AS (name:chararray, cnt:int, cnt2:long);\n" +
+                "STORE A INTO '"+binOutputdir.getAbsolutePath()+"' USING org.apache.pig.impl.io.InterStorage();\n" +
+                "\n" +
+                "B = LOAD '"+binOutputdir.getAbsolutePath()+"' USING org.apache.pig.impl.io.InterStorage();\n";
+
+        pigServer.registerQuery(script);
+        Iterator<Tuple> it = pigServer.openIterator("B");
+        Util.checkQueryOutputsAfterSortRecursive(it, expected,
+                org.apache.pig.newplan.logical.Util.translateSchema(pigServer.dumpSchema("B")));
+
+        Util.deleteDirectory(binOutputdir);
+
+        pigServer.registerQuery(script.replaceAll(inputFile.getAbsolutePath(), inputFile2.getAbsolutePath()));
+        it = pigServer.openIterator("B");
+        Util.checkQueryOutputsAfterSortRecursive(it, expected2,
+                org.apache.pig.newplan.logical.Util.translateSchema(pigServer.dumpSchema("B")));
+
+    }
+
     private void testSerTuple(Tuple t, byte[] expected) throws Exception {
         ByteArrayOutputStream baos = new ByteArrayOutputStream();
         DataOutput out = new DataOutputStream(baos);
diff --git a/test/org/apache/pig/test/TestFRJoin2.java b/test/org/apache/pig/test/TestFRJoin2.java
index 274bf3f..1b205f4 100644
--- a/test/org/apache/pig/test/TestFRJoin2.java
+++ b/test/org/apache/pig/test/TestFRJoin2.java
@@ -414,12 +414,18 @@
         pigServer.registerQuery("C = foreach C generate MAX(B.x) as x;");
         pigServer.registerQuery("D = join A by x, B by x, C by x using 'repl';");
         {
-            // When the replicated input sizes=(12 + 5) is bigger than
-            // pig.join.replicated.max.bytes=16, we throw exception
+            // When the replicated input size is bigger than
+            // pig.join.replicated.max.bytes, we throw exception
+            // Expected replicated size below:
+            //  Alias B: sync marker + 2 records (1 tuple type byte + 2 integers (0 or 1))
+            //  Alias C: sync marker + 1 record (1 tuple type byte + 1 integer (1))
+            long expectedReplicateSize = PigConfiguration.PIG_INTERSTORAGE_SYNCMARKER_SIZE_DEFAULT + 2*(1 +1+1)
+                                       + PigConfiguration.PIG_INTERSTORAGE_SYNCMARKER_SIZE_DEFAULT + 1*(1 +1);
+
             try {
                 pigServer.getPigContext().getProperties().setProperty(
                         PigConfiguration.PIG_JOIN_REPLICATED_MAX_BYTES,
-                        String.valueOf(16));
+                        String.valueOf(expectedReplicateSize-1));
                 pigServer.openIterator("D");
                 Assert.fail();
             } catch (FrontendException e) {
@@ -428,10 +434,10 @@
                         e.getCause().getCause().getCause().getMessage());
             }
 
-            // If we increase the size to 17, it should work
+            // If we increase the max size setting to the expected amount it works
             pigServer.getPigContext().getProperties().setProperty(
                         PigConfiguration.PIG_JOIN_REPLICATED_MAX_BYTES,
-                        String.valueOf(17));
+                        String.valueOf(expectedReplicateSize));
             pigServer.openIterator("D");
         }
     }