diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/checksum/ECFileChecksumHelper.java b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/checksum/ECFileChecksumHelper.java index 1f6e18426aac..a26ca30dcf26 100644 --- a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/checksum/ECFileChecksumHelper.java +++ b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/checksum/ECFileChecksumHelper.java @@ -49,6 +49,11 @@ */ public class ECFileChecksumHelper extends BaseFileChecksumHelper { + // Blocks in the same EC placement group share identical node sets, so the + // STANDALONE checksum pipeline is the same for every block. Cache it to + // avoid rebuilding it (node filtering, UUID computation, object allocation) per block. + private final Map checksumPipelineCache = new HashMap<>(); + public ECFileChecksumHelper(OzoneVolume volume, OzoneBucket bucket, String keyName, long length, OzoneClientConfig.ChecksumCombineMode checksumCombineMode, ClientProtocol rpcClient, OmKeyInfo keyInfo) @@ -64,74 +69,69 @@ protected AbstractBlockChecksumComputer getBlockChecksumComputer(List getChunkInfos(OmKeyLocationInfo - keyLocationInfo) throws IOException { - // To read an EC block, we create a STANDALONE pipeline that contains the - // single location for the block index we want to read. The EC blocks are - // indexed from 1 to N, however the data locations are stored in the - // dataLocations array indexed from zero. + protected List getChunkInfos(OmKeyLocationInfo keyLocationInfo) throws IOException { Token token = keyLocationInfo.getToken(); BlockID blockID = keyLocationInfo.getBlockID(); + Pipeline ecPipeline = keyLocationInfo.getPipeline(); + Pipeline checksumPipeline = checksumPipelineCache.computeIfAbsent( + ecPipeline.getId(), id -> buildChecksumPipeline(ecPipeline)); - Pipeline pipeline = keyLocationInfo.getPipeline(); + List chunks; + XceiverClientSpi xceiverClientSpi = null; + try { + if (LOG.isDebugEnabled()) { + LOG.debug("Initializing BlockInputStream for get key to access {}", + blockID.getContainerID()); + } + xceiverClientSpi = getXceiverClientFactory().acquireClientForReadData(checksumPipeline); - List nodes = new ArrayList<>(); - Map selectedReplicaIndexes = new HashMap<>(); - ECReplicationConfig repConfig = (ECReplicationConfig) - pipeline.getReplicationConfig(); + ContainerProtos.GetBlockResponseProto response = ContainerProtocolCalls + .getBlock(xceiverClientSpi, blockID, token, checksumPipeline.getReplicaIndexes()); - for (DatanodeDetails dn : pipeline.getNodes()) { - int replicaIndex = pipeline.getReplicaIndex(dn); + chunks = response.getBlockData().getChunksList(); + } finally { + if (xceiverClientSpi != null) { + getXceiverClientFactory().releaseClientForReadData(xceiverClientSpi, false); + } + } + return chunks; + } + + // To read an EC block we need a STANDALONE pipeline containing only replica + // index 1 (which holds stripe checksums) and the parity nodes. Blocks in the + // same placement group share the same node set, so this result is cached by + // the caller and built at most once per placement group per file. + // + // The pipeline ID is derived deterministically from the sorted node UUIDs so + // that XceiverClientManager can reuse the same gRPC connection across files + // that land on the same EC placement group (cross-file reuse). + // Pipeline.newBuilder() is used instead of toBuilder() because toBuilder() + // copies the EC nodeStatus; the size mismatch when setNodes() is called + // triggers PipelineID.randomId() and defeats the deterministic ID. + private Pipeline buildChecksumPipeline(Pipeline ecPipeline) { + ECReplicationConfig repConfig = (ECReplicationConfig) ecPipeline.getReplicationConfig(); + List nodes = new ArrayList<>(); + Map replicaIndexes = new HashMap<>(); + for (DatanodeDetails dn : ecPipeline.getNodes()) { + int replicaIndex = ecPipeline.getReplicaIndex(dn); if (replicaIndex == 1 || replicaIndex > repConfig.getData()) { - // The stripe checksum we need to calculate checksums is only stored on - // replica_index = 1 and all the parity nodes. nodes.add(dn); - selectedReplicaIndexes.put(dn, replicaIndex); + replicaIndexes.put(dn, replicaIndex); } } - - // Build a deterministic pipeline ID from the sorted node UUIDs so that - // XceiverClientManager can cache and reuse the gRPC connection across files - // that share the same EC placement group (avoids a new connection per file). String nodeKey = nodes.stream() .map(DatanodeDetails::getUuidString) .sorted() .collect(Collectors.joining(",")); PipelineID deterministicId = PipelineID.valueOf( UUID.nameUUIDFromBytes(nodeKey.getBytes(StandardCharsets.UTF_8))); - - // Use Pipeline.newBuilder() (not toBuilder()) so that nodeStatus starts null. - // toBuilder() would copy the 5-node EC nodeStatus, causing setNodes(3 nodes) - // to detect the size mismatch and call PipelineID.randomId() -> SecureRandom - // even though setId(deterministicId) immediately overrides it. - pipeline = Pipeline.newBuilder() + return Pipeline.newBuilder() .setId(deterministicId) .setReplicationConfig(StandaloneReplicationConfig .getInstance(HddsProtos.ReplicationFactor.THREE)) - .setState(pipeline.getPipelineState()) + .setState(ecPipeline.getPipelineState()) .setNodes(nodes) - .setReplicaIndexes(selectedReplicaIndexes) + .setReplicaIndexes(replicaIndexes) .build(); - - List chunks; - XceiverClientSpi xceiverClientSpi = null; - try { - if (LOG.isDebugEnabled()) { - LOG.debug("Initializing BlockInputStream for get key to access {}", - blockID.getContainerID()); - } - xceiverClientSpi = getXceiverClientFactory().acquireClientForReadData(pipeline); - - ContainerProtos.GetBlockResponseProto response = ContainerProtocolCalls - .getBlock(xceiverClientSpi, blockID, token, pipeline.getReplicaIndexes()); - - chunks = response.getBlockData().getChunksList(); - } finally { - if (xceiverClientSpi != null) { - getXceiverClientFactory().releaseClientForReadData( - xceiverClientSpi, false); - } - } - return chunks; } } diff --git a/hadoop-ozone/client/src/test/java/org/apache/hadoop/ozone/client/checksum/FileChecksumBenchmark.java b/hadoop-ozone/client/src/test/java/org/apache/hadoop/ozone/client/checksum/FileChecksumBenchmark.java index 1f0ba7c5c782..d4ba3321a2f3 100644 --- a/hadoop-ozone/client/src/test/java/org/apache/hadoop/ozone/client/checksum/FileChecksumBenchmark.java +++ b/hadoop-ozone/client/src/test/java/org/apache/hadoop/ozone/client/checksum/FileChecksumBenchmark.java @@ -22,6 +22,7 @@ import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; +import static org.mockito.Mockito.withSettings; import java.io.IOException; import java.util.ArrayList; @@ -37,8 +38,11 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Supplier; import org.apache.hadoop.hdds.client.BlockID; import org.apache.hadoop.hdds.client.ECReplicationConfig; import org.apache.hadoop.hdds.protocol.DatanodeDetails; @@ -60,6 +64,7 @@ import org.apache.ratis.thirdparty.com.google.protobuf.ByteString; import org.junit.jupiter.api.Tag; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; /** * Benchmark for ECFileChecksumHelper that measures the actual checksum @@ -69,25 +74,30 @@ * Validates fixes to BaseFileChecksumHelper and ECFileChecksumHelper: * Fix 1: BaseFileChecksumHelper 7-arg constructor no longer chains to 6-arg * (redundant OM lookupKey RPC eliminated: 2 calls/file -> 1). - * Fix 2: ECFileChecksumHelper uses a deterministic pipeline ID so - * XceiverClientManager can cache and reuse gRPC connections - * (new connection per file -> one connection per placement group). - * Fix 3: Pipeline.newBuilder() instead of toBuilder() avoids an unnecessary - * SecureRandom.nextBytes call on every file. + * Measured by runBenchmark. + * Fix 2+3 combined: ECFileChecksumHelper caches the STANDALONE checksum + * pipeline per EC placement group (Fix 3 intra-file reuse) and + * derives its ID deterministically from the sorted node UUIDs + * (Fix 2 cross-file reuse). Measured by runBlockCacheBenchmark. * * Covers RS-3-2 (5 nodes, 3 data + 2 parity) and RS-6-3 (9 nodes, 6 data + 3 parity). * For RS-3-2 the standalone pipeline has 3 selected nodes (index=1 + parity {4,5}), * stripe checksum = 12 bytes. For RS-6-3: 4 selected nodes (index=1 + parity {7,8,9}), * stripe checksum = 16 bytes. * - * Each latency bucket runs a 10-second warmup followed by a 20-second - * measurement window. Throughput and per-file RPC counts are reported from - * the measurement window only. + * Each latency bucket runs a warmup followed by a 20-second measurement window. + * Throughput and per-file RPC counts are reported from the measurement window only. * * Run with: * mvn test -pl hadoop-ozone/client \ * -Dtest=FileChecksumBenchmark#runBenchmark \ - * -Dsurefire.failIfNoSpecifiedTests=false + * -Dsurefire.failIfNoSpecifiedTests=false \ + * -Dsurefire.fork.timeout=3600 + * + * mvn test -pl hadoop-ozone/client \ + * -Dtest=FileChecksumBenchmark#runBlockCacheBenchmark \ + * -Dsurefire.failIfNoSpecifiedTests=false \ + * -Dsurefire.fork.timeout=3600 * * To profile with async-profiler, pass the agent via surefire argLine, e.g.: * mvn test -pl hadoop-ozone/client \ @@ -103,6 +113,7 @@ public class FileChecksumBenchmark { private static final int WARMUP_SECS = 10; private static final int MEASURE_SECS = 20; private static final int[] LATENCIES_MS = {0, 5, 10}; + private static final int[] BLOCKS_PER_FILE_VARIANTS = {1, 3, 5, 10}; private static final long CONTAINER_ID = 1001L; private static final long FILE_SIZE = 5000L; // 136 = 4 x 34: every 4th file (idx % 4 == 0) gets a freshly built OmKeyInfo @@ -129,16 +140,24 @@ public class FileChecksumBenchmark { private static final ContainerProtos.ContainerCommandResponseProto EC63_GET_BLOCK_RESPONSE = buildGetBlockResponse(16); + // 15-node cluster pool for runBlockCacheBenchmark. Both scenarios draw from this pool. + // RS-3-2 draws 5 from 15 → C(15,5)=3,003 distinct placement groups. + // RS-6-3 draws 9 from 15 → C(15,9)=5,005 distinct placement groups. + // Fix 2 ensures the same node subset maps to the same standalone pipeline ID, + // so the XceiverClient pool is fully warmed after warmup (≈ 0 new connections/file). + private static final List EC15_NODES = buildEcNodes(15); + // --------------------------------------------------------------------------- // Instrumented XceiverClientFactory // --------------------------------------------------------------------------- static class CountingXceiverClientFactory implements XceiverClientFactory { - // Stable pipeline IDs = KEY_POOL - KEY_POOL/UNSTABLE_STEP = 102. - // Cap the pool just above that so stable files always hit the cache while - // unstable files (random pipeline IDs) are never stored and are GC'd immediately. - private static final int MAX_POOL_SIZE = KEY_POOL - KEY_POOL / UNSTABLE_STEP + 10; + // Both scenarios use the 15-node pool, so the pool converges to at most C(15,9)=5,005 + // entries (all placement groups covered during warmup). 2M is a safe upper bound. + private static final int MAX_POOL_SIZE = 2_000_000; private final ContainerProtos.ContainerCommandResponseProto blockResponse; + private final int dnConnectionLatencyMs; + private final int maxPoolSize; private final AtomicLong newConnectionCount = new AtomicLong(); private final AtomicLong reuseCount = new AtomicLong(); // clientPool is intentionally NOT cleared between warmup and measurement: @@ -149,7 +168,14 @@ static class CountingXceiverClientFactory implements XceiverClientFactory { CountingXceiverClientFactory( ContainerProtos.ContainerCommandResponseProto blockResponse) { + this(blockResponse, 0, MAX_POOL_SIZE); + } + + CountingXceiverClientFactory(ContainerProtos.ContainerCommandResponseProto blockResponse, + int dnConnectionLatencyMs, int maxPoolSize) { this.blockResponse = blockResponse; + this.dnConnectionLatencyMs = dnConnectionLatencyMs; + this.maxPoolSize = maxPoolSize; } void resetCounters() { @@ -174,8 +200,16 @@ public XceiverClientSpi acquireClientForReadData(Pipeline pipeline) reuseCount.incrementAndGet(); return existing; } + if (dnConnectionLatencyMs > 0) { + try { + Thread.sleep(dnConnectionLatencyMs); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + throw new IOException(ie); + } + } XceiverClientSpi newClient = createMockDnClient(pipeline, blockResponse); - if (clientPool.size() < MAX_POOL_SIZE) { + if (clientPool.size() < maxPoolSize) { XceiverClientSpi winner = clientPool.putIfAbsent(key, newClient); if (winner != null) { reuseCount.incrementAndGet(); @@ -285,7 +319,7 @@ private static BenchmarkResult measure(int latencyMs, ECReplicationConfig repCon new CountingXceiverClientFactory(blockResponse); AtomicLong fileIdx = new AtomicLong(); - OzoneManagerProtocol mockOm = mock(OzoneManagerProtocol.class); + OzoneManagerProtocol mockOm = mock(OzoneManagerProtocol.class, withSettings().stubOnly()); when(mockOm.lookupKey(any(OmKeyArgs.class))).thenAnswer(invocation -> { omCallCount.incrementAndGet(); if (latencyMs > 0) { @@ -307,13 +341,13 @@ private static BenchmarkResult measure(int latencyMs, ECReplicationConfig repCon return keyPool[idx]; }); - RpcClient mockRpcClient = mock(RpcClient.class); + RpcClient mockRpcClient = mock(RpcClient.class, withSettings().stubOnly()); when(mockRpcClient.getOzoneManagerClient()).thenReturn(mockOm); when(mockRpcClient.getXceiverClientManager()).thenReturn(xceiverFactory); - OzoneVolume mockVolume = mock(OzoneVolume.class); + OzoneVolume mockVolume = mock(OzoneVolume.class, withSettings().stubOnly()); when(mockVolume.getName()).thenReturn("vol"); - OzoneBucket mockBucket = mock(OzoneBucket.class); + OzoneBucket mockBucket = mock(OzoneBucket.class, withSettings().stubOnly()); when(mockBucket.getName()).thenReturn("bucket"); OzoneClientConfig.ChecksumCombineMode combineMode = @@ -356,6 +390,69 @@ private static BenchmarkResult measure(int latencyMs, ECReplicationConfig repCon latencyMs); } + /** + * @param keyInfoSupplier called once per lookupKey invocation to produce a fresh OmKeyInfo; + * must be thread-safe (called concurrently by {@link #NUM_THREADS}). + */ + private static BenchmarkResult measureBlockCache(int blocksPerFile, int dnLatencyMs, + Supplier keyInfoSupplier, + ContainerProtos.ContainerCommandResponseProto blockResponse) + throws Exception { + AtomicLong omCallCount = new AtomicLong(); + CountingXceiverClientFactory xceiverFactory = new CountingXceiverClientFactory( + blockResponse, dnLatencyMs, CountingXceiverClientFactory.MAX_POOL_SIZE); + + long fileLength = (long) blocksPerFile * FILE_SIZE; + + OzoneManagerProtocol mockOm = mock(OzoneManagerProtocol.class, withSettings().stubOnly()); + when(mockOm.lookupKey(any(OmKeyArgs.class))).thenAnswer(invocation -> { + omCallCount.incrementAndGet(); + return keyInfoSupplier.get(); + }); + + RpcClient mockRpcClient = mock(RpcClient.class, withSettings().stubOnly()); + when(mockRpcClient.getOzoneManagerClient()).thenReturn(mockOm); + when(mockRpcClient.getXceiverClientManager()).thenReturn(xceiverFactory); + + OzoneVolume mockVolume = mock(OzoneVolume.class, withSettings().stubOnly()); + when(mockVolume.getName()).thenReturn("vol"); + OzoneBucket mockBucket = mock(OzoneBucket.class, withSettings().stubOnly()); + when(mockBucket.getName()).thenReturn("bucket"); + + OzoneClientConfig.ChecksumCombineMode combineMode = + OzoneClientConfig.ChecksumCombineMode.COMPOSITE_CRC; + + Runnable task = () -> { + try { + OmKeyInfo keyInfo = mockRpcClient.getOzoneManagerClient().lookupKey( + new OmKeyArgs.Builder() + .setVolumeName("vol") + .setBucketName("bucket") + .setKeyName("file") + .setSortDatanodesInPipeline(true) + .setLatestVersionLocation(true) + .build()); + new ECFileChecksumHelper( + mockVolume, mockBucket, keyInfo.getKeyName(), fileLength, combineMode, + mockRpcClient, keyInfo) + .compute(); + } catch (IOException e) { + throw new RuntimeException(e); + } + }; + + runFor(WARMUP_SECS * 1000L, task); + omCallCount.set(0); + xceiverFactory.resetCounters(); + + long start = System.currentTimeMillis(); + long measuredFiles = runFor(MEASURE_SECS * 1000L, task); + long wallMs = System.currentTimeMillis() - start; + + return new BenchmarkResult(measuredFiles, wallMs, omCallCount.get(), + xceiverFactory.getNewConnectionCount(), xceiverFactory.getReuseCount(), dnLatencyMs); + } + private static long runFor(long durationMs, Runnable task) throws Exception { AtomicBoolean running = new AtomicBoolean(true); AtomicLong count = new AtomicLong(); @@ -413,6 +510,68 @@ public void runBenchmark() throws Exception { } + /** + * Measures multi-block EC file checksum collection through the real + * ECFileChecksumHelper.compute() path. All N blocks of each file share one EC + * placement group (the case the checksumPipelineCache is designed for) and the + * files are drawn from a 15-node cluster with random placement, so the + * deterministic standalone pipeline ID keeps the XceiverClient pool warm + * (NewConn/file ≈ 0.00) regardless of the fix under test. + * + * This is a single-arm benchmark: run it as-is for the "after" numbers, then + * run it again with the pre-Fix-3 ECFileChecksumHelper (which rebuilds the + * standalone pipeline for every block instead of caching it per placement + * group) for the "before" numbers. The throughput delta between the two runs + * is the real effect of Fix 3. + */ + @Test + @Timeout(value = 2400, unit = TimeUnit.SECONDS) + public void runBlockCacheBenchmark() throws Exception { + System.out.println(); + System.out.println("=== ECFileChecksum Multi-Block Checksum Benchmark ==="); + System.out.printf("Workload: %d threads, %ds warmup + %ds measurement%n%n", + NUM_THREADS, WARMUP_SECS, MEASURE_SECS); + + String[] ecLabels = {"RS-3-2", "RS-6-3"}; + ECReplicationConfig[] ecConfigs = {EC32, EC63}; + ContainerProtos.ContainerCommandResponseProto[] blockResponses = + {GET_BLOCK_RESPONSE, EC63_GET_BLOCK_RESPONSE}; + + String rowFmt = "%-6d %-10d %-10.1f %-13.2f %-11d %-9d %-9s"; + String hdr = String.format("%-6s %-10s %-10s %-13s %-11s %-9s %-9s", + "Blk/f", "Wall(ms)", "Files/s", "NewConn/file", "XcNew", "XcReuse", "CacheHit%"); + String rule = new String(new char[hdr.length()]).replace('\0', '-'); + + for (int ecIdx = 0; ecIdx < ecLabels.length; ecIdx++) { + System.out.printf("=== %s ===%n", ecLabels[ecIdx]); + ECReplicationConfig ecConfig = ecConfigs[ecIdx]; + ContainerProtos.ContainerCommandResponseProto blockResponse = blockResponses[ecIdx]; + + for (int dnLatencyMs : LATENCIES_MS) { + System.out.printf("DN latency = %dms%n", dnLatencyMs); + System.out.println(" " + hdr); + System.out.println(" " + rule); + for (int blocksPerFile : BLOCKS_PER_FILE_VARIANTS) { + int bpf = blocksPerFile; + BenchmarkResult r = measureBlockCache(blocksPerFile, dnLatencyMs, + () -> buildSharedClusterKeyInfoMultiBlock("file", EC15_NODES, ecConfig, bpf), + blockResponse); + printBlockCacheRow(rowFmt, blocksPerFile, r); + } + System.out.println(); + } + } + } + + private static void printBlockCacheRow(String fmt, int blocksPerFile, BenchmarkResult r) { + double newConnPerFile = r.getMeasuredFiles() == 0 + ? 0 : (double) r.getXceiverNewCount() / r.getMeasuredFiles(); + System.out.printf(" " + fmt + "%n", + blocksPerFile, r.getWallMs(), r.filesPerSec(), newConnPerFile, + r.getXceiverNewCount(), r.getXceiverReuseCount(), + r.cacheHitPercent() + "%"); + } + private static void printRow(BenchmarkResult r) { System.out.printf( "%-10s %-10d %-10d %-10.1f %-12d %-10.2f %-14d %-14d %d%%%n", @@ -435,7 +594,7 @@ private static List buildEcNodes(int count) { List nodes = new ArrayList<>(); for (int i = 0; i < count; i++) { nodes.add(DatanodeDetails.newBuilder() - .setUuid(UUID.fromString("00000000-0000-0000-0000-00000000000" + i)) + .setUuid(UUID.fromString(String.format("00000000-0000-0000-0000-%012d", i))) .setHostName("dn" + i) .setIpAddress("10.0.0." + i) .build()); @@ -531,13 +690,24 @@ private static ContainerProtos.ContainerCommandResponseProto buildGetBlockRespon } /** - * Builds an OmKeyInfo whose EC pipeline has freshly generated random node - * UUIDs. Called on every lookupKey invocation for unstable files so the - * deterministic pipeline ID computed by Fix 2 is also effectively random per - * call, guaranteeing a cache miss. + * Builds an OmKeyInfo whose EC pipeline has freshly generated random node UUIDs. + * Used by unstable files in {@link #measure} and by Scenario A in + * {@link #runBlockCacheBenchmark} so the deterministic pipeline ID computed + * by Fix 2 is also effectively random per call, guaranteeing a cache miss. */ private static OmKeyInfo buildRandomKeyInfo(String keyName, ECReplicationConfig repConfig) { + return buildRandomKeyInfoMultiBlock(keyName, repConfig, 1); + } + + /** + * Like {@link #buildRandomKeyInfo} but produces a file with {@code nBlocks} blocks, + * all backed by the same freshly-randomized EC pipeline. All N blocks share the same + * (unique-per-call) EC pipeline ID, so Fix 3's cache fires for blocks 2..N while + * Fix 2 still produces a unique standalone pipeline ID per file (random nodes). + */ + private static OmKeyInfo buildRandomKeyInfoMultiBlock(String keyName, + ECReplicationConfig repConfig, int nBlocks) { int nodeCount = repConfig.getData() + repConfig.getParity(); List nodes = new ArrayList<>(); Map replicaIndexes = new HashMap<>(); @@ -557,30 +727,94 @@ private static OmKeyInfo buildRandomKeyInfo(String keyName, .setNodes(nodes) .setReplicaIndexes(replicaIndexes) .build(); - OmKeyLocationInfo loc = new OmKeyLocationInfo.Builder() - .setBlockID(new BlockID(CONTAINER_ID, 0)) - .setPipeline(pipeline) - .setLength(FILE_SIZE) - .build(); + List locs = new ArrayList<>(); + for (int b = 0; b < nBlocks; b++) { + locs.add(new OmKeyLocationInfo.Builder() + .setBlockID(new BlockID(CONTAINER_ID + b, 0)) + .setPipeline(pipeline) + .setLength(FILE_SIZE) + .build()); + } return new OmKeyInfo.Builder() .setVolumeName("vol") .setBucketName("bucket") .setKeyName(keyName) .setOmKeyLocationInfos(Collections.singletonList( - new OmKeyLocationInfoGroup(0, - Collections.singletonList(loc)))) + new OmKeyLocationInfoGroup(0, locs))) .setCreationTime(0L) .setModificationTime(0L) - .setDataSize(FILE_SIZE) + .setDataSize((long) nBlocks * FILE_SIZE) .setReplicationConfig(repConfig) .setFileChecksum(null) .setAcls(Collections.emptyList()) .build(); } + /** + * Builds an OmKeyInfo whose EC pipeline has nodes randomly selected from {@code clusterNodes}. + * All N blocks of the file share one EC pipeline (same placement group) — the realistic case + * that ECFileChecksumHelper's checksumPipelineCache is designed to exploit. + */ + private static OmKeyInfo buildSharedClusterKeyInfoMultiBlock(String keyName, + List clusterNodes, ECReplicationConfig ecConfig, int nBlocks) { + int nodeCount = ecConfig.getData() + ecConfig.getParity(); + int clusterSize = clusterNodes.size(); + // O(nodeCount) selection without replacement using a small stack-allocated array + // to track already-picked indices. For nodeCount <= 9 the inner scan is O(1). + int[] picked = new int[nodeCount]; + List selected = new ArrayList<>(nodeCount); + while (selected.size() < nodeCount) { + int idx = ThreadLocalRandom.current().nextInt(clusterSize); + boolean duplicate = false; + for (int p = 0; p < selected.size(); p++) { + if (picked[p] == idx) { + duplicate = true; + break; + } + } + if (!duplicate) { + picked[selected.size()] = idx; + selected.add(clusterNodes.get(idx)); + } + } + Map replicaIndexes = new HashMap<>(); + for (int i = 0; i < selected.size(); i++) { + replicaIndexes.put(selected.get(i), i + 1); + } + Pipeline sharedPipeline = Pipeline.newBuilder() + .setId(PipelineID.randomId()) + .setReplicationConfig(ecConfig) + .setState(Pipeline.PipelineState.CLOSED) + .setNodes(selected) + .setReplicaIndexes(replicaIndexes) + .build(); + List locs = new ArrayList<>(); + for (int b = 0; b < nBlocks; b++) { + locs.add(new OmKeyLocationInfo.Builder() + .setBlockID(new BlockID(CONTAINER_ID + b, 0)) + .setPipeline(sharedPipeline) + .setLength(FILE_SIZE) + .build()); + } + return new OmKeyInfo.Builder() + .setVolumeName("vol") + .setBucketName("bucket") + .setKeyName(keyName) + .setOmKeyLocationInfos(Collections.singletonList( + new OmKeyLocationInfoGroup(0, locs))) + .setCreationTime(0L) + .setModificationTime(0L) + .setDataSize((long) nBlocks * FILE_SIZE) + .setReplicationConfig(ecConfig) + .setFileChecksum(null) + .setAcls(Collections.emptyList()) + .build(); + } + private static XceiverClientSpi createMockDnClient(Pipeline standalonePipeline, ContainerProtos.ContainerCommandResponseProto response) throws IOException { - XceiverClientSpi mockDn = mock(XceiverClientSpi.class, CALLS_REAL_METHODS); + XceiverClientSpi mockDn = mock(XceiverClientSpi.class, + withSettings().stubOnly().defaultAnswer(CALLS_REAL_METHODS)); XceiverClientReply reply = new XceiverClientReply( CompletableFuture.completedFuture(response)); try { diff --git a/hadoop-ozone/client/src/test/java/org/apache/hadoop/ozone/client/checksum/TestFileChecksumHelper.java b/hadoop-ozone/client/src/test/java/org/apache/hadoop/ozone/client/checksum/TestFileChecksumHelper.java index 42243ae2d386..f894e285f3e9 100644 --- a/hadoop-ozone/client/src/test/java/org/apache/hadoop/ozone/client/checksum/TestFileChecksumHelper.java +++ b/hadoop-ozone/client/src/test/java/org/apache/hadoop/ozone/client/checksum/TestFileChecksumHelper.java @@ -22,19 +22,29 @@ import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChecksumType.CRC32; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNotSame; import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.Mockito.any; +import static org.mockito.Mockito.anyBoolean; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import jakarta.annotation.Nonnull; import java.io.IOException; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.UUID; import java.util.concurrent.CompletableFuture; +import java.util.stream.Collectors; import org.apache.hadoop.fs.FileChecksum; import org.apache.hadoop.fs.MD5MD5CRC32FileChecksum; import org.apache.hadoop.fs.MD5MD5CRC32GzipFileChecksum; @@ -43,6 +53,7 @@ import org.apache.hadoop.hdds.client.RatisReplicationConfig; import org.apache.hadoop.hdds.client.ReplicationConfig; import org.apache.hadoop.hdds.client.ReplicationType; +import org.apache.hadoop.hdds.client.StandaloneReplicationConfig; import org.apache.hadoop.hdds.conf.InMemoryConfigurationForTesting; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.protocol.DatanodeDetails; @@ -76,11 +87,17 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.EnumSource; +import org.mockito.ArgumentCaptor; /** * Unit tests for Replicated and EC FileChecksumHelper class. */ public class TestFileChecksumHelper { + // RS-6-3: replica indexes 1..9. The checksum pipeline keeps index 1 plus the + // three parity nodes (indexes 7, 8, 9) and drops data indexes 2..6. + private static final int EC_DATA = 6; + private static final int EC_PARITY = 3; + private final FileChecksum noCachedChecksum = null; private OzoneClient client; private ObjectStore store; @@ -163,6 +180,152 @@ private Pipeline pipeline(ReplicationType type, List datanodeDe .build(); } + private List ecNodes() { + List nodes = new ArrayList<>(); + for (int i = 0; i < EC_DATA + EC_PARITY; i++) { + nodes.add(DatanodeDetails.newBuilder().setUuid(UUID.randomUUID()).build()); + } + return nodes; + } + + private Pipeline ecPipeline(List nodes, PipelineID id) { + Map replicaIndexes = new HashMap<>(); + for (int i = 0; i < nodes.size(); i++) { + replicaIndexes.put(nodes.get(i), i + 1); + } + return Pipeline.newBuilder() + .setId(id) + .setReplicationConfig(new ECReplicationConfig(EC_DATA, EC_PARITY)) + .setState(Pipeline.PipelineState.CLOSED) + .setNodes(nodes) + .setReplicaIndexes(replicaIndexes) + .build(); + } + + private OmKeyLocationInfo ecBlock(long localId, Pipeline pipeline) { + return new OmKeyLocationInfo.Builder() + .setPipeline(pipeline) + .setBlockID(new BlockID(1, localId)) + .setLength(100) + .build(); + } + + private ECFileChecksumHelper ecHelper(List blocks, + XceiverClientFactory factory) throws IOException { + RpcClient mockRpcClient = mock(RpcClient.class); + when(mockRpcClient.getXceiverClientManager()).thenReturn(factory); + OzoneVolume mockVolume = mock(OzoneVolume.class); + when(mockVolume.getName()).thenReturn("vol1"); + OzoneBucket mockBucket = mock(OzoneBucket.class); + when(mockBucket.getName()).thenReturn("bucket1"); + OmKeyInfo keyInfo = omKeyInfo(ReplicationType.EC, noCachedChecksum, blocks); + return new ECFileChecksumHelper(mockVolume, mockBucket, "dummy", + 100L * blocks.size(), OzoneClientConfig.ChecksumCombineMode.MD5MD5CRC, + mockRpcClient, keyInfo); + } + + private XceiverClientFactory ecReadFactory(ArgumentCaptor captor) throws IOException { + OzoneConfiguration conf = new OzoneConfiguration(); + XceiverClientGrpc grpc = new XceiverClientGrpc(pipeline(ReplicationType.EC, ecNodes()), conf) { + @Override + public XceiverClientReply sendCommandAsync( + ContainerProtos.ContainerCommandRequestProto request, DatanodeDetails dn) { + return buildValidResponse(ReplicationType.EC); + } + }; + XceiverClientFactory factory = mock(XceiverClientFactory.class); + when(factory.acquireClientForReadData(captor.capture())).thenReturn(grpc); + return factory; + } + + /** + * Blocks that share the same EC pipeline ID must reuse a single cached + * STANDALONE checksum pipeline, which is filtered down to replica index 1 + * plus the parity nodes and given a deterministic ID from the sorted node + * UUIDs. Exercises the cache-hit path and the node-filter branches. + */ + @Test + public void testEcChecksumPipelineCachedAndFilteredAcrossBlocks() throws IOException { + List nodes = ecNodes(); + Pipeline sharedPipeline = ecPipeline(nodes, PipelineID.randomId()); + List blocks = Arrays.asList( + ecBlock(1, sharedPipeline), ecBlock(2, sharedPipeline)); + + ArgumentCaptor captor = ArgumentCaptor.forClass(Pipeline.class); + XceiverClientFactory factory = ecReadFactory(captor); + ECFileChecksumHelper helper = ecHelper(blocks, factory); + + helper.compute(); + + assertInstanceOf(MD5MD5CRC32GzipFileChecksum.class, helper.getFileChecksum()); + + List acquired = captor.getAllValues(); + assertEquals(2, acquired.size()); + // Both blocks acquired the identical cached instance (cache hit on block 2). + assertSame(acquired.get(0), acquired.get(1)); + + Pipeline checksumPipeline = acquired.get(0); + assertInstanceOf(StandaloneReplicationConfig.class, checksumPipeline.getReplicationConfig()); + // index 1 + parity {7, 8, 9}; data indexes 2..6 are filtered out. + assertEquals(1 + EC_PARITY, checksumPipeline.getNodes().size()); + for (DatanodeDetails dn : checksumPipeline.getNodes()) { + int ri = checksumPipeline.getReplicaIndex(dn); + assertTrue(ri == 1 || ri > EC_DATA, "unexpected replica index " + ri); + } + String nodeKey = checksumPipeline.getNodes().stream() + .map(DatanodeDetails::getUuidString) + .sorted() + .collect(Collectors.joining(",")); + assertEquals(PipelineID.valueOf(UUID.nameUUIDFromBytes(nodeKey.getBytes(UTF_8))), + checksumPipeline.getId()); + } + + /** + * Blocks carrying distinct EC pipeline IDs miss the cache and rebuild the + * checksum pipeline, but the deterministic ID keeps them equal so the + * XceiverClient pool can still be reused. Exercises the cache-miss path. + */ + @Test + public void testEcChecksumPipelineRebuiltForDistinctPipelineIds() throws IOException { + List nodes = ecNodes(); + List blocks = Arrays.asList( + ecBlock(1, ecPipeline(nodes, PipelineID.randomId())), + ecBlock(2, ecPipeline(nodes, PipelineID.randomId()))); + + ArgumentCaptor captor = ArgumentCaptor.forClass(Pipeline.class); + ECFileChecksumHelper helper = ecHelper(blocks, ecReadFactory(captor)); + + helper.compute(); + + assertInstanceOf(MD5MD5CRC32GzipFileChecksum.class, helper.getFileChecksum()); + List acquired = captor.getAllValues(); + assertEquals(2, acquired.size()); + // Distinct EC IDs miss the cache, so each block builds its own instance ... + assertNotSame(acquired.get(0), acquired.get(1)); + // ... yet the deterministic ID (same nodes) is identical for pool reuse. + assertEquals(acquired.get(0).getId(), acquired.get(1).getId()); + } + + /** + * When acquiring the read client fails, getChunkInfos must not attempt to + * release a client. Exercises the {@code xceiverClientSpi == null} branch of + * the finally block. + */ + @Test + public void testEcChecksumAcquireFailureSkipsRelease() throws IOException { + List nodes = ecNodes(); + List blocks = Collections.singletonList( + ecBlock(1, ecPipeline(nodes, PipelineID.randomId()))); + + XceiverClientFactory factory = mock(XceiverClientFactory.class); + when(factory.acquireClientForReadData(any())) + .thenThrow(new IOException("acquire failed")); + ECFileChecksumHelper helper = ecHelper(blocks, factory); + + assertThrows(IOException.class, helper::compute); + verify(factory, never()).releaseClientForReadData(any(), anyBoolean()); + } + @ParameterizedTest @EnumSource(names = {"EC", "RATIS"}) public void testEmptyBlock(ReplicationType helperType) throws IOException {