@Test public void testExecute() throws Exception { String filePath = "/testCompressFile/fadsfa/213"; int bufferSize = 1024 * 128; byte[] bytes = TestCompressionAction.BytesGenerator.get(bufferSize); short replication = 3; long blockSize = DEFAULT_BLOCK_SIZE; OutputStream outputStream = dfsClient.create(filePath, true, replication, blockSize); outputStream.write(bytes); outputStream.close(); dfsClient.setStoragePolicy(filePath, "COLD"); HdfsFileStatus srcFileStatus = dfsClient.getFileInfo(filePath); String bufferSizeForCompression = "10MB"; compression(filePath, bufferSizeForCompression); HdfsFileStatus fileStatus = dfsClient.getFileInfo(filePath); Assert.assertEquals(replication, fileStatus.getReplication()); Assert.assertEquals(blockSize, fileStatus.getBlockSize()); if (srcFileStatus.getStoragePolicy() != 0) { Assert.assertEquals(srcFileStatus.getStoragePolicy(), fileStatus.getStoragePolicy()); } }
@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!"); } if (dfsClient.getFileInfo(filePath).isDir()) { appendLog("Compression is not applicable to a directory."); return; } 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); appendOut = CompatibilityHelperLoader.getHelper(). getDFSClientAppend(dfsClient, filePath, bufferSize); in = dfsClient.open(filePath); out = dfsClient.create(compressTmpPath, true, replication, blockSize); 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()) { dfsClient.setXAttr(compressionFileInfo.getTempPath(), XATTR_NAME, SerializationUtils.serialize(compressionFileState), EnumSet.of(XAttrSetFlag.CREATE)); dfsClient.rename(compressTmpPath, filePath, Options.Rename.OVERWRITE); } else { 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) { } } if (in != null) { in.close(); } if (out != null) { out.close(); } } }
CompressionAction extends HdfsAction { @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!"); } if (dfsClient.getFileInfo(filePath).isDir()) { appendLog("Compression is not applicable to a directory."); return; } 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); appendOut = CompatibilityHelperLoader.getHelper(). getDFSClientAppend(dfsClient, filePath, bufferSize); in = dfsClient.open(filePath); out = dfsClient.create(compressTmpPath, true, replication, blockSize); 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()) { dfsClient.setXAttr(compressionFileInfo.getTempPath(), XATTR_NAME, SerializationUtils.serialize(compressionFileState), EnumSet.of(XAttrSetFlag.CREATE)); dfsClient.rename(compressTmpPath, filePath, Options.Rename.OVERWRITE); } else { 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) { } } if (in != null) { in.close(); } if (out != null) { out.close(); } } } }
CompressionAction extends HdfsAction { @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!"); } if (dfsClient.getFileInfo(filePath).isDir()) { appendLog("Compression is not applicable to a directory."); return; } 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); appendOut = CompatibilityHelperLoader.getHelper(). getDFSClientAppend(dfsClient, filePath, bufferSize); in = dfsClient.open(filePath); out = dfsClient.create(compressTmpPath, true, replication, blockSize); 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()) { dfsClient.setXAttr(compressionFileInfo.getTempPath(), XATTR_NAME, SerializationUtils.serialize(compressionFileState), EnumSet.of(XAttrSetFlag.CREATE)); dfsClient.rename(compressTmpPath, filePath, Options.Rename.OVERWRITE); } else { 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) { } } if (in != null) { in.close(); } if (out != null) { out.close(); } } }  }
CompressionAction extends HdfsAction { @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!"); } if (dfsClient.getFileInfo(filePath).isDir()) { appendLog("Compression is not applicable to a directory."); return; } 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); appendOut = CompatibilityHelperLoader.getHelper(). getDFSClientAppend(dfsClient, filePath, bufferSize); in = dfsClient.open(filePath); out = dfsClient.create(compressTmpPath, true, replication, blockSize); 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()) { dfsClient.setXAttr(compressionFileInfo.getTempPath(), XATTR_NAME, SerializationUtils.serialize(compressionFileState), EnumSet.of(XAttrSetFlag.CREATE)); dfsClient.rename(compressTmpPath, filePath, Options.Rename.OVERWRITE); } else { 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) { } } if (in != null) { in.close(); } if (out != null) { out.close(); } } }  @Override void init(Map<String, String> args); @Override float getProgress(); }
CompressionAction extends HdfsAction { @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!"); } if (dfsClient.getFileInfo(filePath).isDir()) { appendLog("Compression is not applicable to a directory."); return; } 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); appendOut = CompatibilityHelperLoader.getHelper(). getDFSClientAppend(dfsClient, filePath, bufferSize); in = dfsClient.open(filePath); out = dfsClient.create(compressTmpPath, true, replication, blockSize); 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()) { dfsClient.setXAttr(compressionFileInfo.getTempPath(), XATTR_NAME, SerializationUtils.serialize(compressionFileState), EnumSet.of(XAttrSetFlag.CREATE)); dfsClient.rename(compressTmpPath, filePath, Options.Rename.OVERWRITE); } else { 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) { } } if (in != null) { in.close(); } if (out != null) { out.close(); } } }  @Override void init(Map<String, String> args); @Override float getProgress(); static final String BUF_SIZE; static final String CODEC; static final String XATTR_NAME; static final String COMPRESS_TMP; }