steveloughran commented on a change in pull request #1899: HADOOP-16914 Adding
Output Stream Counters in ABFS
URL: https://github.com/apache/hadoop/pull/1899#discussion_r395751470
##########
File path:
hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAbfsOutputStream.java
##########
@@ -0,0 +1,278 @@
+package org.apache.hadoop.fs.azurebfs;
+
+import java.io.IOException;
+
+import org.junit.Test;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.azurebfs.services.AbfsOutputStream;
+import org.apache.hadoop.fs.permission.FsPermission;
+
+/**
+ * Test AbfsOutputStream statistics.
+ */
+public class ITestAbfsOutputStream extends AbstractAbfsIntegrationTest {
+
+ public ITestAbfsOutputStream() throws Exception {
+ }
+
+ /**
+ * Tests to check bytes Uploading in {@link AbfsOutputStream}.
+ *
+ * @throws IOException
+ */
+ @Test
+ public void testAbfsOutputStreamUploadingBytes() throws IOException {
+ describe("Testing Bytes uploaded in AbfsOutputSteam");
+ final AzureBlobFileSystem fs = getFileSystem();
+ Path uploadBytesFilePath = new Path("AbfsOutputStreamStatsPath");
+ AzureBlobFileSystemStore abfss = fs.getAbfsStore();
+ FileSystem.Statistics statistics = fs.getFsStatistics();
+ abfss.getAbfsConfiguration().setDisableOutputStreamFlush(false);
+ String testBytesToUpload = "bytes";
+
+ AbfsOutputStream outForSomeBytes = null;
+ try {
+ outForSomeBytes = (AbfsOutputStream)
abfss.createFile(uploadBytesFilePath,
+ statistics,
+ true,
+ FsPermission.getDefault(), FsPermission.getUMask(fs.getConf()));
+
+ //Test for zero bytes To upload
+ assertValues("bytes to upload", 0,
+ outForSomeBytes.getOutputStreamStatistics().bytesToUpload);
+
+ outForSomeBytes.write(testBytesToUpload.getBytes());
+ outForSomeBytes.flush();
+
+ //Test for some bytes to upload
+ assertValues("bytes to upload", testBytesToUpload.getBytes().length,
+ outForSomeBytes.getOutputStreamStatistics().bytesToUpload);
+
+ //Test for relation between bytesUploadSuccessful, bytesUploadFailed
+ // and bytesToUpload
+ assertValues("bytesUploadSuccessful equal to difference between "
+ + "bytesToUpload and bytesUploadFailed",
+ outForSomeBytes.getOutputStreamStatistics().bytesUploadSuccessful,
+ outForSomeBytes.getOutputStreamStatistics().bytesToUpload -
+ outForSomeBytes.getOutputStreamStatistics().bytesUploadFailed);
+
+ } finally {
+ if (outForSomeBytes != null) {
+ outForSomeBytes.close();
+ }
+ }
+
+ AbfsOutputStream outForLargeBytes = null;
+ try {
+ outForLargeBytes =
+ (AbfsOutputStream) abfss.createFile(uploadBytesFilePath,
+ statistics
+ , true, FsPermission.getDefault(),
+ FsPermission.getUMask(fs.getConf()));
+
+ int largeValue = 100000;
+ for (int i = 0; i < largeValue; i++) {
+ outForLargeBytes.write(testBytesToUpload.getBytes());
+ }
+ outForLargeBytes.flush();
+
+ //Test for large bytes to upload
+ assertValues("bytes to upload",
+ largeValue * (testBytesToUpload.getBytes().length),
+ outForLargeBytes.getOutputStreamStatistics().bytesToUpload);
+
+ //Test for relation between bytesUploadSuccessful, bytesUploadFailed
+ // and bytesToUpload
+ assertValues("bytesUploadSuccessful equal to difference between "
+ + "bytesToUpload and bytesUploadFailed",
+ outForSomeBytes.getOutputStreamStatistics().bytesUploadSuccessful,
+ outForSomeBytes.getOutputStreamStatistics().bytesToUpload -
+ outForSomeBytes.getOutputStreamStatistics().bytesUploadFailed);
+
+ } finally {
+ if (outForLargeBytes != null) {
+ outForLargeBytes.close();
+ }
+ }
+
+ }
+
+ /**
+ * Tests to check time spend on waiting for tasks to be complete on a
+ * blocking queue in {@link AbfsOutputStream}.
+ *
+ * @throws IOException
+ */
+ @Test
+ public void testAbfsOutputStreamTimeSpendOnWaitTask() throws IOException {
+ describe("Testing Time Spend on Waiting for Task to be complete");
+ final AzureBlobFileSystem fs = getFileSystem();
+ Path timeSpendFilePath = new Path("AbfsOutputStreamStatsPath");
+ AzureBlobFileSystemStore abfss = fs.getAbfsStore();
+ FileSystem.Statistics statistics = fs.getFsStatistics();
+
+ AbfsOutputStream out =
+ (AbfsOutputStream) abfss.createFile(timeSpendFilePath,
+ statistics, true,
+ FsPermission.getDefault(), FsPermission.getUMask(fs.getConf()));
+
+ }
+
+ /**
+ * Tests to check number of {@codes shrinkWriteOperationQueue()}
+ * calls.
+ * After writing data, AbfsOutputStream doesn't upload the data until
+ * Flushed. Hence, flush() method is called after write() to test Queue
+ * shrink calls.
+ *
+ * @throws IOException
+ */
+ @Test
+ public void testAbfsOutputStreamQueueShrink() throws IOException {
+ describe("Testing Queue Shrink calls in AbfsOutputStream");
+ final AzureBlobFileSystem fs = getFileSystem();
+ Path queueShrinkFilePath = new Path("AbfsOutputStreamStatsPath");
+ AzureBlobFileSystemStore abfss = fs.getAbfsStore();
+ abfss.getAbfsConfiguration().setDisableOutputStreamFlush(false);
+ FileSystem.Statistics statistics = fs.getFsStatistics();
+ String testQueueShrink = "testQueue";
+
+ AbfsOutputStream outForOneOp = null;
+
+ try {
+ outForOneOp =
+ (AbfsOutputStream) abfss.createFile(queueShrinkFilePath, statistics,
+ true,
+ FsPermission.getDefault(), FsPermission.getUMask(fs.getConf()));
+
+ //Test for shrinking Queue zero time
+ assertValues("number of queueShrink() Calls", 0,
+ outForOneOp.getOutputStreamStatistics().queueShrink);
+
+ outForOneOp.write(testQueueShrink.getBytes());
+ // Queue is shrunk 2 times when outStream is flushed
+ outForOneOp.flush();
+
+ //Test for shrinking Queue 2 times
+ assertValues("number of queueShrink() Calls", 2,
+ outForOneOp.getOutputStreamStatistics().queueShrink);
+
+ } finally {
+ if (outForOneOp != null) {
+ outForOneOp.close();
+ }
+ }
+
+ AbfsOutputStream outForLargeOps = null;
+
+ try {
+ outForLargeOps = (AbfsOutputStream) abfss.createFile(queueShrinkFilePath,
+ statistics, true,
+ FsPermission.getDefault(), FsPermission.getUMask(fs.getConf()));
+
+ int largeValue = 1000;
+ //QueueShrink is called 2 times in 1 flush(), hence 1000 flushes must
+ // give 2000 QueueShrink calls
+ for (int i = 0; i < largeValue; i++) {
+ outForLargeOps.write(testQueueShrink.getBytes());
+ //Flush is quite expensive so 1000 calls only which takes 1 min+
+ outForLargeOps.flush();
+ }
+
+ //Test for 2000 queue shrink calls
+ assertValues("number of queueShrink() Calls",
+ 2 * largeValue,
+ outForLargeOps.getOutputStreamStatistics().queueShrink);
+ } finally {
+ if (outForLargeOps != null) {
+ outForLargeOps.close();
+ }
+ }
+
+ }
+
+ /**
+ * Test to check number of {@codes writeCurrentBufferToService()}
Review comment:
@code
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
