[
https://issues.apache.org/jira/browse/MAPREDUCE-6166?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14225864#comment-14225864
]
Gera Shegalov commented on MAPREDUCE-6166:
------------------------------------------
Sounds good, [~eepayne].
I have a few comments then. Some are in light of a follow-up JIRA.
bq. update this patch with the final keyword on JobConf jobConf
Let us make the instance variable type a more general {{Configuration}} as we
are not doing anything specific to {{JobConf}}.
Instead of introducing a new local variable iFin in
{{OnDiskMapOutput#shuffle}}, we can overwrite it as in
{{InMemoryMapOutput#shuffle}}.
We can either capture the shuffle size in an instance variable as
{{InMemoryMapOutput}} does implicitly via {{memory.length}}. Or we can set
{code}
long bytesLeft = compressedLength - ((IFileInputStream)input).getSize()
{code}
Then we don't need to touch the line {{input.read}} to do {{readWithChecksum}}
Good call adding {{finally}} with {{close}}.
I also have some comments for the test:
{{ios.finish()}} should be removed because it's redundant:
{{IFileOutputStream#close()}} will call it as well.
We don't need PrintStream wrapping and we need to be careful not to leak file
descriptors in case I/O fails.
{code}
new PrintStream(fout).print(bout.toString());
fout.close();
{code}
Should be something like:
{code}
try {
fout.write(bout.toByteArray());
} finally {
fout.close();
}
{code}
Similarly we need to make sure that {{fin.close()}} is in a try-finally block
enclosing header and shuffle read.
Let us not do
{code}
catch(Exception e) {
fail("OnDiskMapOutput.shuffle did not process the map partition file");
{code}
It's redundant because the exception is failing the test already.
Same PrintStream and fout.close remarks for the code creating the corrupted file
{{dataSize/2}}: I believe Sun Java Coding Style require spaces around
arithmetic operations.
In the fragment where we expect the checksum to fail, {{fin.close()}} should be
in some finally.
{{catch(Exception e)}} is too broad. Let us be more specific and maybe even log
it:
{code}
} catch(ChecksumException e) {
LOG.info("Expected checksum exception thrown.", e);
}
{code}
Thinking a bit more about the file.out, it does not seem to be cleaned up after
the test has finished. But we probably don't even need to create files, we can
simply use {{new ByteArrayInputStream(bout.toByteArray())}} and {{new
ByteArrayInputStream(corrupted)}} as input.
> Reducers do not catch bad map output transfers during shuffle if data
> shuffled directly to disk
> -----------------------------------------------------------------------------------------------
>
> Key: MAPREDUCE-6166
> URL: https://issues.apache.org/jira/browse/MAPREDUCE-6166
> Project: Hadoop Map/Reduce
> Issue Type: Bug
> Components: mrv2
> Affects Versions: 2.6.0
> Reporter: Eric Payne
> Assignee: Eric Payne
> Attachments: MAPREDUCE-6166.v1.201411221941.txt,
> MAPREDUCE-6166.v2.201411251627.txt
>
>
> In very large map/reduce jobs (50000 maps, 2500 reducers), the intermediate
> map partition output gets corrupted on disk on the map side. If this
> corrupted map output is too large to shuffle in memory, the reducer streams
> it to disk without validating the checksum. In jobs this large, it could take
> hours before the reducer finally tries to read the corrupted file and fails.
> Since retries of the failed reduce attempt will also take hours, this delay
> in discovering the failure is multiplied greatly.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
