From 7efdbd9320d22faa6ce3905eda2c08fc380972d6 Mon Sep 17 00:00:00 2001
From: David Kyle <david.kyle@elastic.co>
Date: Wed, 14 Dec 2016 13:39:00 +0000
Subject: [PATCH] Fix the reported number of bytes read
 (elastic/elasticsearch#544)

Original commit: elastic/x-pack-elasticsearch@a98902f309ce9ce714cfb1b5f63bd25c13bbdb07
---
 .../job/status/CountingInputStream.java       | 27 +++++--------------
 .../job/status/CountingInputStreamTests.java  | 14 +++-------
 .../rest-api-spec/test/post_data.yaml         |  4 +--
 3 files changed, 11 insertions(+), 34 deletions(-)

diff --git a/elasticsearch/src/main/java/org/elasticsearch/xpack/prelert/job/status/CountingInputStream.java b/elasticsearch/src/main/java/org/elasticsearch/xpack/prelert/job/status/CountingInputStream.java
index 1095aaef4f0..716a29ee96b 100644
--- a/elasticsearch/src/main/java/org/elasticsearch/xpack/prelert/job/status/CountingInputStream.java
+++ b/elasticsearch/src/main/java/org/elasticsearch/xpack/prelert/job/status/CountingInputStream.java
@@ -31,45 +31,30 @@ public class CountingInputStream extends FilterInputStream {
     }
 
     /**
-     * We don't care if the count is one byte out
-     * because we don't check for the case where read
-     * returns -1.
-     * <p>
-     * One of the buffered read(..) methods is more likely to
-     * be called anyway.
+     * Report 1 byte read
      */
     @Override
     public int read() throws IOException {
-        statusReporter.reportBytesRead(1);
+        int read = in.read();
+        statusReporter.reportBytesRead(read < 0 ? 0 : 1);
 
-        return in.read();
+        return read;
     }
 
-    /**
-     * Don't bother checking for the special case where
-     * the stream is closed/finished and read returns -1.
-     * Our count will be 1 byte out.
-     */
     @Override
     public int read(byte[] b) throws IOException {
         int read = in.read(b);
 
-        statusReporter.reportBytesRead(read);
+        statusReporter.reportBytesRead(read < 0 ? 0 : read);
 
         return read;
     }
 
-    /**
-     * Don't bother checking for the special case where
-     * the stream is closed/finished and read returns -1.
-     * Our count will be 1 byte out.
-     */
     @Override
     public int read(byte[] b, int off, int len) throws IOException {
         int read = in.read(b, off, len);
 
-        statusReporter.reportBytesRead(read);
+        statusReporter.reportBytesRead(read < 0 ? 0 : read);
         return read;
     }
-
 }
diff --git a/elasticsearch/src/test/java/org/elasticsearch/xpack/prelert/job/status/CountingInputStreamTests.java b/elasticsearch/src/test/java/org/elasticsearch/xpack/prelert/job/status/CountingInputStreamTests.java
index eb2aa08a42d..156c46fc101 100644
--- a/elasticsearch/src/test/java/org/elasticsearch/xpack/prelert/job/status/CountingInputStreamTests.java
+++ b/elasticsearch/src/test/java/org/elasticsearch/xpack/prelert/job/status/CountingInputStreamTests.java
@@ -29,9 +29,7 @@ public class CountingInputStreamTests extends ESTestCase {
 
         try (CountingInputStream counting = new CountingInputStream(source, statusReporter)) {
             while (counting.read() >= 0) {}
-            // an extra byte is read because we don't check the return
-            // value of the read() method
-            Assert.assertEquals(TEXT.length() + 1, usageReporter.getBytesReadSinceLastReport());
+            Assert.assertEquals(TEXT.length(), usageReporter.getBytesReadSinceLastReport());
 
             Assert.assertEquals(usageReporter.getBytesReadSinceLastReport(), statusReporter.getBytesRead());
         }
@@ -48,10 +46,7 @@ public class CountingInputStreamTests extends ESTestCase {
         try (CountingInputStream counting = new CountingInputStream(source, statusReporter)) {
             byte buf[] = new byte[256];
             while (counting.read(buf) >= 0) {}
-            // one less byte is reported because we don't check
-            // the return value of the read() method
-            Assert.assertEquals(TEXT.length() - 1, usageReporter.getBytesReadSinceLastReport());
-
+            Assert.assertEquals(TEXT.length(), usageReporter.getBytesReadSinceLastReport());
             Assert.assertEquals(usageReporter.getBytesReadSinceLastReport(), statusReporter.getBytesRead());
         }
     }
@@ -67,10 +62,7 @@ public class CountingInputStreamTests extends ESTestCase {
         try (CountingInputStream counting = new CountingInputStream(source, statusReporter)) {
             byte buf[] = new byte[8];
             while (counting.read(buf, 0, 8) >= 0) {}
-            // an extra byte is read because we don't check the return
-            // value of the read() method
-            Assert.assertEquals(TEXT.length() - 1, usageReporter.getBytesReadSinceLastReport());
-
+            Assert.assertEquals(TEXT.length(), usageReporter.getBytesReadSinceLastReport());
             Assert.assertEquals(usageReporter.getBytesReadSinceLastReport(), statusReporter.getBytesRead());
         }
     }
diff --git a/elasticsearch/src/test/resources/rest-api-spec/test/post_data.yaml b/elasticsearch/src/test/resources/rest-api-spec/test/post_data.yaml
index a6483bc67b0..a7388f96b8f 100644
--- a/elasticsearch/src/test/resources/rest-api-spec/test/post_data.yaml
+++ b/elasticsearch/src/test/resources/rest-api-spec/test/post_data.yaml
@@ -31,7 +31,7 @@ setup:
 
   - match: { processed_record_count: 2 }
   - match: { processed_field_count: 4}
-  - match: { input_bytes: 177 }
+  - match: { input_bytes: 178 }
   - match: { input_field_count: 6 }
   - match: { invalid_date_count: 0 }
   - match: { missing_field_count: 0 }
@@ -63,7 +63,7 @@ setup:
 
   - match: { _source.processed_record_count: 2 }
   - match: { _source.processed_field_count: 4}
-  - match: { _source.input_bytes: 177 }
+  - match: { _source.input_bytes: 178 }
   - match: { _source.input_field_count: 6 }
   - match: { _source.invalid_date_count: 0 }
   - match: { _source.missing_field_count: 0 }