npe in bandwidth

Author: steveloughran

src/main/java/org/apache/hadoop/fs/store/commands/Bandwidth.java

@@ -20,13 +20,17 @@
 
 import java.io.IOException;
 import java.io.PrintStream;
+import java.security.MessageDigest;
 import java.time.Duration;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Locale;
 import java.util.Optional;
 import java.util.Random;
 import java.util.concurrent.atomic.AtomicLong;
 
+import javax.annotation.Nullable;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -95,6 +99,8 @@ public class Bandwidth extends StoreEntryPoint {
 
   public static final String DEFAULT_READ_POLICY = "whole-file";
 
+  public static final String DIGEST_ALGORITHM = "SHA-256";
+
   private byte[] dataBuffer;
 
   public Bandwidth() {
@@ -173,10 +179,13 @@ public int run(String[] args) throws Exception {
     println("Writing data as %,d blocks each of size %,d bytes", numberOfBuffersToUpload,
         blockSize);
 
+    MessageDigest uploadDigest =  MessageDigest.getInstance(DIGEST_ALGORITHM);
+    MessageDigest downloadDigest =  MessageDigest.getInstance(DIGEST_ALGORITHM);
+
     /*
       prepare the CSV output if requested
     */
-    CsvWriterWithCRC writer = null;
+    CsvWriterWithCRC csvWriter = null;
     Path csvPath = null;
     if (csvFile != null) {
       csvPath = new Path(csvFile);
@@ -186,10 +195,10 @@ public int run(String[] args) throws Exception {
           .bufferSize(BUFFER_SIZE)
           .overwrite(true)
           .build();
-      writer = new CsvWriterWithCRC(upload, SEPARATOR, EOL, true);
+      csvWriter = new CsvWriterWithCRC(upload, SEPARATOR, EOL, true);
 
-      writer.columns("operation", "iteration", "bytes", "total", "duration");
-      writer.newline();
+      csvWriter.columns("operation", "iteration", "bytes", "total", "duration");
+      csvWriter.newline();
     }
 
     // buffer of randomness
@@ -218,7 +227,7 @@ public int run(String[] args) throws Exception {
           .overwrite(true)
           .build();
       d.finished();
-      row(writer, "create-file", 1, 0, 0, d);
+      row(csvWriter, "create-file", 1, 0, 0, d);
     }
     // set up
     if (!keep) {
@@ -249,6 +258,7 @@ public int run(String[] args) throws Exception {
         StoreDurationInfo duration = new StoreDurationInfo();
         print("Write block %,d", i);
         upload.write(dataBuffer);
+        uploadDigest.update(dataBuffer);
         if (flush) {
           upload.flush();
         }
@@ -258,7 +268,7 @@ public int run(String[] args) throws Exception {
         duration.finished();
         blockUploads.add(duration.value());
         println(" in %.3f seconds", duration.value() / 1000.0);
-        row(writer, "upload-block",  i + 1, blockSize, total += blockSize, duration);
+        row(csvWriter, "upload-block",  i + 1, blockSize, total += blockSize, duration);
       }
       println();
 
@@ -270,7 +280,7 @@ public int run(String[] args) throws Exception {
       } finally {
         closeDuration.close();
       }
-      row(writer, "close-upload", 1, 0, fileSizeBytes, closeDuration);
+      row(csvWriter, "close-upload", 1, 0, fileSizeBytes, closeDuration);
 
       println();
       // print out progress info
@@ -282,7 +292,7 @@ public int run(String[] args) throws Exception {
     }
     // upload is done, print some statistics
     uploadDurationTracker.finished();
-    row(writer, "upload", 1, blockSize, blockSize, uploadDurationTracker);
+    row(csvWriter, "upload", 1, blockSize, blockSize, uploadDurationTracker);
 
     // end of upload
     printFSInfoInVerbose(fs);
@@ -300,7 +310,7 @@ public int run(String[] args) throws Exception {
         fs.rename(uploadPath, downloadPath);
       }
       rd.finished();
-      row(writer, "rename", 1, fileSizeBytes, 0, rd);
+      row(csvWriter, "rename", 1, fileSizeBytes, 0, rd);
     }
 
     /*
@@ -318,7 +328,7 @@ public int run(String[] args) throws Exception {
           .build().get();
     } finally {
       openDuration.finished();
-      row(writer, "open-for-download", 1, 0, 0, openDuration);
+      row(csvWriter, "open-for-download", 1, 0, 0, openDuration);
     }
     MinMeanMax blockDownload = new MinMeanMax("block read duration");
 
@@ -330,20 +340,25 @@ public int run(String[] args) throws Exception {
         print("Read block %,d", i);
         StoreDurationInfo duration = new StoreDurationInfo();
         download.readFully(pos, dataBuffer);
+        downloadDigest.update(dataBuffer);
+
         pos += blockSize;
         duration.finished();
         blockDownload.add(duration.value());
         println(" in %.3f seconds", duration.value() / 1000.0);
-        row(writer, "download-block", i + 1, blockSize, total += blockSize, duration);
+        row(csvWriter, "download-block", i + 1, blockSize, total += blockSize, duration);
       }
       println();
       try (StoreDurationInfo d = new StoreDurationInfo(out, "download stream close()")) {
         download.close();
       }
       downloadDurationTracker.finished();
-      row(writer, "download", 1, fileSizeBytes, fileSizeBytes, downloadDurationTracker);
+      row(csvWriter, "download", 1, fileSizeBytes, fileSizeBytes, downloadDurationTracker);
     } finally {
-      writer.close();
+      if (csvWriter != null) {
+        csvWriter.flush();
+        csvWriter.close();
+      }
       printIfVerbose("Download Stream: %s", download);
     }
 
@@ -383,12 +398,25 @@ public int run(String[] args) throws Exception {
     summarize("Download", downloadDurationTracker, fileSizeBytes,
         "Blocks downloaded:", blockDownload);
 
+    int exitCode = 0;
+
+    final byte[] uploadHash = uploadDigest.digest();
+    final byte[] downloadHash = downloadDigest.digest();
+    if (!Arrays.equals(uploadHash, downloadHash)) {
+      errorln("Upload hash does not match download hash: data corrupted!");
+      exitCode = -1;
+    } else {
+      println("Data checksums match: the data has not been corrupted during the test");
+    }
+
+
     if (csvPath != null) {
       print("CSV formatted data saved to %s", csvPath);
     }
+
     println();
 
-    return 0;
+    return exitCode;
 
   }
 
@@ -403,7 +431,7 @@ public int run(String[] args) throws Exception {
    * @throws IOException write failure
    */
   private static void row(
-      final CsvWriterWithCRC writer,
+      @Nullable final CsvWriterWithCRC writer,
       final String a,
       final int iteration,
       final long opBytes,

src/main/java/org/apache/hadoop/fs/store/diag/StoreDiag.java

@@ -82,7 +82,6 @@
 import static org.apache.hadoop.fs.store.CommonParameters.STANDARD_OPTS;
 import static org.apache.hadoop.fs.store.CommonParameters.SYSPROP;
 import static org.apache.hadoop.fs.store.CommonParameters.TOKENFILE;
-import static org.apache.hadoop.fs.store.CommonParameters.VERBOSE;
 import static org.apache.hadoop.fs.store.StoreExitCodes.E_ERROR;
 import static org.apache.hadoop.fs.store.StoreExitCodes.E_NOT_FOUND;
 import static org.apache.hadoop.fs.store.StoreExitCodes.E_NO_ACCESS;