diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/NoFetchingInput.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/NoFetchingInput.java
index 252d8ae001851..47962aa60064e 100644
--- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/NoFetchingInput.java
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/NoFetchingInput.java
@@ -73,7 +73,7 @@ protected int require(int required) throws KryoException {
         position = 0;
         int bytesRead = 0;
         int count;
-        while (true) {
+        while (bytesRead < required) {
             count = fill(buffer, bytesRead, required - bytesRead);
 
             if (count == -1) {
@@ -81,9 +81,6 @@ protected int require(int required) throws KryoException {
             }
 
             bytesRead += count;
-            if (bytesRead == required) {
-                break;
-            }
         }
         limit = required;
         return required;
@@ -121,7 +118,7 @@ public void readBytes(byte[] bytes, int offset, int count) throws KryoException
             int bytesRead = 0;
             int c;
 
-            while (true) {
+            while (bytesRead < count) {
                 c = inputStream.read(bytes, offset + bytesRead, count - bytesRead);
 
                 if (c == -1) {
@@ -129,10 +126,6 @@ public void readBytes(byte[] bytes, int offset, int count) throws KryoException
                 }
 
                 bytesRead += c;
-
-                if (bytesRead == count) {
-                    break;
-                }
             }
         } catch (IOException ex) {
             throw new KryoException(ex);