@Test public void testComputePacketChunkSize() throws Exception { DistributedFileSystem fs = cluster.getFileSystem(); FSDataOutputStream os = fs.create(new Path("/test")); DFSOutputStream dos = (DFSOutputStream) Whitebox.getInternalState(os, "wrappedStream"); final int packetSize = 64*1024; final int bytesPerChecksum = 512; Method method = dos.getClass().getDeclaredMethod("computePacketChunkSize", int.class, int.class); method.setAccessible(true); method.invoke(dos, packetSize, bytesPerChecksum); Field field = dos.getClass().getDeclaredField("packetSize"); field.setAccessible(true); Assert.assertTrue((Integer) field.get(dos) + 33 < packetSize); Assert.assertTrue((Integer) field.get(dos) + 257 < packetSize); }
private void computePacketChunkSize(int psize, int csize) { final int bodySize = psize - PacketHeader.PKT_MAX_HEADER_LEN; final int chunkSize = csize + getChecksumSize(); chunksPerPacket = Math.max(bodySize/chunkSize, 1); packetSize = chunkSize*chunksPerPacket; if (DFSClient.LOG.isDebugEnabled()) { DFSClient.LOG.debug("computePacketChunkSize: src=" + src + ", chunkSize=" + chunkSize + ", chunksPerPacket=" + chunksPerPacket + ", packetSize=" + packetSize); } }
DFSOutputStream extends FSOutputSummer implements Syncable, CanSetDropBehind { private void computePacketChunkSize(int psize, int csize) { final int bodySize = psize - PacketHeader.PKT_MAX_HEADER_LEN; final int chunkSize = csize + getChecksumSize(); chunksPerPacket = Math.max(bodySize/chunkSize, 1); packetSize = chunkSize*chunksPerPacket; if (DFSClient.LOG.isDebugEnabled()) { DFSClient.LOG.debug("computePacketChunkSize: src=" + src + ", chunkSize=" + chunkSize + ", chunksPerPacket=" + chunksPerPacket + ", packetSize=" + packetSize); } } }
DFSOutputStream extends FSOutputSummer implements Syncable, CanSetDropBehind { private void computePacketChunkSize(int psize, int csize) { final int bodySize = psize - PacketHeader.PKT_MAX_HEADER_LEN; final int chunkSize = csize + getChecksumSize(); chunksPerPacket = Math.max(bodySize/chunkSize, 1); packetSize = chunkSize*chunksPerPacket; if (DFSClient.LOG.isDebugEnabled()) { DFSClient.LOG.debug("computePacketChunkSize: src=" + src + ", chunkSize=" + chunkSize + ", chunksPerPacket=" + chunksPerPacket + ", packetSize=" + packetSize); } } private  DFSOutputStream(DFSClient dfsClient, String src, Progressable progress,
      HdfsFileStatus stat, DataChecksum checksum); private  DFSOutputStream(DFSClient dfsClient, String src, HdfsFileStatus stat,
      EnumSet<CreateFlag> flag, Progressable progress,
      DataChecksum checksum, String[] favoredNodes); private  DFSOutputStream(DFSClient dfsClient, String src,
      EnumSet<CreateFlag> flags, Progressable progress, LocatedBlock lastBlock,
      HdfsFileStatus stat, DataChecksum checksum); }
DFSOutputStream extends FSOutputSummer implements Syncable, CanSetDropBehind { private void computePacketChunkSize(int psize, int csize) { final int bodySize = psize - PacketHeader.PKT_MAX_HEADER_LEN; final int chunkSize = csize + getChecksumSize(); chunksPerPacket = Math.max(bodySize/chunkSize, 1); packetSize = chunkSize*chunksPerPacket; if (DFSClient.LOG.isDebugEnabled()) { DFSClient.LOG.debug("computePacketChunkSize: src=" + src + ", chunkSize=" + chunkSize + ", chunksPerPacket=" + chunksPerPacket + ", packetSize=" + packetSize); } } private  DFSOutputStream(DFSClient dfsClient, String src, Progressable progress,
      HdfsFileStatus stat, DataChecksum checksum); private  DFSOutputStream(DFSClient dfsClient, String src, HdfsFileStatus stat,
      EnumSet<CreateFlag> flag, Progressable progress,
      DataChecksum checksum, String[] favoredNodes); private  DFSOutputStream(DFSClient dfsClient, String src,
      EnumSet<CreateFlag> flags, Progressable progress, LocatedBlock lastBlock,
      HdfsFileStatus stat, DataChecksum checksum); @VisibleForTesting synchronized DatanodeInfo[] getPipeline(); @Deprecated void sync(); @Override void hflush(); @Override void hsync(); void hsync(EnumSet<SyncFlag> syncFlags); @Deprecated synchronized int getNumCurrentReplicas(); synchronized int getCurrentBlockReplication(); @Override void close(); @VisibleForTesting void setArtificialSlowdown(long period); @VisibleForTesting synchronized void setChunksPerPacket(int value); long getInitialLen(); FileEncryptionInfo getFileEncryptionInfo(); @Override void setDropBehind(Boolean dropBehind); @VisibleForTesting long getFileId(); }
DFSOutputStream extends FSOutputSummer implements Syncable, CanSetDropBehind { private void computePacketChunkSize(int psize, int csize) { final int bodySize = psize - PacketHeader.PKT_MAX_HEADER_LEN; final int chunkSize = csize + getChecksumSize(); chunksPerPacket = Math.max(bodySize/chunkSize, 1); packetSize = chunkSize*chunksPerPacket; if (DFSClient.LOG.isDebugEnabled()) { DFSClient.LOG.debug("computePacketChunkSize: src=" + src + ", chunkSize=" + chunkSize + ", chunksPerPacket=" + chunksPerPacket + ", packetSize=" + packetSize); } } private  DFSOutputStream(DFSClient dfsClient, String src, Progressable progress,
      HdfsFileStatus stat, DataChecksum checksum); private  DFSOutputStream(DFSClient dfsClient, String src, HdfsFileStatus stat,
      EnumSet<CreateFlag> flag, Progressable progress,
      DataChecksum checksum, String[] favoredNodes); private  DFSOutputStream(DFSClient dfsClient, String src,
      EnumSet<CreateFlag> flags, Progressable progress, LocatedBlock lastBlock,
      HdfsFileStatus stat, DataChecksum checksum); @VisibleForTesting synchronized DatanodeInfo[] getPipeline(); @Deprecated void sync(); @Override void hflush(); @Override void hsync(); void hsync(EnumSet<SyncFlag> syncFlags); @Deprecated synchronized int getNumCurrentReplicas(); synchronized int getCurrentBlockReplication(); @Override void close(); @VisibleForTesting void setArtificialSlowdown(long period); @VisibleForTesting synchronized void setChunksPerPacket(int value); long getInitialLen(); FileEncryptionInfo getFileEncryptionInfo(); @Override void setDropBehind(Boolean dropBehind); @VisibleForTesting long getFileId();  }