[ https://issues.apache.org/jira/browse/FLINK-18811?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Piotr Nowojski closed FLINK-18811. ---------------------------------- Fix Version/s: 1.12.0 Resolution: Fixed merged commit e38716f into apache:master > if a disk is damaged, taskmanager should choose another disk for temp dir , > rather than throw an IOException, which causes flink job restart over and > over again > ---------------------------------------------------------------------------------------------------------------------------------------------------------------- > > Key: FLINK-18811 > URL: https://issues.apache.org/jira/browse/FLINK-18811 > Project: Flink > Issue Type: Improvement > Components: Runtime / Network > Environment: flink-1.10 > Reporter: Kai Chen > Assignee: Kai Chen > Priority: Major > Labels: pull-request-available > Fix For: 1.12.0 > > Attachments: flink_disk_error.png > > > I met this Exception when a hard disk was damaged: > !flink_disk_error.png! > I checked the code and found that flink will create a temp file when Record > length > 5 MB: > > {code:java} > // SpillingAdaptiveSpanningRecordDeserializer.java > if (nextRecordLength > THRESHOLD_FOR_SPILLING) { > // create a spilling channel and put the data there > this.spillingChannel = createSpillingChannel(); > ByteBuffer toWrite = partial.segment.wrap(partial.position, numBytesChunk); > FileUtils.writeCompletely(this.spillingChannel, toWrite); > } > {code} > The tempDir is random picked from all `tempDirs`. Well on yarn mode, one > `tempDir` usually represents one hard disk. > > In may opinion, if a hard disk is damaged, taskmanager should pick another > disk(tmpDir) for Spilling Channel, rather than throw an IOException, which > causes flink job restart over and over again. > If we could just change “SpillingAdaptiveSpanningRecordDeserializer" like > this: > {code:java} > // SpillingAdaptiveSpanningRecordDeserializer.java > private FileChannel createSpillingChannel() throws IOException { > if (spillFile != null) { > throw new IllegalStateException("Spilling file already exists."); > } > // try to find a unique file name for the spilling channel > int maxAttempts = 10; > String[] tempDirs = this.tempDirs; > for (int attempt = 0; attempt < maxAttempts; attempt++) { > int dirIndex = rnd.nextInt(tempDirs.length); > String directory = tempDirs[dirIndex]; > spillFile = new File(directory, randomString(rnd) + ".inputchannel"); > try { > if (spillFile.createNewFile()) { > return new RandomAccessFile(spillFile, "rw").getChannel(); > } > } catch (IOException e) { > // if there is no tempDir left to try > if(tempDirs.length <= 1) { > throw e; > } > LOG.warn("Caught an IOException when creating spill file: " + > directory + ". Attempt " + attempt, e); > tempDirs = (String[])ArrayUtils.remove(tempDirs,dirIndex); > } > } > {code} -- This message was sent by Atlassian Jira (v8.3.4#803005)