checkstyle

Author: bbeaudreault

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/ClientContext.java

@@ -136,13 +136,13 @@ public class ClientContext {
   private volatile DeadNodeDetector deadNodeDetector = null;
 
   /**
-   * The switch for the {@link LocatedBlocksRefresher}
+   * The switch for the {@link LocatedBlocksRefresher}.
    */
   private final boolean locatedBlocksRefresherEnabled;
 
   /**
    * Periodically refresh the {@link org.apache.hadoop.hdfs.protocol.LocatedBlocks} backing
-   * registered {@link DFSInputStream}s, to take advantage of changes in block placement
+   * registered {@link DFSInputStream}s, to take advantage of changes in block placement.
    */
   private volatile LocatedBlocksRefresher locatedBlocksRefresher = null;
 
@@ -321,7 +321,7 @@ public DeadNodeDetector getDeadNodeDetector() {
 
   /**
    * If true, LocatedBlocksRefresher will be periodically refreshing LocatedBlocks
-   * of registered DFSInputStreams
+   * of registered DFSInputStreams.
    */
   public boolean isLocatedBlocksRefresherEnabled() {
     return locatedBlocksRefresherEnabled;

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSInputStream.java

@@ -307,10 +307,10 @@ private LocatedBlocks fetchAndCheckLocatedBlocks(LocatedBlocks existing)
     return newInfo;
   }
 
-  private long getLastBlockLength(LocatedBlocks locatedBlocks) throws IOException{
+  private long getLastBlockLength(LocatedBlocks blocks) throws IOException{
     long lastBlockBeingWrittenLength = 0;
-    if (!locatedBlocks.isLastBlockComplete()) {
-      final LocatedBlock last = locatedBlocks.getLastLocatedBlock();
+    if (!blocks.isLastBlockComplete()) {
+      final LocatedBlock last = blocks.getLastLocatedBlock();
       if (last != null) {
         if (last.getLocations().length == 0) {
           if (last.getBlockSize() == 0) {
@@ -1948,7 +1948,7 @@ protected void maybeRegisterBlockRefresh() {
   }
 
   /**
-   * De-register periodic refresh of this inputstream, if it was added to begin with
+   * De-register periodic refresh of this inputstream, if it was added to begin with.
    */
   private void maybeDeRegisterBlockRefresh() {
     if (refreshingBlockLocations.get()) {
@@ -1957,24 +1957,24 @@ private void maybeDeRegisterBlockRefresh() {
   }
 
   /**
-   * Refresh blocks for the input stream, if necessary
+   * Refresh blocks for the input stream, if necessary.
    *
    * @param addressCache optional map to use as a cache for resolving datanode InetSocketAddress
    * @return whether a refresh was performed or not
    */
   boolean refreshBlockLocations(Map<String, InetSocketAddress> addressCache) {
-    LocatedBlocks locatedBlocks;
+    LocatedBlocks blocks;
     synchronized (infoLock) {
-      locatedBlocks = getLocatedBlocks();
+      blocks = getLocatedBlocks();
     }
 
-    if (getLocalDeadNodes().isEmpty() && allBlocksLocal(locatedBlocks, addressCache)) {
+    if (getLocalDeadNodes().isEmpty() && allBlocksLocal(blocks, addressCache)) {
       return false;
     }
 
     try {
       DFSClient.LOG.debug("Refreshing {} for path {}", this, getSrc());
-      LocatedBlocks newLocatedBlocks = fetchAndCheckLocatedBlocks(locatedBlocks);
+      LocatedBlocks newLocatedBlocks = fetchAndCheckLocatedBlocks(blocks);
       long lastBlockLength = getLastBlockLength(newLocatedBlocks);
       if (lastBlockLength == -1) {
         DFSClient.LOG.debug(
@@ -1995,10 +1995,10 @@ boolean refreshBlockLocations(Map<String, InetSocketAddress> addressCache) {
    * Once new LocatedBlocks have been fetched, sets them on the DFSInputStream and
    * updates stateful read location within the necessary locks.
    */
-  private synchronized void setRefreshedValues(LocatedBlocks locatedBlocks, long lastBlockLength)
+  private synchronized void setRefreshedValues(LocatedBlocks blocks, long lastBlockLength)
       throws IOException {
     synchronized (infoLock) {
-      setLocatedBlocksFields(locatedBlocks, lastBlockLength);
+      setLocatedBlocksFields(blocks, lastBlockLength);
     }
 
     getLocalDeadNodes().clear();
@@ -2009,14 +2009,15 @@ private synchronized void setRefreshedValues(LocatedBlocks locatedBlocks, long l
     }
   }
 
-  private boolean allBlocksLocal(LocatedBlocks locatedBlocks, Map<String, InetSocketAddress> addressCache) {
+  private boolean allBlocksLocal(LocatedBlocks blocks,
+      Map<String, InetSocketAddress> addressCache) {
     if (addressCache == null) {
       addressCache = new HashMap<>();
     }
 
     // we only need to check the first location of each block, because the blocks are already
     // sorted by distance from the current host
-    for (LocatedBlock lb : locatedBlocks.getLocatedBlocks()) {
+    for (LocatedBlock lb : blocks.getLocatedBlocks()) {
       if (lb.getLocations().length == 0) {
         return false;
       }

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/LocatedBlocksRefresher.java

@@ -44,8 +44,9 @@
 import org.slf4j.LoggerFactory;
 
 /**
- * Periodically refresh the underlying cached {@link LocatedBlocks} for eligible registered {@link DFSInputStream}s.
- * DFSInputStreams are eligible for refreshing if they have any deadNodes or any blocks are lacking local replicas.
+ * Periodically refresh the underlying cached {@link LocatedBlocks} for eligible registered
+ * {@link DFSInputStream}s.  DFSInputStreams are eligible for refreshing if they have any
+ * deadNodes or any blocks are lacking local replicas.
  * Disabled by default, unless an interval is configured.
  */
 public class LocatedBlocksRefresher extends Daemon {
@@ -59,9 +60,10 @@ public class LocatedBlocksRefresher extends Daemon {
   private final long jitter;
   private final ExecutorService refreshThreadPool;
 
-  // Use WeakHashMap so that we don't hold onto references that might have not been explicitly closed
-  // because they were created and thrown away.
-  private final Set<DFSInputStream> registeredInputStreams = Collections.newSetFromMap(new WeakHashMap<>());
+  // Use WeakHashMap so that we don't hold onto references that might have not been explicitly
+  // closed because they were created and thrown away.
+  private final Set<DFSInputStream> registeredInputStreams =
+      Collections.newSetFromMap(new WeakHashMap<>());
 
   private int runCount;
   private int refreshCount;
@@ -116,7 +118,8 @@ public void run() {
         phaser.register();
         refreshThreadPool.submit(() -> {
           try {
-            if (isInputStreamTracked(inputStream) && inputStream.refreshBlockLocations(addressCache)) {
+            if (isInputStreamTracked(inputStream) &&
+                inputStream.refreshBlockLocations(addressCache)) {
               neededRefresh.incrementAndGet();
             }
           } finally {
@@ -175,11 +178,11 @@ public void shutdown() {
   }
 
   /**
-   * Collects the DFSInputStreams to a list within synchronization, so that we can iterate them without
-   * potentially blocking callers to {@link #addInputStream(DFSInputStream)} or
+   * Collects the DFSInputStreams to a list within synchronization, so that we can iterate them
+   * without potentially blocking callers to {@link #addInputStream(DFSInputStream)} or
    * {@link #removeInputStream(DFSInputStream)}. We don't care so much about missing additions,
-   * and we'll guard against removals by doing an additional {@link #isInputStreamTracked(DFSInputStream)}
-   * track during iteration.
+   * and we'll guard against removals by doing an additional
+   * {@link #isInputStreamTracked(DFSInputStream)} track during iteration.
    */
   private synchronized Collection<DFSInputStream> getInputStreams() {
     return new ArrayList<>(registeredInputStreams);

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSInputStreamBlockLocations.java

@@ -235,10 +235,10 @@ private void testWithRegistrationMethod(ThrowingConsumer registrationMethod) thr
   }
 
   private Path createFile(String fileName) throws IOException {
-    Path filePath = new Path(fileName);
-    try (FSDataOutputStream fout = fs.create(filePath, REPLICATION_FACTOR)) {
+    Path path = new Path(fileName);
+    try (FSDataOutputStream fout = fs.create(path, REPLICATION_FACTOR)) {
       fout.write(new byte[(fileLength)]);
     }
-    return filePath;
+    return path;
   }
-}
+}

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestLocatedBlocksRefresher.java

@@ -105,7 +105,8 @@ public void testDisabledOnZeroInterval() throws IOException {
   @Test
   public void testEnabledOnNonZeroInterval() throws Exception {
     setupTest(1000);
-    LocatedBlocksRefresher refresher = cluster.getFileSystem().getClient().getLocatedBlockRefresher();
+    LocatedBlocksRefresher refresher =
+        cluster.getFileSystem().getClient().getLocatedBlockRefresher();
     assertNotNull(refresher);
     assertNoMoreRefreshes(refresher);
   }
@@ -205,15 +206,17 @@ private void assertNoMoreRefreshes(LocatedBlocksRefresher refresher) throws Inte
     int runCount = refresher.getRunCount();
     int refreshCount = refresher.getRefreshCount();
 
-    LOG.info("Waiting for at least {} runs, from current {}, expecting no refreshes", runCount + 3, runCount);
+    LOG.info("Waiting for at least {} runs, from current {}, expecting no refreshes",
+        runCount + 3, runCount);
     // wait for it to run 3 times, with some buffer
     awaitWithTimeout(() -> refresher.getRunCount() > runCount + 3, 5 * interval);
 
     // it should not have refreshed anything, because no DFSInputStreams registered anymore
     assertEquals(refreshCount, refresher.getRefreshCount());
   }
 
-  private void assertRefreshes(LocatedBlocksRefresher refresher, int expectedRefreshes) throws InterruptedException {
+  private void assertRefreshes(LocatedBlocksRefresher refresher, int expectedRefreshes)
+      throws InterruptedException {
     int runCount = refresher.getRunCount();
     int refreshCount = refresher.getRefreshCount();
     int expectedRuns = 3;
@@ -222,18 +225,22 @@ private void assertRefreshes(LocatedBlocksRefresher refresher, int expectedRefre
       expectedRefreshes = expectedRuns;
     }
 
-    LOG.info("Waiting for at least {} runs, from current {}. Expecting {} refreshes, from current {}",
-        runCount + expectedRuns, runCount, refreshCount + expectedRefreshes, refreshCount);
+    LOG.info(
+        "Waiting for at least {} runs, from current {}. Expecting {} refreshes, from current {}",
+        runCount + expectedRuns, runCount, refreshCount + expectedRefreshes, refreshCount
+    );
 
     // wait for it to run 3 times
     awaitWithTimeout(() -> refresher.getRunCount() >= runCount + expectedRuns, 10_000);
 
-    // the values may not be identical due to any refreshes that occurred before we opened the DFSInputStream
-    // but the difference should be identical since we are refreshing every time
+    // the values may not be identical due to any refreshes that occurred before we opened
+    // the DFSInputStream but the difference should be identical since we are refreshing
+    // every time
     assertEquals(expectedRefreshes, refresher.getRefreshCount() - refreshCount);
   }
 
-  private void awaitWithTimeout(Supplier<Boolean> test, long timeoutMillis) throws InterruptedException {
+  private void awaitWithTimeout(Supplier<Boolean> test, long timeoutMillis)
+      throws InterruptedException {
     long now = Time.monotonicNow();
 
     while(!test.get()) {