test_class.file: smart-hadoop-support/smart-hadoop/src/test/java/org/smartdata/hdfs/action/TestCompressionAction.java
test_class.identifier: TestCompressionAction
test_class.superclass: extends MiniClusterHarness
test_case.body: @Test
    public void testExecute() throws Exception {
      String filePath = "/testCompressFile/fadsfa/213";
      int bufferSize = 1024 * 128;
  //    String compressionImpl = "Lz4";
  //    String compressionImpl = "Bzip2";
  //    String compressionImpl = "Zlib";
      byte[] bytes = TestCompressionAction.BytesGenerator.get(bufferSize);
      short replication = 3;
      long blockSize = DEFAULT_BLOCK_SIZE;
  
      // Create HDFS file
      OutputStream outputStream = dfsClient.create(filePath, true,
        replication, blockSize);
      outputStream.write(bytes);
      outputStream.close();
      dfsClient.setStoragePolicy(filePath, "COLD");
      HdfsFileStatus srcFileStatus = dfsClient.getFileInfo(filePath);
  
      // Generate compressed file
      String bufferSizeForCompression = "10MB";
      compression(filePath, bufferSizeForCompression);
  
      // Check HdfsFileStatus
      HdfsFileStatus fileStatus = dfsClient.getFileInfo(filePath);
      Assert.assertEquals(replication, fileStatus.getReplication());
      Assert.assertEquals(blockSize, fileStatus.getBlockSize());
  
      // 0 means unspecified.
      if (srcFileStatus.getStoragePolicy() != 0) {
        // To make sure the consistency of storage policy
        Assert.assertEquals(srcFileStatus.getStoragePolicy(),
            fileStatus.getStoragePolicy());
      }
    }
test_case.class_method_signature: TestCompressionAction.testExecute()
test_case.full_signature: @Test public void testExecute()
test_case.identifier: testExecute
test_case.invocations: get create write close setStoragePolicy getFileInfo compression getFileInfo assertEquals getReplication assertEquals getBlockSize getStoragePolicy assertEquals getStoragePolicy getStoragePolicy
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testExecute()
focal_class.declarator: LOG =
        LoggerFactory.getLogger(CompressionAction.class)
focal_class.modifier: private static final
focal_class.original_string: private static final Logger LOG =
        LoggerFactory.getLogger(CompressionAction.class);
focal_class.type: Logger
focal_class.var_name: LOG
focal_class.declarator: BUF_SIZE = "-bufSize"
focal_class.modifier: public static final
focal_class.original_string: public static final String BUF_SIZE = "-bufSize";
focal_class.type: String
focal_class.var_name: BUF_SIZE
focal_class.declarator: CODEC = "-codec"
focal_class.modifier: public static final
focal_class.original_string: public static final String CODEC = "-codec";
focal_class.type: String
focal_class.var_name: CODEC
focal_class.declarator: compressionCodecList = CompressionCodec.CODEC_LIST
focal_class.modifier: private static
focal_class.original_string: private static List<String> compressionCodecList = CompressionCodec.CODEC_LIST;
focal_class.type: List<String>
focal_class.var_name: compressionCodecList
focal_class.declarator: filePath
focal_class.modifier: private
focal_class.original_string: private String filePath;
focal_class.type: String
focal_class.var_name: filePath
focal_class.declarator: conf
focal_class.modifier: private
focal_class.original_string: private Configuration conf;
focal_class.type: Configuration
focal_class.var_name: conf
focal_class.declarator: progress
focal_class.modifier: private
focal_class.original_string: private MutableFloat progress;
focal_class.type: MutableFloat
focal_class.var_name: progress
focal_class.declarator: bufferSize = 1024 * 1024
focal_class.modifier: private
focal_class.original_string: private int bufferSize = 1024 * 1024;
focal_class.type: int
focal_class.var_name: bufferSize
focal_class.declarator: maxSplit
focal_class.modifier: private
focal_class.original_string: private int maxSplit;
focal_class.type: int
focal_class.var_name: maxSplit
focal_class.declarator: compressCodec
focal_class.modifier: private
focal_class.original_string: private String compressCodec;
focal_class.type: String
focal_class.var_name: compressCodec
focal_class.declarator: userDefinedBufferSize
focal_class.modifier: private
focal_class.original_string: private int userDefinedBufferSize;
focal_class.type: int
focal_class.var_name: userDefinedBufferSize
focal_class.declarator: XATTR_NAME =
        SmartConstants.SMART_FILE_STATE_XATTR_NAME
focal_class.modifier: public static final
focal_class.original_string: public static final String XATTR_NAME =
        SmartConstants.SMART_FILE_STATE_XATTR_NAME;
focal_class.type: String
focal_class.var_name: XATTR_NAME
focal_class.declarator: compressionFileInfo
focal_class.modifier: private
focal_class.original_string: private CompressionFileInfo compressionFileInfo;
focal_class.type: CompressionFileInfo
focal_class.var_name: compressionFileInfo
focal_class.declarator: compressionFileState
focal_class.modifier: private
focal_class.original_string: private CompressionFileState compressionFileState;
focal_class.type: CompressionFileState
focal_class.var_name: compressionFileState
focal_class.declarator: compressTmpPath
focal_class.modifier: private
focal_class.original_string: private String compressTmpPath;
focal_class.type: String
focal_class.var_name: compressTmpPath
focal_class.declarator: COMPRESS_TMP = "-compressTmp"
focal_class.modifier: public static final
focal_class.original_string: public static final String COMPRESS_TMP = "-compressTmp";
focal_class.type: String
focal_class.var_name: COMPRESS_TMP
focal_class.file: smart-hadoop-support/smart-hadoop/src/main/java/org/smartdata/hdfs/action/CompressionAction.java
focal_class.identifier: CompressionAction
focal_class.class_method_signature: CompressionAction.init(Map<String, String> args)
focal_class.full_signature: @Override public void init(Map<String, String> args)
focal_class.identifier: init
focal_class.modifiers: @Override public
focal_class.parameters: (Map<String, String> args)
focal_class.return: void
focal_class.signature: void init(Map<String, String> args)
focal_class.class_method_signature: CompressionAction.execute()
focal_class.full_signature: @Override protected void execute()
focal_class.identifier: execute
focal_class.modifiers: @Override protected
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void execute()
focal_class.class_method_signature: CompressionAction.compress(InputStream inputStream, OutputStream outputStream)
focal_class.full_signature: private void compress(InputStream inputStream, OutputStream outputStream)
focal_class.identifier: compress
focal_class.modifiers: private
focal_class.parameters: (InputStream inputStream, OutputStream outputStream)
focal_class.return: void
focal_class.signature: void compress(InputStream inputStream, OutputStream outputStream)
focal_class.class_method_signature: CompressionAction.getActualBuffSize(long fileSize)
focal_class.full_signature: private int getActualBuffSize(long fileSize)
focal_class.identifier: getActualBuffSize
focal_class.modifiers: private
focal_class.parameters: (long fileSize)
focal_class.return: int
focal_class.signature: int getActualBuffSize(long fileSize)
focal_class.class_method_signature: CompressionAction.getProgress()
focal_class.full_signature: @Override public float getProgress()
focal_class.identifier: getProgress
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: float
focal_class.signature: float getProgress()
focal_class.superclass: extends HdfsAction
focal_method.body: @Override
    protected void execute() throws Exception {
      if (filePath == null) {
        throw new IllegalArgumentException("File path is missing.");
      }
      if (compressTmpPath == null) {
        throw new IllegalArgumentException("Compression tmp path is not specified!");
      }
      if (!compressionCodecList.contains(compressCodec)) {
        throw new ActionException(
            "Compression Action failed due to unsupported codec: " + compressCodec);
      }
      appendLog(
          String.format("Compression Action started at %s for %s",
              Utils.getFormatedCurrentTime(), filePath));
  
      if (!dfsClient.exists(filePath)) {
        throw new ActionException(
            "Failed to execute Compression Action: the given file doesn't exist!");
      }
  
      // Consider directory case.
      if (dfsClient.getFileInfo(filePath).isDir()) {
        appendLog("Compression is not applicable to a directory.");
        return;
      }
  
      // Generate compressed file
      HdfsFileStatus srcFile = dfsClient.getFileInfo(filePath);
      compressionFileState = new CompressionFileState(filePath, bufferSize, compressCodec);
      compressionFileState.setOriginalLength(srcFile.getLen());
  
      OutputStream appendOut = null;
      DFSInputStream in = null;
      OutputStream out = null;
      try {
        if (srcFile.getLen() == 0) {
          compressionFileInfo = new CompressionFileInfo(false, compressionFileState);
        } else {
          short replication = srcFile.getReplication();
          long blockSize = srcFile.getBlockSize();
          long fileSize = srcFile.getLen();
          appendLog("File length: " + fileSize);
          bufferSize = getActualBuffSize(fileSize);
  
          // SmartDFSClient will fail to open compressing file with PROCESSING FileStage
          // set by Compression scheduler. But considering DfsClient may be used, we use
          // append operation to lock the file to avoid any modification.
          appendOut = CompatibilityHelperLoader.getHelper().
              getDFSClientAppend(dfsClient, filePath, bufferSize);
          in = dfsClient.open(filePath);
          out = dfsClient.create(compressTmpPath,
              true, replication, blockSize);
  
          // Keep storage policy consistent.
          // The below statement is not supported on Hadoop-2.7.3 or CDH-5.10.1
          // String storagePolicyName = dfsClient.getStoragePolicy(filePath).getName();
          byte storagePolicyId = dfsClient.getFileInfo(filePath).getStoragePolicy();
          String storagePolicyName = SmartConstants.STORAGE_POLICY_MAP.get(storagePolicyId);
          if (!storagePolicyName.equals("UNDEF")) {
            dfsClient.setStoragePolicy(compressTmpPath, storagePolicyName);
          }
  
          compress(in, out);
          HdfsFileStatus destFile = dfsClient.getFileInfo(compressTmpPath);
          dfsClient.setOwner(compressTmpPath, srcFile.getOwner(), srcFile.getGroup());
          dfsClient.setPermission(compressTmpPath, srcFile.getPermission());
          compressionFileState.setCompressedLength(destFile.getLen());
          appendLog("Compressed file length: " + destFile.getLen());
          compressionFileInfo =
              new CompressionFileInfo(true, compressTmpPath, compressionFileState);
        }
        compressionFileState.setBufferSize(bufferSize);
        appendLog("Compression buffer size: " + bufferSize);
        appendLog("Compression codec: " + compressCodec);
        String compressionInfoJson = new Gson().toJson(compressionFileInfo);
        appendResult(compressionInfoJson);
        LOG.warn(compressionInfoJson);
        if (compressionFileInfo.needReplace()) {
          // Add to temp path
          // Please make sure content write to Xatte is less than 64K
          dfsClient.setXAttr(compressionFileInfo.getTempPath(),
              XATTR_NAME, SerializationUtils.serialize(compressionFileState),
              EnumSet.of(XAttrSetFlag.CREATE));
          // Rename operation is moved from CompressionScheduler.
          // Thus, modification for original file will be avoided.
          dfsClient.rename(compressTmpPath, filePath, Options.Rename.OVERWRITE);
        } else {
          // Add to raw path
          dfsClient.setXAttr(filePath,
              XATTR_NAME, SerializationUtils.serialize(compressionFileState),
              EnumSet.of(XAttrSetFlag.CREATE));
        }
      } catch (IOException e) {
        throw new IOException(e);
      } finally {
        if (appendOut != null) {
          try {
            appendOut.close();
          } catch (IOException e) {
            // Hide the expected exception that the original file is missing.
          }
        }
        if (in != null) {
          in.close();
        }
        if (out != null) {
          out.close();
        }
      }
    }
focal_method.class_method_signature: CompressionAction.execute()
focal_method.full_signature: @Override protected void execute()
focal_method.identifier: execute
focal_method.invocations: contains appendLog format getFormatedCurrentTime exists isDir getFileInfo appendLog getFileInfo setOriginalLength getLen getLen getReplication getBlockSize getLen appendLog getActualBuffSize getDFSClientAppend getHelper open create getStoragePolicy getFileInfo get equals setStoragePolicy compress getFileInfo setOwner getOwner getGroup setPermission getPermission setCompressedLength getLen appendLog getLen setBufferSize appendLog appendLog toJson appendResult warn needReplace setXAttr getTempPath serialize of rename setXAttr serialize of close close close
focal_method.modifiers: @Override protected
focal_method.parameters: ()
focal_method.return: void
focal_method.signature: void execute()