move prefetching stat names to StreamStatistics

Author: ahmarsuhail

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/statistics/StreamStatisticNames.java

@@ -387,6 +387,46 @@ public final class StreamStatisticNames {
   public static final String BLOCKS_RELEASED
       = "blocks_released";
 
+  /**
+   * Total number of prefetching operations executed.
+   */
+  public static final String STREAM_READ_PREFETCH_OPERATIONS
+      = "stream_read_prefetch_operations";
+
+  /**
+   * Total number of block in disk cache.
+   */
+  public static final String STREAM_READ_BLOCKS_IN_FILE_CACHE
+      = "stream_read_blocks_in_cache";
+
+  /**
+   * Total number of active prefetch operations.
+   */
+  public static final String STREAM_READ_ACTIVE_PREFETCH_OPERATIONS
+      = "stream_read_active_prefetch_operations";
+
+  /**
+   * Total bytes of memory in use by this input stream.
+   */
+  public static final String STREAM_READ_ACTIVE_MEMORY_IN_USE
+      = "stream_read_active_memory_in_use";
+
+  /**
+   * count/duration of reading a remote block.
+   * IO.
+   * Value: {@value}.
+   */
+  public static final String STREAM_READ_REMOTE_BLOCK_READ
+      = "stream_read_block_read";
+
+  /**
+   * count/duration of acquiring a buffer and reading to it
+   * IO.
+   * Value: {@value}.
+   */
+  public static final String STREAM_READ_BLOCK_ACQUIRE_AND_READ
+      = "stream_read_block_read";
+
   private StreamStatisticNames() {
   }
 

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java

@@ -674,46 +674,6 @@ private Constants() {
   public static final String STREAM_READ_GAUGE_INPUT_POLICY =
       "stream_read_gauge_input_policy";
 
-  /**
-   * Total number of prefetching operations executed.
-   */
-  public static final String STREAM_READ_PREFETCH_OPERATIONS
-      = "stream_read_prefetch_operations";
-
-  /**
-   * Total number of block in disk cache.
-   */
-  public static final String STREAM_READ_BLOCKS_IN_FILE_CACHE
-      = "stream_read_blocks_in_cache";
-
-  /**
-   * Total number of active prefetch operations.
-   */
-  public static final String STREAM_READ_ACTIVE_PREFETCH_OPERATIONS
-      = "stream_read_active_prefetch_operations";
-
-  /**
-   * Total bytes of memory in use by this input stream.
-   */
-  public static final String STREAM_READ_ACTIVE_MEMORY_IN_USE
-      = "stream_read_active_memory_in_use";
-
-  /**
-   * count/duration of reading a remote block.
-   * IO.
-   * Value: {@value}.
-   */
-  public static final String STREAM_READ_REMOTE_BLOCK_READ
-      = "stream_read_block_read";
-
-  /**
-   * count/duration of acquiring a buffer and reading to it
-   * IO.
-   * Value: {@value}.
-   */
-  public static final String STREAM_READ_BLOCK_ACQUIRE_AND_READ
-      = "stream_read_block_read";
-
   @InterfaceAudience.Private
   @InterfaceStability.Unstable
   public static final String S3_CLIENT_FACTORY_IMPL =

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AInstrumentation.java

@@ -72,10 +72,10 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 
-import static org.apache.hadoop.fs.s3a.Constants.STREAM_READ_BLOCK_ACQUIRE_AND_READ;
 import static org.apache.hadoop.fs.s3a.Constants.STREAM_READ_GAUGE_INPUT_POLICY;
-import static org.apache.hadoop.fs.s3a.Constants.STREAM_READ_PREFETCH_OPERATIONS;
-import static org.apache.hadoop.fs.s3a.Constants.STREAM_READ_REMOTE_BLOCK_READ;
+import static org.apache.hadoop.fs.statistics.StreamStatisticNames.STREAM_READ_BLOCK_ACQUIRE_AND_READ;
+import static org.apache.hadoop.fs.statistics.StreamStatisticNames.STREAM_READ_PREFETCH_OPERATIONS;
+import static org.apache.hadoop.fs.statistics.StreamStatisticNames.STREAM_READ_REMOTE_BLOCK_READ;
 import static org.apache.hadoop.fs.statistics.IOStatisticsLogging.demandStringifyIOStatistics;
 import static org.apache.hadoop.fs.statistics.IOStatisticsSupport.snapshotIOStatistics;
 import static org.apache.hadoop.fs.statistics.StoreStatisticNames.ACTION_EXECUTOR_ACQUIRED;

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Statistic.java

@@ -358,15 +358,15 @@ public enum Statistic {
       "Total count of bytes read from an input stream",
       TYPE_COUNTER),
   STREAM_READ_BLOCKS_IN_FILE_CACHE(
-      Constants.STREAM_READ_BLOCKS_IN_FILE_CACHE,
+      StreamStatisticNames.STREAM_READ_BLOCKS_IN_FILE_CACHE,
       "Gauge of blocks in disk cache",
       TYPE_GAUGE),
   STREAM_READ_ACTIVE_PREFETCH_OPERATIONS(
-      Constants.STREAM_READ_ACTIVE_PREFETCH_OPERATIONS,
+      StreamStatisticNames.STREAM_READ_ACTIVE_PREFETCH_OPERATIONS,
       "Gauge of active prefetches",
       TYPE_GAUGE),
   STREAM_READ_ACTIVE_MEMORY_IN_USE(
-      Constants.STREAM_READ_ACTIVE_MEMORY_IN_USE,
+      StreamStatisticNames.STREAM_READ_ACTIVE_MEMORY_IN_USE,
       "Gauge of active memory in use",
       TYPE_GAUGE),
 

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/read/S3CachingInputStream.java

@@ -33,7 +33,7 @@
 import org.apache.hadoop.fs.s3a.S3ObjectAttributes;
 import org.apache.hadoop.fs.s3a.statistics.S3AInputStreamStatistics;
 
-import static org.apache.hadoop.fs.s3a.Constants.STREAM_READ_BLOCK_ACQUIRE_AND_READ;
+import static org.apache.hadoop.fs.statistics.StreamStatisticNames.STREAM_READ_BLOCK_ACQUIRE_AND_READ;
 import static org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding.invokeTrackingDuration;
 
 /**

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/read/S3Reader.java

@@ -33,7 +33,7 @@
 import org.apache.hadoop.fs.s3a.Invoker;
 import org.apache.hadoop.fs.s3a.statistics.S3AInputStreamStatistics;
 
-import static org.apache.hadoop.fs.s3a.Constants.STREAM_READ_REMOTE_BLOCK_READ;
+import static org.apache.hadoop.fs.statistics.StreamStatisticNames.STREAM_READ_REMOTE_BLOCK_READ;
 import static org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding.trackDurationOfOperation;
 
 /**

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/common/EmptyPrefetchingStatistics.java

@@ -0,0 +1,59 @@
+ /*
+  * Licensed to the Apache Software Foundation (ASF) under one
+  * or more contributor license agreements. See the NOTICE file
+  * distributed with this work for additional information
+  * regarding copyright ownership. The ASF licenses this file
+  * to you under the Apache License, Version 2.0 (the
+  * "License"); you may not use this file except in compliance
+  * with the License. You may obtain a copy of the License at
+  *
+  *   http://www.apache.org/licenses/LICENSE-2.0
+  *
+  * Unless required by applicable law or agreed to in writing,
+  * software distributed under the License is distributed on an
+  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+  * KIND, either express or implied. See the License for the
+  * specific language governing permissions and limitations
+  * under the License.
+  */
+
+package org.apache.hadoop.fs.common;
+
+import java.time.Duration;
+
+public final class EmptyPrefetchingStatistics implements PrefetchingStatistics {
+  @Override
+  public void prefetchOperationStarted() {
+
+  }
+
+  @Override
+  public void blockAddedToFileCache() {
+
+  }
+
+  @Override
+  public void blockRemovedFromFileCache() {
+
+  }
+
+  @Override
+  public void prefetchOperationCompleted() {
+
+  }
+
+  @Override
+  public void executorAcquired(Duration timeInQueue) {
+
+  }
+
+  @Override
+  public void memoryAllocated(int size) {
+
+  }
+
+  @Override
+  public void memoryFreed(int size) {
+
+  }
+}

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/common/TestBlockCache.java

@@ -40,7 +40,7 @@ public class TestBlockCache extends AbstractHadoopTestBase {
   public void testArgChecks() throws Exception {
     // Should not throw.
     BlockCache cache =
-        new SingleFilePerBlockCache(EmptyS3AStatisticsContext.EMPTY_INPUT_STREAM_STATISTICS);
+        new SingleFilePerBlockCache(new EmptyPrefetchingStatistics());
 
     ByteBuffer buffer = ByteBuffer.allocate(16);