diff --git a/common/network-common/src/main/java/org/apache/spark/network/util/TransportConf.java b/common/network-common/src/main/java/org/apache/spark/network/util/TransportConf.java index a37979bf30acf..29b7964885100 100644 --- a/common/network-common/src/main/java/org/apache/spark/network/util/TransportConf.java +++ b/common/network-common/src/main/java/org/apache/spark/network/util/TransportConf.java @@ -613,4 +613,24 @@ public long mergedShuffleCleanerShutdownTimeout() { return JavaUtils.timeStringAsSec( conf.get("spark.shuffle.push.server.mergedShuffleCleaner.shutdown.timeout", "60s")); } + + /** + * Whether the shuffle server calculates a checksum for every chunk of a merged shuffle + * partition while merging pushed blocks. The checksums are stored alongside the merged shuffle + * data and are only used to diagnose the cause of a corrupted shuffle chunk. + */ + public boolean mergedShuffleChecksumEnabled() { + return conf.getBoolean("spark.shuffle.push.server.mergedShuffleChecksum.enabled", true); + } + + /** + * The algorithm used to calculate the checksums of the merged shuffle chunks. The reducer + * calculates the checksum of a corrupted chunk with spark.shuffle.checksum.algorithm, so + * corruption of a merged shuffle chunk can only be diagnosed when the two match. + */ + public String mergedShuffleChecksumAlgorithm() { + // Upper cased like the spark.shuffle.checksum.algorithm of the application is + return conf.get("spark.shuffle.push.server.mergedShuffleChecksum.algorithm", "ADLER32") + .toUpperCase(Locale.ROOT); + } } diff --git a/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/BlockStoreClient.java b/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/BlockStoreClient.java index ceb5d64699744..ba7b86cd1cf7a 100644 --- a/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/BlockStoreClient.java +++ b/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/BlockStoreClient.java @@ -90,6 +90,48 @@ public Cause diagnoseCorruption( } } + /** + * Send the diagnosis request for the corrupted chunk of a merged shuffle partition to the + * shuffle server which merged it. + * + * @param host the host of the shuffle server which merged the chunk. + * @param port the port of the shuffle server which merged the chunk. + * @param shuffleId the shuffleId of the corrupted shuffle chunk + * @param shuffleMergeId the shuffleMergeId of the corrupted shuffle chunk + * @param reduceId the reduceId of the corrupted shuffle chunk + * @param chunkId the chunkId of the corrupted shuffle chunk + * @param checksum the shuffle checksum which calculated at client side for the corrupted + * shuffle chunk + * @param algorithm the checksum algorithm which is used for calculating checksum + * @return The cause of the shuffle chunk corruption + */ + public Cause diagnoseShuffleChunkCorruption( + String host, + int port, + int shuffleId, + int shuffleMergeId, + int reduceId, + int chunkId, + long checksum, + String algorithm) { + try { + TransportClient client = clientFactory.createClient(host, port); + ByteBuffer response = client.sendRpcSync( + new DiagnoseShuffleChunkCorruption( + appId, shuffleId, shuffleMergeId, reduceId, chunkId, checksum, algorithm).toByteBuffer(), + transportConf.connectionTimeoutMs() + ); + CorruptionCause cause = + (CorruptionCause) BlockTransferMessage.Decoder.fromByteBuffer(response); + return cause.cause; + } catch (Exception e) { + // A shuffle service that does not support this request yet answers it with an error, so + // the exception is logged to tell that case apart from a genuinely unknown cause. + logger.warn("Failed to get the corruption cause of the shuffle chunk.", e); + return Cause.UNKNOWN_ISSUE; + } + } + /** * Fetch a sequence of blocks from a remote node asynchronously, * diff --git a/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalBlockHandler.java b/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalBlockHandler.java index a0551254c9e8c..69f84518862b7 100644 --- a/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalBlockHandler.java +++ b/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalBlockHandler.java @@ -280,6 +280,13 @@ protected void handleMessage( // In any cases of the error, diagnoseShuffleBlockCorruption should return UNKNOWN_ISSUE, // so it should always reply as success. callback.onSuccess(new CorruptionCause(cause).toByteBuffer()); + } else if (msgObj instanceof DiagnoseShuffleChunkCorruption msg) { + checkAuth(client, msg.appId); + Cause cause = mergeManager.diagnoseShuffleChunkCorruption(msg.appId, msg.shuffleId, + msg.shuffleMergeId, msg.reduceId, msg.chunkId, msg.checksum, msg.algorithm); + // In any cases of the error, diagnoseShuffleChunkCorruption should return UNKNOWN_ISSUE, + // so it should always reply as success. + callback.onSuccess(new CorruptionCause(cause).toByteBuffer()); } else { throw new UnsupportedOperationException("Unexpected message: " + msgObj); } diff --git a/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/MergedShuffleFileManager.java b/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/MergedShuffleFileManager.java index cd5bb507dbea5..9ffa59af00d9c 100644 --- a/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/MergedShuffleFileManager.java +++ b/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/MergedShuffleFileManager.java @@ -25,6 +25,7 @@ import org.apache.spark.annotation.Evolving; import org.apache.spark.network.buffer.ManagedBuffer; import org.apache.spark.network.client.StreamCallbackWithID; +import org.apache.spark.network.shuffle.checksum.Cause; import org.apache.spark.network.shuffle.protocol.ExecutorShuffleInfo; import org.apache.spark.network.shuffle.protocol.FinalizeShuffleMerge; import org.apache.spark.network.shuffle.protocol.MergeStatuses; @@ -133,6 +134,34 @@ MergedBlockMeta getMergedBlockMeta( */ void removeShuffleMerge(RemoveShuffleMerge removeShuffleMerge); + /** + * Diagnose the cause of the corruption of a merged shuffle chunk by comparing the checksum + * calculated by the reducer against the one calculated while the chunk was merged. This is + * best effort, so it returns {@link Cause#UNKNOWN_ISSUE} rather than failing when the + * checksum of the chunk is unavailable. + * + * @param appId application ID + * @param shuffleId shuffle ID + * @param shuffleMergeId shuffleMergeId is used to uniquely identify merging process + * of shuffle by an indeterminate stage attempt. + * @param reduceId reducer ID + * @param chunkId the ID of the corrupted chunk of the merged shuffle partition + * @param checksumByReader the checksum of the chunk calculated by the reducer + * @param algorithm the checksum algorithm the reducer used + * @return the cause of the corruption + * @since 4.4.0 + */ + default Cause diagnoseShuffleChunkCorruption( + String appId, + int shuffleId, + int shuffleMergeId, + int reduceId, + int chunkId, + long checksumByReader, + String algorithm) { + return Cause.UNKNOWN_ISSUE; + } + /** * Optionally close any resources associated the MergedShuffleFileManager, such as the * leveldb for state persistence. diff --git a/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java b/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java index 43f5e9c530e53..3dcd3787a6b41 100644 --- a/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java +++ b/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java @@ -28,6 +28,7 @@ import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; +import java.nio.file.StandardOpenOption; import java.util.ArrayList; import java.util.Arrays; import java.util.HashMap; @@ -42,6 +43,7 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; +import java.util.zip.Checksum; import com.fasterxml.jackson.annotation.JsonCreator; @@ -73,6 +75,8 @@ import org.apache.spark.network.client.StreamCallbackWithID; import org.apache.spark.network.server.BlockPushNonFatalFailure; import org.apache.spark.network.server.BlockPushNonFatalFailure.ReturnCode; +import org.apache.spark.network.shuffle.checksum.Cause; +import org.apache.spark.network.shuffle.checksum.ShuffleChecksumHelper; import org.apache.spark.network.shuffle.protocol.BlockPushReturnCode; import org.apache.spark.network.shuffle.protocol.ExecutorShuffleInfo; import org.apache.spark.network.shuffle.protocol.FinalizeShuffleMerge; @@ -106,6 +110,11 @@ public class RemoteBlockPushResolver implements MergedShuffleFileManager { public static final String ATTEMPT_ID_KEY = "attemptId"; private static final int UNDEFINED_ATTEMPT_ID = -1; + /** + * Marks the running checksum of a chunk as no longer describing a prefix of that chunk. + */ + private static final long INVALID_CHECKSUM_POS = -1L; + /** * The flag for deleting all merged shuffle data. */ @@ -149,6 +158,11 @@ public class RemoteBlockPushResolver implements MergedShuffleFileManager { private final int minChunkSize; private final int ioExceptionsThresholdDuringMerge; + // Whether a checksum is calculated for every chunk of a merged shuffle partition while the + // pushed blocks are merged. This is disabled when the configured algorithm is not supported. + private final boolean checksumEnabled; + private final String checksumAlgorithm; + @SuppressWarnings("UnstableApiUsage") private final LoadingCache indexCache; @@ -170,6 +184,9 @@ public RemoteBlockPushResolver(TransportConf conf, File recoveryFile) throws IOE this.cleanerShutdownTimeout = conf.mergedShuffleCleanerShutdownTimeout(); this.minChunkSize = conf.minChunkSizeInMergedShuffleFile(); this.ioExceptionsThresholdDuringMerge = conf.ioExceptionsThresholdDuringMerge(); + this.checksumAlgorithm = conf.mergedShuffleChecksumAlgorithm(); + this.checksumEnabled = conf.mergedShuffleChecksumEnabled() && + isSupportedChecksumAlgorithm(this.checksumAlgorithm); CacheLoader indexCacheLoader = new CacheLoader() { @Override @@ -196,6 +213,22 @@ public ShuffleIndexInformation load(String filePath) throws IOException { this.pushMergeMetrics = new PushMergeMetrics(); } + /** + * Merging the pushed blocks should not fail because of an unusable checksum algorithm, so an + * unsupported one only disables the checksum calculation. + */ + private static boolean isSupportedChecksumAlgorithm(String algorithm) { + try { + ShuffleChecksumHelper.getChecksumByAlgorithm(algorithm); + return true; + } catch (UnsupportedOperationException e) { + logger.warn("Checksums of the merged shuffle chunks are not calculated because {} is not " + + "a supported shuffle checksum algorithm", + MDC.of(LogKeys.CHECKSUM_ALGORITHM, algorithm)); + return false; + } + } + @VisibleForTesting protected static ErrorHandler.BlockPushErrorHandler createErrorHandler() { return new ErrorHandler.BlockPushErrorHandler() { @@ -319,9 +352,15 @@ AppShufflePartitionInfo newAppShufflePartitionInfo( File dataFile, File indexFile, File metaFile) throws IOException { + MergeShuffleFile checksumFile = null; + if (checksumEnabled) { + checksumFile = new MergeShuffleFile(appShuffleInfo.getMergedShuffleChecksumFile( + shuffleId, shuffleMergeId, reduceId, checksumAlgorithm)); + } return new AppShufflePartitionInfo(new AppAttemptShuffleMergeId( appShuffleInfo.appId, appShuffleInfo.attemptId, shuffleId, shuffleMergeId), - reduceId, dataFile, new MergeShuffleFile(indexFile), new MergeShuffleFile(metaFile)); + reduceId, dataFile, new MergeShuffleFile(indexFile), new MergeShuffleFile(metaFile), + checksumFile, checksumAlgorithm); } @Override @@ -396,6 +435,56 @@ public ManagedBuffer getMergedBlockData( } } + @Override + public Cause diagnoseShuffleChunkCorruption( + String appId, + int shuffleId, + int shuffleMergeId, + int reduceId, + int chunkId, + long checksumByReader, + String algorithm) { + if (!checksumEnabled) { + logger.warn("Cannot diagnose the corruption of shuffle chunk {} of shuffle {} " + + "shuffleMerge {} reduceId {} because the checksums of the merged shuffle chunks are " + + "not calculated", + MDC.of(LogKeys.CHUNK_ID, chunkId), + MDC.of(LogKeys.SHUFFLE_ID, shuffleId), + MDC.of(LogKeys.SHUFFLE_MERGE_ID, shuffleMergeId), + MDC.of(LogKeys.REDUCE_ID, reduceId)); + return Cause.UNKNOWN_ISSUE; + } + if (!checksumAlgorithm.equals(algorithm)) { + // Both algorithms may well be supported on their own, but the checksum of the chunk cannot + // be compared against the one the reducer calculated with a different algorithm. + logger.warn("Cannot diagnose the corruption of a shuffle chunk calculated with {} because " + + "the merged shuffle chunks are checksummed with {}. Set " + + "spark.shuffle.checksum.algorithm of the application and " + + "spark.shuffle.push.server.mergedShuffleChecksum.algorithm of this shuffle service to " + + "the same algorithm", + MDC.of(LogKeys.CHECKSUM_ALGORITHM, algorithm), + MDC.of(LogKeys.MERGED_SHUFFLE_CHECKSUM_ALGORITHM, checksumAlgorithm)); + return Cause.UNSUPPORTED_CHECKSUM_ALGORITHM; + } + try { + AppShuffleInfo appShuffleInfo = validateAndGetAppShuffleInfo(appId); + File checksumFile = appShuffleInfo.getMergedShuffleChecksumFile( + shuffleId, shuffleMergeId, reduceId, algorithm); + ManagedBuffer chunkData = + getMergedBlockData(appId, shuffleId, shuffleMergeId, reduceId, chunkId); + return ShuffleChecksumHelper.diagnoseCorruption( + algorithm, checksumFile, chunkId, chunkData, checksumByReader); + } catch (Exception e) { + logger.warn("Unable to diagnose the corruption of shuffle chunk {} of shuffle {} " + + "shuffleMerge {} reduceId {}", e, + MDC.of(LogKeys.CHUNK_ID, chunkId), + MDC.of(LogKeys.SHUFFLE_ID, shuffleId), + MDC.of(LogKeys.SHUFFLE_MERGE_ID, shuffleMergeId), + MDC.of(LogKeys.REDUCE_ID, reduceId)); + return Cause.UNKNOWN_ISSUE; + } + } + @Override public String[] getMergedBlockDirs(String appId) { AppShuffleInfo appShuffleInfo = validateAndGetAppShuffleInfo(appId); @@ -577,6 +666,7 @@ void deleteMergedFiles( int dataFilesDeleteCnt = 0; int indexFilesDeleteCnt = 0; int metaFilesDeleteCnt = 0; + int checksumFilesDeleteCnt = 0; for (int reduceId : reduceIds) { File dataFile = appShuffleInfo.getMergedShuffleDataFile(shuffleId, shuffleMergeId, reduceId); @@ -593,11 +683,19 @@ void deleteMergedFiles( if (metaFile.delete()) { metaFilesDeleteCnt++; } + if (checksumEnabled) { + File checksumFile = appShuffleInfo.getMergedShuffleChecksumFile( + shuffleId, shuffleMergeId, reduceId, checksumAlgorithm); + if (checksumFile.delete()) { + checksumFilesDeleteCnt++; + } + } } - logger.info("Delete {} data files, {} index files, {} meta files for {}", + logger.info("Delete {} data files, {} index files, {} meta files, {} checksum files for {}", MDC.of(LogKeys.NUM_DATA_FILES, dataFilesDeleteCnt), MDC.of(LogKeys.NUM_INDEX_FILES, indexFilesDeleteCnt), MDC.of(LogKeys.NUM_META_FILES, metaFilesDeleteCnt), + MDC.of(LogKeys.NUM_CHECKSUM_FILE, checksumFilesDeleteCnt), MDC.of(LogKeys.APP_ATTEMPT_SHUFFLE_MERGE_ID, appAttemptShuffleMergeId)); } @@ -1294,6 +1392,11 @@ public ByteBuffer getCompletionResponse() { * block. */ private void writeBuf(ByteBuffer buf) throws IOException { + // The running checksum is updated only once the whole buffer has been written. A buffer + // that is only partially written leaves the checksum behind the data that reached the file, + // which invalidates it at the next write instead of corrupting it. + long writePos = partitionInfo.getDataFilePos() + length; + ByteBuffer checksumBuf = partitionInfo.isChecksumEnabled() ? buf.duplicate() : null; while (buf.hasRemaining()) { long updatedPos = partitionInfo.getDataFilePos() + length; logger.debug("{} current pos {} updated pos {}", partitionInfo, @@ -1302,6 +1405,9 @@ private void writeBuf(ByteBuffer buf) throws IOException { length += bytesWritten; mergeManager.pushMergeMetrics.blockBytesWritten.mark(bytesWritten); } + if (checksumBuf != null) { + partitionInfo.updateChunkChecksum(writePos, checksumBuf); + } } /** @@ -1761,6 +1867,19 @@ public static class AppShufflePartitionInfo { // The meta file for a particular merged shuffle contains all the map indices that belong to // every chunk. The entry per chunk is a serialized bitmap. private final MergeShuffleFile metaFile; + // The checksum file for a particular merged shuffle contains the checksum of every chunk, + // one long per chunk. It is null when the checksum calculation is disabled. Unlike the index + // file, it has no leading entry, so it holds one entry less than the index file. + private final MergeShuffleFile checksumFile; + // The checksum of the chunk that is currently being merged, null when checksumFile is null. + private final Checksum chunkChecksum; + // The offset in the data file up to which chunkChecksum has consumed the merged data, or + // INVALID_CHECKSUM_POS when chunkChecksum no longer describes a prefix of the current chunk. + private long checksumPos; + // Whether the checksum file still holds the checksum of every chunk of this partition. It is + // unset when the checksums cannot be kept up to date, which only disables the diagnosis of a + // corrupted chunk of this partition and never fails the merge itself. + private boolean checksumUsable; private final Cleaner.Cleanable cleanable; // Location offset of the last successfully merged block for this shuffle partition private long dataFilePos; @@ -1781,7 +1900,9 @@ public static class AppShufflePartitionInfo { int reduceId, File dataFile, MergeShuffleFile indexFile, - MergeShuffleFile metaFile) throws IOException { + MergeShuffleFile metaFile, + MergeShuffleFile checksumFile, + String checksumAlgorithm) throws IOException { this.appAttemptShuffleMergeId = appAttemptShuffleMergeId; this.reduceId = reduceId; // Create FileOutputStream with append mode set to false by default. @@ -1792,6 +1913,11 @@ public static class AppShufflePartitionInfo { this.dataFile = dataFile; this.indexFile = indexFile; this.metaFile = metaFile; + this.checksumFile = checksumFile; + this.chunkChecksum = checksumFile == null ? null : + ShuffleChecksumHelper.getChecksumByAlgorithm(checksumAlgorithm); + this.checksumPos = 0; + this.checksumUsable = checksumFile != null; this.currentMapIndex = -1; // Writing 0 offset so that we can reuse ShuffleIndexInformation.getIndex() updateChunkInfo(0L, -1); @@ -1799,7 +1925,7 @@ public static class AppShufflePartitionInfo { this.mapTracker = new RoaringBitmap(); this.chunkTracker = new RoaringBitmap(); this.cleanable = CLEANER.register(this, new ResourceCleaner(dataChannel, indexFile, - metaFile, appAttemptShuffleMergeId, reduceId)); + metaFile, checksumFile, appAttemptShuffleMergeId, reduceId)); } public long getDataFilePos() { @@ -1836,12 +1962,15 @@ void resetChunkTracker() { } /** - * Appends the chunk offset to the index file and adds the map index to the chunk tracker. + * Appends the chunk offset to the index file, the chunk checksum to the checksum file and + * adds the map index to the chunk tracker. * * @param chunkOffset the offset of the chunk in the data file. * @param mapIndex the map index to be added to chunk tracker. */ void updateChunkInfo(long chunkOffset, int mapIndex) throws IOException { + boolean chunkSealed = chunkOffset > lastChunkOffset; + boolean checksumWritten = false; try { logger.trace("{} index current {} updated {}", this, this.lastChunkOffset, chunkOffset); @@ -1849,12 +1978,23 @@ void updateChunkInfo(long chunkOffset, int mapIndex) throws IOException { indexFile.getChannel().position(indexFile.getPos()); } indexFile.getDos().writeLong(chunkOffset); + // The checksum is written before the chunk bitmap for the same reason the bitmap is + // written after the offset: nothing that can throw may run once the meta file position + // has been advanced, otherwise a retry would append a second bitmap for this chunk. + if (chunkSealed) { + checksumWritten = writeChunkChecksum(chunkOffset); + } // Chunk bitmap should be written to the meta file after the index file because if there are // any exceptions during writing the offset to the index file, meta file should not be // updated. If the update to the index file is successful but the update to meta file isn't // then the index file position is not updated. writeChunkTracker(mapIndex); indexFile.updatePos(8); + if (checksumWritten) { + checksumFile.updatePos(8); + chunkChecksum.reset(); + checksumPos = chunkOffset; + } this.lastChunkOffset = chunkOffset; indexMetaUpdateFailed = false; } catch (IOException ioe) { @@ -1868,6 +2008,91 @@ void updateChunkInfo(long chunkOffset, int mapIndex) throws IOException { } } + /** + * Appends the checksum of the chunk that ends at chunkOffset to the checksum file. The + * running checksum is recomputed from the merged data file first if it does not describe + * exactly the data of this chunk, which happens when a block that was partially written is + * abandoned, or when a block is still being written while the shuffle merge is finalized. + * + * @return whether the checksum of the chunk was appended. The checksums are only used to + * diagnose a corrupted chunk, so a failure here gives up on the checksums of this + * partition rather than propagating to the merge of the block. + */ + private boolean writeChunkChecksum(long chunkOffset) { + if (!isChecksumEnabled()) { + return false; + } + try { + if (checksumPos != chunkOffset) { + recomputeChunkChecksum(chunkOffset); + } + if (indexMetaUpdateFailed) { + checksumFile.getChannel().position(checksumFile.getPos()); + } + checksumFile.getDos().writeLong(chunkChecksum.getValue()); + return true; + } catch (IOException ioe) { + logger.warn("{} reduceId {} failed to update the checksums of the merged chunks, the " + + "corruption of a chunk of this shuffle partition cannot be diagnosed", ioe, + MDC.of(LogKeys.APP_ATTEMPT_SHUFFLE_MERGE_ID, appAttemptShuffleMergeId), + MDC.of(LogKeys.REDUCE_ID, reduceId)); + checksumUsable = false; + return false; + } + } + + /** + * Recalculates the running checksum over the data of the current chunk, which is the range + * [lastChunkOffset, chunkOffset) of the merged shuffle data file. The data file is reopened + * for reading because dataChannel is write-only. + */ + private void recomputeChunkChecksum(long chunkOffset) throws IOException { + logger.debug("{} reduceId {} recomputing the checksum of the chunk [{}, {})", + appAttemptShuffleMergeId, reduceId, lastChunkOffset, chunkOffset); + chunkChecksum.reset(); + try (FileChannel readChannel = FileChannel.open(dataFile.toPath(), StandardOpenOption.READ)) { + ByteBuffer buffer = + ByteBuffer.allocate(ShuffleChecksumHelper.CHECKSUM_CALCULATION_BUFFER); + long pos = lastChunkOffset; + while (pos < chunkOffset) { + buffer.clear(); + buffer.limit((int) Math.min(buffer.capacity(), chunkOffset - pos)); + int bytesRead = readChannel.read(buffer, pos); + if (bytesRead <= 0) { + throw new IOException(String.format( + "Cannot read the merged shuffle data file %s at position %d", + dataFile.getAbsolutePath(), pos)); + } + pos += bytesRead; + buffer.flip(); + chunkChecksum.update(buffer); + } + } + checksumPos = chunkOffset; + } + + boolean isChecksumEnabled() { + return chunkChecksum != null && checksumUsable; + } + + /** + * Feeds the data of a block that has just been written to the merged shuffle data file to the + * running checksum of the current chunk. The running checksum is invalidated when the data is + * not written right where the checksum stopped, which happens when the partial data of an + * abandoned block is overwritten. It is then recomputed when the chunk gets sealed. + */ + void updateChunkChecksum(long writePos, ByteBuffer data) { + if (!isChecksumEnabled()) { + return; + } + if (checksumPos != writePos) { + checksumPos = INVALID_CHECKSUM_POS; + return; + } + checksumPos += data.remaining(); + chunkChecksum.update(data); + } + private void writeChunkTracker(int mapIndex) throws IOException { if (mapIndex == -1) { return; @@ -1907,6 +2132,29 @@ private void finalizePartition() throws IOException { dataChannel.truncate(lastChunkOffset); indexFile.getChannel().truncate(indexFile.getPos()); metaFile.getChannel().truncate(metaFile.getPos()); + finalizeChecksumFile(); + } + + /** + * Discards any checksum that does not describe a chunk of the finalized partition. The + * checksums are only used to diagnose a corrupted chunk, so the checksum file is deleted + * instead of failing the finalization of an otherwise correctly merged partition. + */ + private void finalizeChecksumFile() { + if (checksumFile == null) { + return; + } + if (checksumUsable) { + try { + checksumFile.getChannel().truncate(checksumFile.getPos()); + return; + } catch (IOException ioe) { + logger.warn("{} reduceId {} failed to truncate the checksum file", + MDC.of(LogKeys.APP_ATTEMPT_SHUFFLE_MERGE_ID, appAttemptShuffleMergeId), + MDC.of(LogKeys.REDUCE_ID, reduceId)); + } + } + checksumFile.delete(); } private void deleteAllFiles() { @@ -1917,6 +2165,9 @@ private void deleteAllFiles() { } metaFile.delete(); indexFile.delete(); + if (checksumFile != null) { + checksumFile.delete(); + } } @Override @@ -1937,6 +2188,11 @@ MergeShuffleFile getMetaFile() { return metaFile; } + @VisibleForTesting + MergeShuffleFile getChecksumFile() { + return checksumFile; + } + @VisibleForTesting FileChannel getDataChannel() { return dataChannel; @@ -1961,12 +2217,13 @@ private record ResourceCleaner( FileChannel dataChannel, MergeShuffleFile indexFile, MergeShuffleFile metaFile, + MergeShuffleFile checksumFile, AppAttemptShuffleMergeId appAttemptShuffleMergeId, int reduceId) implements Runnable { @Override public void run() { - closeAllFiles(dataChannel, indexFile, metaFile, appAttemptShuffleMergeId, + closeAllFiles(dataChannel, indexFile, metaFile, checksumFile, appAttemptShuffleMergeId, reduceId); } @@ -1974,6 +2231,7 @@ private void closeAllFiles( FileChannel dataChannel, MergeShuffleFile indexFile, MergeShuffleFile metaFile, + MergeShuffleFile checksumFile, AppAttemptShuffleMergeId appAttemptShuffleMergeId, int reduceId) { try { @@ -1999,6 +2257,15 @@ private void closeAllFiles( MDC.of(LogKeys.APP_ATTEMPT_SHUFFLE_MERGE_ID, appAttemptShuffleMergeId), MDC.of(LogKeys.REDUCE_ID, reduceId)); } + try { + if (checksumFile != null) { + checksumFile.close(); + } + } catch (IOException ioe) { + logger.warn("Error closing checksum file for {} reduceId {}", + MDC.of(LogKeys.APP_ATTEMPT_SHUFFLE_MERGE_ID, appAttemptShuffleMergeId), + MDC.of(LogKeys.REDUCE_ID, reduceId)); + } } } } @@ -2151,6 +2418,22 @@ public File getMergedShuffleMetaFile( shuffleMergeId, reduceId)); return new File(getFilePath(metaName)); } + + /** + * The checksum algorithm is part of the file name, like it is for the checksum file of a + * non-merged shuffle block, so that the checksums are never compared against those of a + * different algorithm after the shuffle server is reconfigured. + */ + public File getMergedShuffleChecksumFile( + int shuffleId, + int shuffleMergeId, + int reduceId, + String algorithm) { + String checksumName = ShuffleChecksumHelper.getChecksumFileName( + String.format("%s.checksum", generateFileName(appId, shuffleId, shuffleMergeId, reduceId)), + algorithm); + return new File(getFilePath(checksumName)); + } } @VisibleForTesting diff --git a/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/protocol/BlockTransferMessage.java b/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/protocol/BlockTransferMessage.java index be057104a1c06..50ff7506d770f 100644 --- a/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/protocol/BlockTransferMessage.java +++ b/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/protocol/BlockTransferMessage.java @@ -48,7 +48,8 @@ public enum Type { FETCH_SHUFFLE_BLOCKS(9), GET_LOCAL_DIRS_FOR_EXECUTORS(10), LOCAL_DIRS_FOR_EXECUTORS(11), PUSH_BLOCK_STREAM(12), FINALIZE_SHUFFLE_MERGE(13), MERGE_STATUSES(14), FETCH_SHUFFLE_BLOCK_CHUNKS(15), DIAGNOSE_CORRUPTION(16), CORRUPTION_CAUSE(17), - PUSH_BLOCK_RETURN_CODE(18), REMOVE_SHUFFLE_MERGE(19); + PUSH_BLOCK_RETURN_CODE(18), REMOVE_SHUFFLE_MERGE(19), + DIAGNOSE_SHUFFLE_CHUNK_CORRUPTION(20); private final byte id; @@ -85,6 +86,7 @@ public static BlockTransferMessage fromByteBuffer(ByteBuffer msg) { case 17 -> CorruptionCause.decode(buf); case 18 -> BlockPushReturnCode.decode(buf); case 19 -> RemoveShuffleMerge.decode(buf); + case 20 -> DiagnoseShuffleChunkCorruption.decode(buf); default -> throw new IllegalArgumentException("Unknown message type: " + type); }; } diff --git a/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/protocol/DiagnoseShuffleChunkCorruption.java b/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/protocol/DiagnoseShuffleChunkCorruption.java new file mode 100644 index 0000000000000..64160eec43203 --- /dev/null +++ b/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/protocol/DiagnoseShuffleChunkCorruption.java @@ -0,0 +1,128 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.network.shuffle.protocol; + +import io.netty.buffer.ByteBuf; +import org.apache.spark.network.protocol.Encoders; + +/** + * Request to get the cause of a corrupted chunk of a merged shuffle partition. Unlike + * {@link DiagnoseCorruption}, this is only served by the external shuffle service that merged + * the chunk, so it identifies the chunk by shuffleMergeId and chunkId instead of by the executor + * and map which produced it. Returns {@link CorruptionCause} + */ +public class DiagnoseShuffleChunkCorruption extends BlockTransferMessage { + public final String appId; + public final int shuffleId; + public final int shuffleMergeId; + public final int reduceId; + public final int chunkId; + public final long checksum; + public final String algorithm; + + public DiagnoseShuffleChunkCorruption( + String appId, + int shuffleId, + int shuffleMergeId, + int reduceId, + int chunkId, + long checksum, + String algorithm) { + this.appId = appId; + this.shuffleId = shuffleId; + this.shuffleMergeId = shuffleMergeId; + this.reduceId = reduceId; + this.chunkId = chunkId; + this.checksum = checksum; + this.algorithm = algorithm; + } + + @Override + protected Type type() { + return Type.DIAGNOSE_SHUFFLE_CHUNK_CORRUPTION; + } + + @Override + public String toString() { + return "DiagnoseShuffleChunkCorruption[appId=" + appId + ",shuffleId=" + shuffleId + + ",shuffleMergeId=" + shuffleMergeId + ",reduceId=" + reduceId + ",chunkId=" + chunkId + + ",checksum=" + checksum + ",algorithm=" + algorithm + "]"; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + DiagnoseShuffleChunkCorruption that = (DiagnoseShuffleChunkCorruption) o; + + if (checksum != that.checksum) return false; + if (shuffleId != that.shuffleId) return false; + if (shuffleMergeId != that.shuffleMergeId) return false; + if (reduceId != that.reduceId) return false; + if (chunkId != that.chunkId) return false; + if (!algorithm.equals(that.algorithm)) return false; + return appId.equals(that.appId); + } + + @Override + public int hashCode() { + int result = appId.hashCode(); + result = 31 * result + Integer.hashCode(shuffleId); + result = 31 * result + Integer.hashCode(shuffleMergeId); + result = 31 * result + Integer.hashCode(reduceId); + result = 31 * result + Integer.hashCode(chunkId); + result = 31 * result + Long.hashCode(checksum); + result = 31 * result + algorithm.hashCode(); + return result; + } + + @Override + public int encodedLength() { + return Encoders.Strings.encodedLength(appId) + + 4 /* encoded length of shuffleId */ + + 4 /* encoded length of shuffleMergeId */ + + 4 /* encoded length of reduceId */ + + 4 /* encoded length of chunkId */ + + 8 /* encoded length of checksum */ + + Encoders.Strings.encodedLength(algorithm); /* encoded length of algorithm */ + } + + @Override + public void encode(ByteBuf buf) { + Encoders.Strings.encode(buf, appId); + buf.writeInt(shuffleId); + buf.writeInt(shuffleMergeId); + buf.writeInt(reduceId); + buf.writeInt(chunkId); + buf.writeLong(checksum); + Encoders.Strings.encode(buf, algorithm); + } + + public static DiagnoseShuffleChunkCorruption decode(ByteBuf buf) { + String appId = Encoders.Strings.decode(buf); + int shuffleId = buf.readInt(); + int shuffleMergeId = buf.readInt(); + int reduceId = buf.readInt(); + int chunkId = buf.readInt(); + long checksum = buf.readLong(); + String algorithm = Encoders.Strings.decode(buf); + return new DiagnoseShuffleChunkCorruption( + appId, shuffleId, shuffleMergeId, reduceId, chunkId, checksum, algorithm); + } +} diff --git a/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/BlockTransferMessagesSuite.java b/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/BlockTransferMessagesSuite.java index 379a321324631..0a8bb763039a5 100644 --- a/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/BlockTransferMessagesSuite.java +++ b/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/BlockTransferMessagesSuite.java @@ -24,6 +24,7 @@ import java.util.HashMap; import java.util.Map; +import org.apache.spark.network.shuffle.checksum.Cause; import org.apache.spark.network.shuffle.protocol.*; /** Verifies that all BlockTransferMessages can be serialized correctly. */ @@ -62,6 +63,15 @@ public void testLocalDirsMessages() { } } + @Test + public void testDiagnosisMessages() { + checkSerializeDeserialize( + new DiagnoseCorruption("app-1", "exec-2", 0, 1L, 2, 12345L, "ADLER32")); + checkSerializeDeserialize( + new DiagnoseShuffleChunkCorruption("app-1", 0, 1, 2, 3, 12345L, "ADLER32")); + checkSerializeDeserialize(new CorruptionCause(Cause.DISK_ISSUE)); + } + private BlockTransferMessage checkSerializeDeserialize(BlockTransferMessage msg) { BlockTransferMessage msg2 = BlockTransferMessage.Decoder.fromByteBuffer(msg.toByteBuffer()); assertEquals(msg, msg2); diff --git a/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/ExternalBlockHandlerSuite.java b/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/ExternalBlockHandlerSuite.java index 2a3135e3c8aeb..fb983a4e99200 100644 --- a/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/ExternalBlockHandlerSuite.java +++ b/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/ExternalBlockHandlerSuite.java @@ -49,6 +49,7 @@ import org.apache.spark.network.shuffle.protocol.BlockTransferMessage; import org.apache.spark.network.shuffle.protocol.CorruptionCause; import org.apache.spark.network.shuffle.protocol.DiagnoseCorruption; +import org.apache.spark.network.shuffle.protocol.DiagnoseShuffleChunkCorruption; import org.apache.spark.network.shuffle.protocol.ExecutorShuffleInfo; import org.apache.spark.network.shuffle.protocol.FetchShuffleBlocks; import org.apache.spark.network.shuffle.protocol.FetchShuffleBlockChunks; @@ -223,6 +224,33 @@ public void testShuffleCorruptionDiagnosisCRC32C() throws IOException { checkDiagnosisResult("CRC32C", Cause.CHECKSUM_VERIFY_PASS); } + @Test + public void testShuffleChunkCorruptionDiagnosis() { + String appId = "app0"; + int shuffleId = 0; + int shuffleMergeId = 1; + int reduceId = 2; + int chunkId = 3; + long checksumByReader = 12345L; + String algorithm = "ADLER32"; + when(mergedShuffleManager.diagnoseShuffleChunkCorruption(appId, shuffleId, shuffleMergeId, + reduceId, chunkId, checksumByReader, algorithm)).thenReturn(Cause.DISK_ISSUE); + + when(client.getClientId()).thenReturn(appId); + RpcResponseCallback callback = mock(RpcResponseCallback.class); + DiagnoseShuffleChunkCorruption diagnoseMsg = new DiagnoseShuffleChunkCorruption( + appId, shuffleId, shuffleMergeId, reduceId, chunkId, checksumByReader, algorithm); + handler.receive(client, diagnoseMsg.toByteBuffer(), callback); + + ArgumentCaptor response = ArgumentCaptor.forClass(ByteBuffer.class); + verify(callback, times(1)).onSuccess(response.capture()); + verify(callback, never()).onFailure(any()); + + CorruptionCause cause = + (CorruptionCause) BlockTransferMessage.Decoder.fromByteBuffer(response.getValue()); + assertEquals(Cause.DISK_ISSUE, cause.cause); + } + @Test public void testFetchShuffleBlocks() { when(blockResolver.getBlockData("app0", "exec1", 0, 0, 0)).thenReturn(blockMarkers[0]); diff --git a/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/RemoteBlockPushResolverSuite.java b/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/RemoteBlockPushResolverSuite.java index b7e24fe3da8fe..bfeb81587b31c 100644 --- a/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/RemoteBlockPushResolverSuite.java +++ b/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/RemoteBlockPushResolverSuite.java @@ -20,8 +20,10 @@ import com.codahale.metrics.Counter; import com.codahale.metrics.Meter; import com.codahale.metrics.Metric; +import java.io.DataInputStream; import java.io.DataOutputStream; import java.io.File; +import java.io.FileInputStream; import java.io.FileOutputStream; import java.io.IOException; import java.nio.ByteBuffer; @@ -30,6 +32,7 @@ import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; +import java.nio.file.StandardOpenOption; import java.util.Arrays; import java.util.List; import java.util.Map; @@ -37,6 +40,8 @@ import java.util.concurrent.Semaphore; import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; +import java.util.zip.CheckedInputStream; +import java.util.zip.Checksum; import com.fasterxml.jackson.databind.ObjectMapper; @@ -53,8 +58,11 @@ import static org.junit.jupiter.api.Assertions.*; import org.apache.spark.network.buffer.FileSegmentManagedBuffer; +import org.apache.spark.network.buffer.ManagedBuffer; import org.apache.spark.network.client.StreamCallbackWithID; import org.apache.spark.network.server.BlockPushNonFatalFailure; +import org.apache.spark.network.shuffle.checksum.Cause; +import org.apache.spark.network.shuffle.checksum.ShuffleChecksumHelper; import org.apache.spark.network.shuffle.RemoteBlockPushResolver.MergeShuffleFile; import org.apache.spark.network.shuffle.RemoteBlockPushResolver.PushMergeMetrics; import org.apache.spark.network.shuffle.protocol.BlockPushReturnCode; @@ -87,6 +95,7 @@ public class RemoteBlockPushResolverSuite { private final String INVALID_MERGE_DIRECTORY_META = "shuffleManager:{\"mergeDirInvalid\": \"merge_manager_2\", \"attemptId\": \"2\"}"; private final String BLOCK_MANAGER_DIR = "blockmgr-193d8401"; + private static final String CHECKSUM_ALGORITHM = "ADLER32"; private TransportConf conf; private RemoteBlockPushResolver pushResolver; @@ -168,6 +177,158 @@ public void testDividingMergedBlocksIntoChunks() throws IOException { verifyMetrics(13, 0, 0, 0, 0, 0, 0); } + @Test + public void testChecksumsOfMergedChunks() throws IOException { + PushBlock[] pushBlocks = new PushBlock[] { + new PushBlock(0, 0, 0, 0, createBuffer(2)), + new PushBlock(0, 0, 1, 0, createBuffer(3)), + new PushBlock(0, 0, 2, 0, createBuffer(5)), + new PushBlock(0, 0, 3, 0, createBuffer(3)) + }; + pushBlockHelper(TEST_APP, NO_ATTEMPT_ID, pushBlocks); + pushResolver.finalizeShuffleMerge(new FinalizeShuffleMerge(TEST_APP, NO_ATTEMPT_ID, 0, 0)); + MergedBlockMeta meta = pushResolver.getMergedBlockMeta(TEST_APP, 0, 0, 0); + // validateChunks verifies that every stored checksum matches the data of its chunk + validateChunks(TEST_APP, 0, 0, 0, meta, new int[]{5, 5, 3}, new int[][]{{0, 1}, {2}, {3}}); + // The checksum file holds one entry per chunk while the index file has an extra leading one + File indexFile = new File(pushResolver.validateAndGetAppShuffleInfo(TEST_APP) + .getMergedShuffleIndexFilePath(0, 0, 0)); + assertEquals(indexFile.length() - 8L, readChunkChecksums(TEST_APP, 0, 0, 0).length * 8L); + } + + @Test + public void testChecksumOfChunkAfterFailedBlockPush() throws IOException { + StreamCallbackWithID stream1 = pushResolver.receiveBlockDataAsStream( + new PushBlockStream(TEST_APP, NO_ATTEMPT_ID, 0, 0, 0, 0, 0)); + stream1.onData(stream1.getID(), createBuffer(2)); + // The partially written block is abandoned, so the next block overwrites its data. The data + // of the abandoned block must not be part of the checksum of the chunk. + stream1.onFailure(stream1.getID(), new RuntimeException("forced error")); + StreamCallbackWithID stream2 = pushResolver.receiveBlockDataAsStream( + new PushBlockStream(TEST_APP, NO_ATTEMPT_ID, 0, 0, 1, 0, 0)); + stream2.onData(stream2.getID(), createBuffer(5)); + stream2.onComplete(stream2.getID()); + pushResolver.finalizeShuffleMerge(new FinalizeShuffleMerge(TEST_APP, NO_ATTEMPT_ID, 0, 0)); + MergedBlockMeta meta = pushResolver.getMergedBlockMeta(TEST_APP, 0, 0, 0); + validateChunks(TEST_APP, 0, 0, 0, meta, new int[]{5}, new int[][]{{1}}); + } + + @Test + public void testChecksumOfChunkWhenBlockPushIsInFlightAtFinalize() throws IOException { + StreamCallbackWithID stream1 = pushResolver.receiveBlockDataAsStream( + new PushBlockStream(TEST_APP, NO_ATTEMPT_ID, 0, 0, 0, 0, 0)); + stream1.onData(stream1.getID(), createBuffer(5)); + stream1.onComplete(stream1.getID()); + StreamCallbackWithID stream2 = pushResolver.receiveBlockDataAsStream( + new PushBlockStream(TEST_APP, NO_ATTEMPT_ID, 0, 0, 1, 0, 0)); + stream2.onData(stream2.getID(), createBuffer(2)); + stream2.onComplete(stream2.getID()); + // This block is still being written when the shuffle merge is finalized, so its data is + // truncated away and must not be part of the checksum of the last chunk. + StreamCallbackWithID stream3 = pushResolver.receiveBlockDataAsStream( + new PushBlockStream(TEST_APP, NO_ATTEMPT_ID, 0, 0, 2, 0, 0)); + stream3.onData(stream3.getID(), createBuffer(3)); + pushResolver.finalizeShuffleMerge(new FinalizeShuffleMerge(TEST_APP, NO_ATTEMPT_ID, 0, 0)); + MergedBlockMeta meta = pushResolver.getMergedBlockMeta(TEST_APP, 0, 0, 0); + validateChunks(TEST_APP, 0, 0, 0, meta, new int[]{5, 2}, new int[][]{{0}, {1}}); + } + + @Test + public void testFailureToWriteChecksumsDoesNotFailTheMerge() throws IOException { + useTestFiles(false, false, true); + StreamCallbackWithID stream = pushResolver.receiveBlockDataAsStream( + new PushBlockStream(TEST_APP, NO_ATTEMPT_ID, 0, 0, 0, 0, 0)); + stream.onData(stream.getID(), createBuffer(5)); + RemoteBlockPushResolver.AppShufflePartitionInfo partitionInfo = + ((RemoteBlockPushResolver.PushBlockStreamCallback) stream).getPartitionInfo(); + // Closing the checksum file makes every write to it fail + ((TestMergeShuffleFile) partitionInfo.getChecksumFile()).close(); + stream.onComplete(stream.getID()); + // The merge of the partition is unaffected by the failure to write the checksums + MergeStatuses statuses = pushResolver.finalizeShuffleMerge( + new FinalizeShuffleMerge(TEST_APP, NO_ATTEMPT_ID, 0, 0)); + validateMergeStatuses(statuses, new int[] {0}, new long[] {5}); + MergedBlockMeta meta = pushResolver.getMergedBlockMeta(TEST_APP, 0, 0, 0); + assertEquals(1, meta.getNumChunks(), "num chunks"); + assertEquals(0, partitionInfo.getNumIOExceptions(), "no IOExceptions counted for the merge"); + // The checksums no longer describe the chunks, so they are discarded and the corruption of + // the chunk can no longer be diagnosed + assertFalse(pushResolver.validateAndGetAppShuffleInfo(TEST_APP) + .getMergedShuffleChecksumFile(0, 0, 0, CHECKSUM_ALGORITHM).exists(), "no checksum file"); + assertEquals(Cause.UNKNOWN_ISSUE, pushResolver.diagnoseShuffleChunkCorruption( + TEST_APP, 0, 0, 0, 0, 0L, CHECKSUM_ALGORITHM)); + } + + @Test + public void testShuffleChunkCorruptionDiagnosisChecksumVerifyPass() throws IOException { + prepareChunkForDiagnosis(); + long checksumByReader = readChunkChecksums(TEST_APP, 0, 0, 0)[0]; + assertEquals(Cause.CHECKSUM_VERIFY_PASS, pushResolver.diagnoseShuffleChunkCorruption( + TEST_APP, 0, 0, 0, 0, checksumByReader, CHECKSUM_ALGORITHM)); + } + + @Test + public void testShuffleChunkCorruptionDiagnosisNetworkIssue() throws IOException { + prepareChunkForDiagnosis(); + // The chunk on disk is intact, but the reducer read something else + long checksumByReader = readChunkChecksums(TEST_APP, 0, 0, 0)[0] + 1; + assertEquals(Cause.NETWORK_ISSUE, pushResolver.diagnoseShuffleChunkCorruption( + TEST_APP, 0, 0, 0, 0, checksumByReader, CHECKSUM_ALGORITHM)); + } + + @Test + public void testShuffleChunkCorruptionDiagnosisDiskIssue() throws IOException { + prepareChunkForDiagnosis(); + long checksumByReader = readChunkChecksums(TEST_APP, 0, 0, 0)[0]; + // Corrupt the merged shuffle data after it was merged and checksummed + File dataFile = pushResolver.validateAndGetAppShuffleInfo(TEST_APP) + .getMergedShuffleDataFile(0, 0, 0); + try (FileChannel channel = FileChannel.open(dataFile.toPath(), StandardOpenOption.WRITE)) { + channel.write(ByteBuffer.wrap(new byte[] {(byte) 0xff, (byte) 0xff}), 0); + } + assertEquals(Cause.DISK_ISSUE, pushResolver.diagnoseShuffleChunkCorruption( + TEST_APP, 0, 0, 0, 0, checksumByReader, CHECKSUM_ALGORITHM)); + } + + @Test + public void testShuffleChunkCorruptionDiagnosisUnsupportedAlgorithm() throws IOException { + prepareChunkForDiagnosis(); + // The reducer calculated its checksum with an algorithm the chunk was not merged with + assertEquals(Cause.UNSUPPORTED_CHECKSUM_ALGORITHM, + pushResolver.diagnoseShuffleChunkCorruption(TEST_APP, 0, 0, 0, 0, 0L, "CRC32")); + } + + @Test + public void testShuffleChunkCorruptionDiagnosisOfUnknownChunk() throws IOException { + prepareChunkForDiagnosis(); + assertEquals(Cause.UNKNOWN_ISSUE, pushResolver.diagnoseShuffleChunkCorruption( + TEST_APP, 0, 0, 0, 5, 0L, CHECKSUM_ALGORITHM)); + } + + @Test + public void testShuffleChunkCorruptionDiagnosisWhenChecksumIsDisabled() throws IOException { + MapConfigProvider provider = new MapConfigProvider(Map.of( + "spark.shuffle.push.server.minChunkSizeInMergedShuffleFile", "4", + "spark.shuffle.push.server.mergedShuffleChecksum.enabled", "false")); + pushResolver = new RemoteBlockPushResolver(new TransportConf("shuffle", provider), null); + registerExecutor(TEST_APP, prepareLocalDirs(localDirs, MERGE_DIRECTORY), MERGE_DIRECTORY_META); + pushBlockHelper(TEST_APP, NO_ATTEMPT_ID, new PushBlock[] { + new PushBlock(0, 0, 0, 0, createBuffer(5)) + }); + pushResolver.finalizeShuffleMerge(new FinalizeShuffleMerge(TEST_APP, NO_ATTEMPT_ID, 0, 0)); + assertFalse(pushResolver.validateAndGetAppShuffleInfo(TEST_APP) + .getMergedShuffleChecksumFile(0, 0, 0, CHECKSUM_ALGORITHM).exists(), "no checksum file"); + assertEquals(Cause.UNKNOWN_ISSUE, pushResolver.diagnoseShuffleChunkCorruption( + TEST_APP, 0, 0, 0, 0, 0L, CHECKSUM_ALGORITHM)); + } + + private void prepareChunkForDiagnosis() throws IOException { + pushBlockHelper(TEST_APP, NO_ATTEMPT_ID, new PushBlock[] { + new PushBlock(0, 0, 0, 0, createBuffer(5)) + }); + pushResolver.finalizeShuffleMerge(new FinalizeShuffleMerge(TEST_APP, NO_ATTEMPT_ID, 0, 0)); + } + @Test public void testFinalizeWithMultipleReducePartitions() throws IOException { PushBlock[] pushBlocks = new PushBlock[] { @@ -1220,6 +1381,9 @@ void closeAndDeleteOutdatedPartitions( "Meta files on the disk should be cleaned up"); assertFalse(new File(appShuffleInfo.getMergedShuffleIndexFilePath(0, 1, 0)).exists(), "Index files on the disk should be cleaned up"); + assertFalse( + appShuffleInfo.getMergedShuffleChecksumFile(0, 1, 0, CHECKSUM_ALGORITHM).exists(), + "Checksum files on the disk should be cleaned up"); stream2.onData(stream2.getID(), ByteBuffer.wrap(new byte[2])); stream2.onData(stream2.getID(), ByteBuffer.wrap(new byte[2])); // stream 2 now completes @@ -1411,12 +1575,14 @@ void deleteMergedFiles( assertTrue(shuffleInfo.getMergedShuffleMetaFile(0, 1, 0).exists()); assertTrue(new File(shuffleInfo.getMergedShuffleIndexFilePath(0, 1, 0)).exists()); assertTrue(shuffleInfo.getMergedShuffleDataFile(0, 1, 0).exists()); + assertTrue(shuffleInfo.getMergedShuffleChecksumFile(0, 1, 0, CHECKSUM_ALGORITHM).exists()); pushResolver.removeShuffleMerge( new RemoveShuffleMerge(testApp, NO_ATTEMPT_ID, 0, 1)); closed.tryAcquire(10, TimeUnit.SECONDS); assertFalse(shuffleInfo.getMergedShuffleMetaFile(0, 1, 0).exists()); assertFalse(new File(shuffleInfo.getMergedShuffleIndexFilePath(0, 1, 0)).exists()); assertFalse(shuffleInfo.getMergedShuffleDataFile(0, 1, 0).exists()); + assertFalse(shuffleInfo.getMergedShuffleChecksumFile(0, 1, 0, CHECKSUM_ALGORITHM).exists()); // 1.2 Cleaned up the merged files when msg.shuffleMergeId is DELETE_ALL_MERGED_SHUFFLE StreamCallbackWithID streamCallback1 = pushResolver.receiveBlockDataAsStream( @@ -1489,6 +1655,13 @@ void deleteMergedFiles( } private void useTestFiles(boolean useTestIndexFile, boolean useTestMetaFile) throws IOException { + useTestFiles(useTestIndexFile, useTestMetaFile, false); + } + + private void useTestFiles( + boolean useTestIndexFile, + boolean useTestMetaFile, + boolean useTestChecksumFile) throws IOException { pushResolver = new RemoteBlockPushResolver(conf, null) { @Override AppShufflePartitionInfo newAppShufflePartitionInfo( @@ -1505,9 +1678,14 @@ AppShufflePartitionInfo newAppShufflePartitionInfo( MergeShuffleFile mergedMetaFile = useTestMetaFile ? new TestMergeShuffleFile(metaFile) : new MergeShuffleFile(metaFile); + File checksumFile = appShuffleInfo.getMergedShuffleChecksumFile( + shuffleId, shuffleMergeId, reduceId, CHECKSUM_ALGORITHM); + MergeShuffleFile mergedChecksumFile = useTestChecksumFile ? + new TestMergeShuffleFile(checksumFile) : + new MergeShuffleFile(checksumFile); return new AppShufflePartitionInfo(new AppAttemptShuffleMergeId( appShuffleInfo.appId, appShuffleInfo.attemptId, shuffleId, shuffleMergeId), reduceId, - dataFile, mergedIndexFile, mergedMetaFile); + dataFile, mergedIndexFile, mergedMetaFile, mergedChecksumFile, CHECKSUM_ALGORITHM); } }; registerExecutor(TEST_APP, prepareLocalDirs(localDirs, MERGE_DIRECTORY), MERGE_DIRECTORY_META); @@ -1572,6 +1750,62 @@ private void validateChunks( shuffleMergeId, reduceId, i); assertEquals(expectedSizes[i], mb.getLength()); } + validateChunkChecksums(appId, shuffleId, shuffleMergeId, reduceId, meta.getNumChunks()); + } + + /** + * Verifies that the checksum file holds exactly one checksum per chunk and that every one of + * them describes the data the chunk ended up with in the merged shuffle data file. + */ + private void validateChunkChecksums( + String appId, + int shuffleId, + int shuffleMergeId, + int reduceId, + int numChunks) throws IOException { + long[] checksums = readChunkChecksums(appId, shuffleId, shuffleMergeId, reduceId); + assertEquals(numChunks, checksums.length, "num checksums"); + for (int i = 0; i < numChunks; i++) { + ManagedBuffer chunk = + pushResolver.getMergedBlockData(appId, shuffleId, shuffleMergeId, reduceId, i); + assertEquals(calculateChecksum(chunk), checksums[i], "checksum of chunk " + i); + } + } + + private long[] readChunkChecksums( + String appId, + int shuffleId, + int shuffleMergeId, + int reduceId) throws IOException { + File checksumFile = pushResolver.validateAndGetAppShuffleInfo(appId) + .getMergedShuffleChecksumFile(shuffleId, shuffleMergeId, reduceId, CHECKSUM_ALGORITHM); + assertTrue(checksumFile.exists(), "checksum file " + checksumFile.getName() + " exists"); + long[] checksums = new long[(int) (checksumFile.length() / 8L)]; + try (DataInputStream in = new DataInputStream(new FileInputStream(checksumFile))) { + for (int i = 0; i < checksums.length; i++) { + checksums[i] = in.readLong(); + } + } + return checksums; + } + + private long calculateChecksum(ManagedBuffer data) throws IOException { + Checksum checksum = ShuffleChecksumHelper.getChecksumByAlgorithm(CHECKSUM_ALGORITHM); + byte[] buffer = new byte[ShuffleChecksumHelper.CHECKSUM_CALCULATION_BUFFER]; + try (CheckedInputStream in = new CheckedInputStream(data.createInputStream(), checksum)) { + while (in.read(buffer) != -1) {} + return checksum.getValue(); + } + } + + /** + * Blocks of random data, so that the checksum of a chunk actually depends on which data ended + * up in it. + */ + private ByteBuffer createBuffer(int size) { + byte[] bytes = new byte[size]; + ThreadLocalRandom.current().nextBytes(bytes); + return ByteBuffer.wrap(bytes); } private void pushBlockHelper( diff --git a/common/utils-java/src/main/java/org/apache/spark/internal/LogKeys.java b/common/utils-java/src/main/java/org/apache/spark/internal/LogKeys.java index 91bff3120f002..4881cd3ce4418 100644 --- a/common/utils-java/src/main/java/org/apache/spark/internal/LogKeys.java +++ b/common/utils-java/src/main/java/org/apache/spark/internal/LogKeys.java @@ -94,7 +94,9 @@ public enum LogKeys implements LogKey { CHECKPOINT_ROOT, CHECKPOINT_TIME, CHECKSUM, + CHECKSUM_ALGORITHM, CHOSEN_WATERMARK, + CHUNK_ID, CLASSIFIER, CLASS_LOADER, CLASS_NAME, @@ -397,6 +399,7 @@ public enum LogKeys implements LogKey { MEMORY_POOL_NAME, MEMORY_SIZE, MEMORY_THRESHOLD_SIZE, + MERGED_SHUFFLE_CHECKSUM_ALGORITHM, MERGE_BYTES_WRITTEN, MERGE_DIR_NAME, MERGE_FACTOR, diff --git a/core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala b/core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala index cb15e954bb38a..bf271738849c7 100644 --- a/core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala +++ b/core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala @@ -922,16 +922,26 @@ final class ShuffleBlockFetcherIterator( } } catch { case e: IOException => - // When shuffle checksum is enabled, for a block that is corrupted twice, - // we'd calculate the checksum of the block by consuming the remaining data - // in the buf. So, we should release the buf later. - if (!(checksumEnabled && corruptedBlocks.contains(blockId))) { + // When shuffle checksum is enabled, for a block that is corrupted twice or for a + // corrupted shuffle chunk, we'd calculate the checksum of the block by consuming + // the remaining data in the buf. So, we should release the buf later. + if (!(checksumEnabled && + (corruptedBlocks.contains(blockId) || blockId.isShuffleChunk))) { buf.release() } if (blockId.isShuffleChunk) { shuffleMetrics.incCorruptMergedBlockChunks(1) - // TODO (SPARK-36284): Add shuffle checksum support for push-based shuffle + if (checksumEnabled) { + // Diagnose the cause of data corruption if shuffle checksum is enabled. The + // diagnosis is only informational since the fallback below recovers from the + // corruption either way. + try { + logWarning(diagnoseCorruption(checkedIn, address, blockId)) + } finally { + buf.release() + } + } // Retrying a corrupt block may result again in a corrupt block. For shuffle // chunks, we opt to fallback on the original shuffle blocks that belong to that // corrupt shuffle chunk immediately instead of retrying to fetch the corrupt @@ -1122,46 +1132,25 @@ final class ShuffleBlockFetcherIterator( logInfo("Start corruption diagnosis.") blockId match { case shuffleBlock: ShuffleBlockId => - val startTimeNs = System.nanoTime() - val buffer = new Array[Byte](ShuffleChecksumHelper.CHECKSUM_CALCULATION_BUFFER) - // consume the remaining data to calculate the checksum - var cause: Cause = null - try { - while (checkedIn.read(buffer) != -1) {} - val checksum = checkedIn.getChecksum.getValue - cause = shuffleClient.diagnoseCorruption(address.host, address.port, address.executorId, + diagnoseCorruptionWithChecksum(checkedIn, blockId, "Block") { checksum => + shuffleClient.diagnoseCorruption(address.host, address.port, address.executorId, shuffleBlock.shuffleId, shuffleBlock.mapId, shuffleBlock.reduceId, checksum, checksumAlgorithm) - } catch { - case e: Exception => - logWarning("Unable to diagnose the corruption cause of the corrupted block", e) - cause = Cause.UNKNOWN_ISSUE } - val duration = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTimeNs) - val diagnosisResponse = cause match { - case Cause.UNSUPPORTED_CHECKSUM_ALGORITHM => - s"Block $blockId is corrupted but corruption diagnosis failed due to " + - s"unsupported checksum algorithm: $checksumAlgorithm" - - case Cause.CHECKSUM_VERIFY_PASS => - s"Block $blockId is corrupted but checksum verification passed" - - case Cause.UNKNOWN_ISSUE => - s"Block $blockId is corrupted but the cause is unknown" - - case otherCause => - s"Block $blockId is corrupted due to $otherCause" - } - logInfo(log"Finished corruption diagnosis in ${MDC(DURATION, duration)} ms. " + - log"${MDC(STATUS, diagnosisResponse)}") - diagnosisResponse case shuffleBlockChunk: ShuffleBlockChunkId => - // TODO SPARK-36284 Add shuffle checksum support for push-based shuffle - logWarning(log"BlockChunk ${MDC(SHUFFLE_BLOCK_INFO, shuffleBlockChunk)} " + - log"is corrupted but corruption diagnosis is skipped due to lack of shuffle " + - log"checksum support for push-based shuffle.") - s"BlockChunk $shuffleBlockChunk is corrupted but corruption " + - s"diagnosis is skipped due to lack of shuffle checksum support for push-based shuffle." + // A shuffle chunk is always merged and served by an external shuffle service, but the + // address of a push-merged-local chunk carries the port of the block manager of this + // executor instead of the one of the shuffle service. + val port = if (pushBasedFetchHelper.isLocalPushMergedBlockAddress(address)) { + blockManager.externalShuffleServicePort + } else { + address.port + } + diagnoseCorruptionWithChecksum(checkedIn, blockId, "BlockChunk") { checksum => + shuffleClient.diagnoseShuffleChunkCorruption(address.host, port, + shuffleBlockChunk.shuffleId, shuffleBlockChunk.shuffleMergeId, + shuffleBlockChunk.reduceId, shuffleBlockChunk.chunkId, checksum, checksumAlgorithm) + } case shuffleBlockBatch: ShuffleBlockBatchId => logWarning(log"BlockBatch ${MDC(SHUFFLE_BLOCK_INFO, shuffleBlockBatch)} is corrupted " + log"but corruption diagnosis is skipped due to lack of shuffle checksum support for " + @@ -1174,6 +1163,53 @@ final class ShuffleBlockFetcherIterator( } } + /** + * Consumes the remaining data of the corrupted block to get the checksum calculated by this + * reader, asks the server which wrote the block for the cause of the corruption, and turns the + * cause into the diagnosis message. + * + * @param checkedIn the [[CheckedInputStream]] which is used to calculate the checksum. + * @param blockId the blockId of the corrupted block. + * @param blockKind how the corrupted block is named in the diagnosis message. + * @param diagnose sends the diagnosis request for the given reader checksum to the server. + * @return The corruption diagnosis response for different causes. + */ + private def diagnoseCorruptionWithChecksum( + checkedIn: CheckedInputStream, + blockId: BlockId, + blockKind: String)(diagnose: Long => Cause): String = { + val startTimeNs = System.nanoTime() + val buffer = new Array[Byte](ShuffleChecksumHelper.CHECKSUM_CALCULATION_BUFFER) + // consume the remaining data to calculate the checksum + var cause: Cause = null + try { + while (checkedIn.read(buffer) != -1) {} + cause = diagnose(checkedIn.getChecksum.getValue) + } catch { + case e: Exception => + logWarning("Unable to diagnose the corruption cause of the corrupted block", e) + cause = Cause.UNKNOWN_ISSUE + } + val duration = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTimeNs) + val diagnosisResponse = cause match { + case Cause.UNSUPPORTED_CHECKSUM_ALGORITHM => + s"$blockKind $blockId is corrupted but corruption diagnosis failed due to " + + s"unsupported checksum algorithm: $checksumAlgorithm" + + case Cause.CHECKSUM_VERIFY_PASS => + s"$blockKind $blockId is corrupted but checksum verification passed" + + case Cause.UNKNOWN_ISSUE => + s"$blockKind $blockId is corrupted but the cause is unknown" + + case otherCause => + s"$blockKind $blockId is corrupted due to $otherCause" + } + logInfo(log"Finished corruption diagnosis in ${MDC(DURATION, duration)} ms. " + + log"${MDC(STATUS, diagnosisResponse)}") + diagnosisResponse + } + def toCompletionIterator: Iterator[(BlockId, InputStream)] = { CompletionIterator[(BlockId, InputStream), this.type](this, onCompleteCallback.onComplete(context)) diff --git a/core/src/test/scala/org/apache/spark/storage/ShuffleBlockFetcherIteratorSuite.scala b/core/src/test/scala/org/apache/spark/storage/ShuffleBlockFetcherIteratorSuite.scala index f9d2fdf941d02..825fb87836c8a 100644 --- a/core/src/test/scala/org/apache/spark/storage/ShuffleBlockFetcherIteratorSuite.scala +++ b/core/src/test/scala/org/apache/spark/storage/ShuffleBlockFetcherIteratorSuite.scala @@ -19,6 +19,7 @@ package org.apache.spark.storage import java.io._ import java.nio.ByteBuffer +import java.nio.charset.StandardCharsets import java.util.UUID import java.util.concurrent.{CompletableFuture, Semaphore} import java.util.zip.CheckedInputStream @@ -42,6 +43,7 @@ import org.apache.spark.MapOutputTracker.SHUFFLE_PUSH_MAP_ID import org.apache.spark.network._ import org.apache.spark.network.buffer.{FileSegmentManagedBuffer, ManagedBuffer} import org.apache.spark.network.shuffle.{BlockFetchingListener, DownloadFileManager, ExternalBlockStoreClient, MergedBlockMeta, MergedBlocksMetaListener} +import org.apache.spark.network.shuffle.checksum.{Cause, ShuffleChecksumHelper} import org.apache.spark.network.util.LimitedInputStream import org.apache.spark.shuffle.{FetchFailedException, ShuffleReadMetricsReporter} import org.apache.spark.storage.BlockManagerId.SHUFFLE_MERGER_IDENTIFIER @@ -1975,6 +1977,81 @@ class ShuffleBlockFetcherIteratorSuite extends SparkFunSuite { } } + test("SPARK-36284: diagnose the corruption of a push-merged shuffle chunk") { + val remoteBmId = BlockManagerId("test-client-1", "test-client-1", 2) + val blocks = Map[BlockId, ManagedBuffer]( + ShuffleBlockId(0, 0, 0) -> createMockManagedBuffer()) + answerFetchBlocks { invocation => + val listener = invocation.getArgument[BlockFetchingListener](4) + listener.onBlockFetchSuccess(ShuffleBlockId(0, 0, 0).toString, createMockManagedBuffer()) + } + val blockManager = createMockBlockManager() + doReturn(7337).when(blockManager).externalShuffleServicePort + val iterator = createShuffleBlockIteratorWithDefaults( + Map(remoteBmId -> toBlockList(blocks.keys, 1L, 0)), + blockManager = Some(blockManager)) + val chunkId = ShuffleBlockChunkId(0, 3, 2, 1) + + // A push-merged-local chunk is diagnosed against the local shuffle service, whose port is + // not the one the address of the chunk carries. + when(transfer.diagnoseShuffleChunkCorruption( + any(), any(), any(), any(), any(), any(), any(), any())).thenReturn(Cause.DISK_ISSUE) + val localMergerBmId = BlockManagerId(SHUFFLE_MERGER_IDENTIFIER, "test-local-host", 1) + val localResponse = iterator.diagnoseCorruption( + createCheckedInputStream("chunk data"), localMergerBmId, chunkId) + verify(transfer, times(1)).diagnoseShuffleChunkCorruption( + meq("test-local-host"), meq(7337), meq(0), meq(3), meq(2), meq(1), any(), meq("ADLER32")) + assert(localResponse === s"BlockChunk $chunkId is corrupted due to DISK_ISSUE") + + // A remote chunk is diagnosed against the shuffle service its address points at + val remoteMergerBmId = BlockManagerId(SHUFFLE_MERGER_IDENTIFIER, "test-remote-host", 7337) + when(transfer.diagnoseShuffleChunkCorruption( + any(), any(), any(), any(), any(), any(), any(), any())) + .thenReturn(Cause.CHECKSUM_VERIFY_PASS) + val remoteResponse = iterator.diagnoseCorruption( + createCheckedInputStream("chunk data"), remoteMergerBmId, chunkId) + verify(transfer, times(1)).diagnoseShuffleChunkCorruption( + meq("test-remote-host"), meq(7337), meq(0), meq(3), meq(2), meq(1), any(), meq("ADLER32")) + assert(remoteResponse === + s"BlockChunk $chunkId is corrupted but checksum verification passed") + } + + test("SPARK-36284: diagnose a corrupt push-merged shuffle chunk before falling back") { + val blockManager = mock(classOf[BlockManager]) + val localDirs = Array("local-dir") + val blocksByAddress = prepareForFallbackToLocalBlocks( + blockManager, Map(SHUFFLE_MERGER_IDENTIFIER -> localDirs)) + val corruptBuffer = createMockManagedBuffer(2) + doReturn(Seq({corruptBuffer})).when(blockManager) + .getLocalMergedBlockData(ShuffleMergedBlockId(0, 0, 2), localDirs) + val corruptStream = mock(classOf[InputStream]) + when(corruptStream.read(any(), any(), any())).thenThrow(new IOException("corrupt")) + doReturn(corruptStream).when(corruptBuffer).createInputStream() + val taskContext = TaskContext.empty() + val shuffleMetrics = taskContext.taskMetrics.createTempShuffleReadMetrics() + val logAppender = new LogAppender("diagnose corruption of a shuffle chunk") + withLogAppender(logAppender) { + val iterator = createShuffleBlockIteratorWithDefaults( + blocksByAddress, + blockManager = Some(blockManager), + taskContext = Some(taskContext), + shuffleMetrics = Some(shuffleMetrics), + streamWrapperLimitSize = Some(100)) + // The corruption is diagnosed, and the original shuffle blocks are fetched either way + verifyLocalBlocksFromFallback(iterator) + } + assert(logAppender.loggingEvents.count( + _.getMessage.getFormattedMessage.contains("Start corruption diagnosis")) === 1) + assert(shuffleMetrics.corruptMergedBlockChunks === 1) + assert(shuffleMetrics.mergedFetchFallbackCount === 1) + } + + private def createCheckedInputStream(data: String): CheckedInputStream = { + new CheckedInputStream( + new ByteArrayInputStream(data.getBytes(StandardCharsets.UTF_8)), + ShuffleChecksumHelper.getChecksumByAlgorithm("ADLER32")) + } + test("SPARK-52395: Fast fail when fetch failure happens for local blocks") { val blockManager = createMockBlockManager() val localBmId = blockManager.blockManagerId diff --git a/docs/configuration.md b/docs/configuration.md index 38460273a823b..522e79688bfd6 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -4124,6 +4124,22 @@ Push-based shuffle helps improve the reliability and performance of spark shuffl 3.2.0 + + spark.shuffle.push.server.mergedShuffleChecksum.enabled + true + + Whether the external shuffle service calculates the checksum of every chunk of a merged shuffle partition while it merges the pushed blocks. The checksums are stored next to the merged shuffle data and are only used to diagnose the cause of a corrupted shuffle chunk, in the same way spark.shuffle.checksum.enabled is used for the shuffle blocks that are not merged. Diagnosing a corrupted shuffle chunk additionally requires spark.shuffle.checksum.enabled to be set in the application reading the chunk. + + 4.4.0 + + + spark.shuffle.push.server.mergedShuffleChecksum.algorithm + ADLER32 + + The algorithm used to calculate the checksums of the merged shuffle chunks. Currently, it supports ADLER32, CRC32 and CRC32C. The reducer calculates the checksum of a corrupted chunk with spark.shuffle.checksum.algorithm, so the corruption of a merged shuffle chunk can only be diagnosed when the two configurations match. An unsupported algorithm only disables the calculation of the checksums instead of failing the merge of the pushed blocks. + + 4.4.0 + ### Client side configuration options