Skip to content

HDFS-17293. First packet data + checksum size will be set to 516 bytes when writing to a new block. #6368

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
zhangshuyan0 merged 7 commits into from
Jan 22, 2024
Merged

HDFS-17293. First packet data + checksum size will be set to 516 bytes when writing to a new block. #6368

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
9a4a6ba
HDFS-17293. First packet data + checksum size will be set to 516 byte…
hfutatzhanghb Dec 18, 2023
338aa6f
add unit test.
hfutatzhanghb Dec 27, 2023
87ae30a
fix checkstyle.
hfutatzhanghb Dec 28, 2023
c1ba0b5
optimize united tests.
hfutatzhanghb Dec 28, 2023
bad42b7
trigger yetus again.
hfutatzhanghb Dec 28, 2023
38d52f6
fix checkstyle
hfutatzhanghb Dec 28, 2023
96da2c8
fix unit test
hfutatzhanghb Jan 19, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 7 additions & 2 deletions ...hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSOutputStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -536,8 +536,13 @@ protected void adjustChunkBoundary() {
}

if (!getStreamer().getAppendChunk()) {
final int psize = (int) Math
.min(blockSize - getStreamer().getBytesCurBlock(), writePacketSize);
int psize = 0;
if (blockSize == getStreamer().getBytesCurBlock()) {
psize = writePacketSize;
} else {
psize = (int) Math
.min(blockSize - getStreamer().getBytesCurBlock(), writePacketSize);
}
computePacketChunkSize(psize, bytesPerChecksum);
}
}
Expand Down
41 changes: 41 additions & 0 deletions ...op-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSOutputStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,7 @@
import org.apache.hadoop.hdfs.protocol.HdfsConstants;
import org.apache.hadoop.hdfs.protocol.HdfsFileStatus;
import org.apache.hadoop.hdfs.protocol.datatransfer.BlockConstructionStage;
import org.apache.hadoop.hdfs.protocol.datatransfer.PacketHeader;
import org.apache.hadoop.hdfs.protocol.datatransfer.PacketReceiver;
import org.apache.hadoop.hdfs.server.blockmanagement.BlockManager;
import org.apache.hadoop.hdfs.server.blockmanagement.DatanodeDescriptor;
Expand All @@ -58,6 +59,7 @@
import org.apache.hadoop.test.GenericTestUtils;
import org.apache.hadoop.test.PathUtils;
import org.apache.hadoop.test.Whitebox;
import org.apache.hadoop.util.DataChecksum;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.BeforeClass;
Expand Down Expand Up @@ -508,6 +510,45 @@ public void testExceptionInCloseWithoutRecoverLease() throws Exception {
}
}

@Test(timeout=60000)
public void testFirstPacketSizeInNewBlocks() throws IOException {
final long blockSize = (long) 1024 * 1024;
MiniDFSCluster dfsCluster = cluster;
DistributedFileSystem fs = dfsCluster.getFileSystem();
Configuration dfsConf = fs.getConf();

EnumSet<CreateFlag> flags = EnumSet.of(CreateFlag.CREATE);
try(FSDataOutputStream fos = fs.create(new Path("/testfile.dat"),
FsPermission.getDefault(),
flags, 512, (short)3, blockSize, null)) {

DataChecksum crc32c = DataChecksum.newDataChecksum(
DataChecksum.Type.CRC32C, 512);

long loop = 0;
Random r = new Random();
byte[] buf = new byte[(int) blockSize];
r.nextBytes(buf);
fos.write(buf);
fos.hflush();

int chunkSize = crc32c.getBytesPerChecksum() + crc32c.getChecksumSize();
int packetContentSize = (dfsConf.getInt(DFS_CLIENT_WRITE_PACKET_SIZE_KEY,
DFS_CLIENT_WRITE_PACKET_SIZE_DEFAULT) -
PacketHeader.PKT_MAX_HEADER_LEN) / chunkSize * chunkSize;

while (loop < 20) {
r.nextBytes(buf);
fos.write(buf);
fos.hflush();
loop++;
Assert.assertEquals(((DFSOutputStream) fos.getWrappedStream()).packetSize,
packetContentSize);
}
}
fs.delete(new Path("/testfile.dat"), true);
}

@AfterClass
public static void tearDown() {
if (cluster != null) {
Expand Down