fixed a bug in fs translog where it wasn't seeking correctly, and not counting the number of operations

modules/elasticsearch/src/main/java/org/elasticsearch/index/translog/fs/FsSnapshot.java

@@ -78,6 +78,8 @@ public class FsSnapshot implements Translog.Snapshot {
             int opSize = dis.readInt();
             position += 4;
             if ((position + opSize) > length) {
+                // restore the position to before we read the opSize
+                position -= 4;
                 return false;
             }
             position += opSize;

modules/elasticsearch/src/main/java/org/elasticsearch/index/translog/fs/FsTranslog.java

@@ -108,6 +108,7 @@ public class FsTranslog extends AbstractIndexShardComponent implements Translog
                 bosOs.flush();
                 raf.raf().writeInt(bos.size());
                 raf.raf().write(bos.unsafeByteArray(), 0, bos.size());
+                operationCounter.incrementAndGet();
             } catch (Exception e) {
                 throw new TranslogException(shardId, "Failed to write operation [" + operation + "]", e);
             }
@@ -134,7 +135,7 @@ public class FsTranslog extends AbstractIndexShardComponent implements Translog
                 FsSnapshot fsSnapshot = (FsSnapshot) snapshot;
                 raf.increaseRefCount();
                 FsSnapshot newSnapshot = new FsSnapshot(shardId, id, raf, raf.raf().getFilePointer());
-                newSnapshot.seekForward(fsSnapshot.length());
+                newSnapshot.seekForward(fsSnapshot.position());
                 return newSnapshot;
             } catch (IOException e) {
                 throw new TranslogException(shardId, "Failed to snapshot", e);

modules/elasticsearch/src/main/java/org/elasticsearch/index/translog/memory/MemoryTranslog.java

@@ -95,7 +95,7 @@ public class MemoryTranslog extends AbstractIndexShardComponent implements Trans
                 return snapshot();
             }
             MemorySnapshot newSnapshot = new MemorySnapshot(currentId(), operations, operationCounter.get());
-            newSnapshot.seekForward(memorySnapshot.length());
+            newSnapshot.seekForward(memorySnapshot.position());
             return newSnapshot;
         }
     }

modules/elasticsearch/src/test/java/org/elasticsearch/index/translog/AbstractSimpleTranslogTests.java

@@ -124,6 +124,9 @@ public abstract class AbstractSimpleTranslogTests {
         snapshot.release();
 
         Translog.Snapshot snapshot1 = translog.snapshot();
+        // we use the translogSize to also navigate to the last position on this snapshot
+        // so snapshot(Snapshot) will work properly
+        assertThat(snapshot1, translogSize(1));
 
         translog.add(new Translog.Index("test", "2", new byte[]{2}));
         snapshot = translog.snapshot(snapshot1);