diff --git a/parquet-avro/src/test/java/parquet/avro/TestSpecificReadWrite.java b/parquet-avro/src/test/java/parquet/avro/TestSpecificReadWrite.java
index 48855a0323..03224c2c14 100644
--- a/parquet-avro/src/test/java/parquet/avro/TestSpecificReadWrite.java
+++ b/parquet-avro/src/test/java/parquet/avro/TestSpecificReadWrite.java
@@ -76,6 +76,52 @@ public void testFilterMatchesMultiple() throws IOException {
     assertNull(reader.read());
   }
 
+  @Test
+  public void testFilterMatchesMultipleBlocks() throws IOException {
+    Path path = writeCarsToParquetFile(10000, CompressionCodecName.UNCOMPRESSED, false, DEFAULT_BLOCK_SIZE/64, DEFAULT_PAGE_SIZE/64);
+    ParquetReader<Car> reader = new AvroParquetReader<Car>(path, column("make", equalTo("Volkswagen")));
+    for (int i = 0; i < 10000; i++) {
+      assertEquals(getVwPolo().toString(), reader.read().toString());
+      assertEquals(getVwPassat().toString(), reader.read().toString());
+    }
+    assertNull(reader.read());
+  }
+
+  @Test
+  public void testFilterMatchesNoBlocks() throws IOException {
+    Path path = writeCarsToParquetFile(10000, CompressionCodecName.UNCOMPRESSED, false, DEFAULT_BLOCK_SIZE/64, DEFAULT_PAGE_SIZE/64);
+    ParquetReader<Car> reader = new AvroParquetReader<Car>(path, column("make", equalTo("Bogus")));
+    assertNull(reader.read());
+  }
+
+  @Test
+  public void testFilterMatchesFinalBlockOnly() throws IOException {
+    File tmp = File.createTempFile(getClass().getSimpleName(), ".tmp");
+    tmp.deleteOnExit();
+    tmp.delete();
+    Path path = new Path(tmp.getPath());
+
+    Car vwPolo   = getVwPolo();
+    Car vwPassat = getVwPassat();
+    Car bmwMini  = getBmwMini();
+
+    ParquetWriter<Car> writer = new AvroParquetWriter<Car>(path, Car.SCHEMA$,
+        CompressionCodecName.UNCOMPRESSED, DEFAULT_BLOCK_SIZE/128, DEFAULT_PAGE_SIZE/128,
+        false);
+    for (int i = 0; i < 10000; i++) {
+      writer.write(vwPolo);
+      writer.write(vwPassat);
+      writer.write(vwPolo);
+    }
+    writer.write(bmwMini); // only write BMW in last block
+    writer.close();
+
+    ParquetReader<Car> reader = new AvroParquetReader<Car>(path, column("make",
+        equalTo("BMW")));
+    assertEquals(getBmwMini().toString(), reader.read().toString());
+    assertNull(reader.read());
+  }
+
   @Test
   public void testFilterWithDictionary() throws IOException {
     Path path = writeCarsToParquetFile(1,CompressionCodecName.UNCOMPRESSED,true);
@@ -159,6 +205,10 @@ public void testAvroReadSchema() throws IOException {
   }
 
   private Path writeCarsToParquetFile( int num, CompressionCodecName compression, boolean enableDictionary) throws IOException {
+    return writeCarsToParquetFile(num, compression, enableDictionary, DEFAULT_BLOCK_SIZE, DEFAULT_PAGE_SIZE);
+  }
+
+  private Path writeCarsToParquetFile( int num, CompressionCodecName compression, boolean enableDictionary, int blockSize, int pageSize) throws IOException {
     File tmp = File.createTempFile(getClass().getSimpleName(), ".tmp");
     tmp.deleteOnExit();
     tmp.delete();
@@ -169,7 +219,7 @@ private Path writeCarsToParquetFile( int num, CompressionCodecName compression,
     Car bmwMini  = getBmwMini();
 
     ParquetWriter<Car> writer = new AvroParquetWriter<Car>(path,Car.SCHEMA$, compression,
-        DEFAULT_BLOCK_SIZE, DEFAULT_PAGE_SIZE, enableDictionary);
+        blockSize, pageSize, enableDictionary);
     for (int i = 0; i < num; i++) {
       writer.write(vwPolo);
       writer.write(vwPassat);
diff --git a/parquet-hadoop/src/main/java/parquet/hadoop/InternalParquetRecordReader.java b/parquet-hadoop/src/main/java/parquet/hadoop/InternalParquetRecordReader.java
index 8d99a290f8..f3aa81f530 100644
--- a/parquet-hadoop/src/main/java/parquet/hadoop/InternalParquetRecordReader.java
+++ b/parquet-hadoop/src/main/java/parquet/hadoop/InternalParquetRecordReader.java
@@ -53,7 +53,7 @@ class InternalParquetRecordReader<T> {
 
   private T currentValue;
   private long total;
-  private int current = 0;
+  private long current = 0;
   private int currentBlock = -1;
   private ParquetFileReader reader;
   private parquet.io.RecordReader<T> recordReader;
@@ -173,8 +173,18 @@ public boolean nextKeyValue() throws IOException, InterruptedException {
       try {
         checkRead();
         currentValue = recordReader.read();
-        if (DEBUG) LOG.debug("read value: " + currentValue);
         current ++;
+        while (currentValue == null) { // only happens with FilteredRecordReader at end of block
+          current = totalCountLoadedSoFar;
+          if (current < total) {
+            checkRead();
+            currentValue = recordReader.read();
+            current ++;
+            continue;
+          }
+          return false;
+        }
+        if (DEBUG) LOG.debug("read value: " + currentValue);
       } catch (RuntimeException e) {
         throw new ParquetDecodingException(format("Can not read value at %d in block %d in file %s", current, currentBlock, file), e);
       }