Fix when we record to only capture actual writes.

Author: henningandersen

x-pack/plugin/blob-cache/src/main/java/org/elasticsearch/blobcache/shared/SharedBlobCacheService.java

@@ -1312,13 +1312,11 @@ private int readSingleRegion(
                 mapSubRangeToRegion(rangeToWrite, region),
                 mapSubRangeToRegion(rangeToRead, region),
                 readerWithOffset(reader, fileRegion, Math.toIntExact(rangeToRead.start() - regionStart)),
-                writerWithOffset(writer, fileRegion, Math.toIntExact(rangeToWrite.start() - regionStart)),
+                metricRecordingWriter(writerWithOffset(writer, fileRegion, Math.toIntExact(rangeToWrite.start() - regionStart))),
                 ioExecutor,
                 readFuture
             );
-            int result = readFuture.get();
-            blobCacheMetrics.recordWrite();
-            return result;
+            return readFuture.get();
         }
 
         private int readMultiRegions(
@@ -1346,7 +1344,9 @@ private int readMultiRegions(
                             mapSubRangeToRegion(rangeToWrite, region),
                             subRangeToRead,
                             readerWithOffset(reader, fileRegion, Math.toIntExact(rangeToRead.start() - regionStart)),
-                            writerWithOffset(writer, fileRegion, Math.toIntExact(rangeToWrite.start() - regionStart)),
+                            metricRecordingWriter(
+                                writerWithOffset(writer, fileRegion, Math.toIntExact(rangeToWrite.start() - regionStart))
+                            ),
                             ioExecutor,
                             listener
                         );
@@ -1357,7 +1357,6 @@ private int readMultiRegions(
                 }
             }
             readsComplete.get();
-            blobCacheMetrics.recordWrites(endRegion - startRegion + 1);
             return bytesRead.get();
         }
 
@@ -1422,6 +1421,16 @@ public void fillCacheRange(
             return adjustedWriter;
         }
 
+        private RangeMissingHandler metricRecordingWriter(RangeMissingHandler writer) {
+            return new DelegatingRangeMissingHandler(writer) {
+                @Override
+                public SourceInputStreamFactory sharedInputStreamFactory(List<SparseFileTracker.Gap> gaps) {
+                    blobCacheMetrics.recordWrite();
+                    return super.sharedInputStreamFactory(gaps);
+                }
+            };
+        }
+
         private RangeAvailableHandler readerWithOffset(RangeAvailableHandler reader, CacheFileRegion<KeyType> fileRegion, int readOffset) {
             final RangeAvailableHandler adjustedReader = (channel, channelPos, relativePos, len) -> reader.onRangeAvailable(
                 channel,