Skip to content

[SPARK-8498][TUNGSTEN] fix npe in errorhandling path in unsafeshuffle writer #6918

New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Closed
holdenk wants to merge 7 commits into from
Closed

[SPARK-8498][TUNGSTEN] fix npe in errorhandling path in unsafeshuffle writer #6918

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
2e6abf7
Be more cautious when cleaning up during failed write and re-throw us…
holdenk Jun 20, 2015
ae0b7a7
Fix the test
holdenk Jun 20, 2015
e503b8c
Improve the test to ensure we aren't masking the underlying exception
holdenk Jun 20, 2015
30e558d
go back to try/finally
holdenk Jun 21, 2015
039d620
Add missing closeandwriteoutput
holdenk Jun 21, 2015
855f9aa
Spelling - not my strongest suite. Fix Propegates to Propagates.
holdenk Jun 22, 2015
f807832
Log error if we can't throw it
holdenk Jun 23, 2015
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 16 additions & 2 deletions core/src/main/java/org/apache/spark/shuffle/unsafe/UnsafeShuffleWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -139,6 +139,9 @@ public void write(Iterator<Product2<K, V>> records) throws IOException {

@Override
public void write(scala.collection.Iterator<Product2<K, V>> records) throws IOException {
// Keep track of success so we know if we ecountered an exception
// We do this rather than a standard try/catch/re-throw to handle
// generic throwables.
boolean success = false;
try {
while (records.hasNext()) {
Expand All @@ -147,8 +150,19 @@ public void write(scala.collection.Iterator<Product2<K, V>> records) throws IOEx
closeAndWriteOutput();
success = true;
} finally {
if (!success) {
sorter.cleanupAfterError();
if (sorter != null) {
try {
sorter.cleanupAfterError();
} catch (Exception e) {
Copy link

@stephenh stephenh Jun 21, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You could potentially use Utils.tryWithSafeFinally, which does this sort of annoying "try/catch in a try/catch" thing.

Copy link
Contributor

@JoshRosen JoshRosen Jun 21, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This class, along with a lot of the other Tungsten code, is written in Java, so we can't use Utils.tryWithSafeFinally here.

// Only throw this error if we won't be masking another
// error.
if (success) {
throw e;
} else {
logger.error("In addition to a failure during writing, we failed during " +
"cleanup.", e);
}
Copy link
Contributor

@JoshRosen JoshRosen Jun 22, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Shouldn't we log in the else branch to make sure that the error is never silently swallowed?

}
}
}
}
Expand Down
17 changes: 17 additions & 0 deletions core/src/test/java/org/apache/spark/shuffle/unsafe/UnsafeShuffleWriterSuite.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -253,6 +253,23 @@ public void doNotNeedToCallWriteBeforeUnsuccessfulStop() throws IOException {
createWriter(false).stop(false);
}

class PandaException extends RuntimeException {
}

@Test(expected=PandaException.class)
public void writeFailurePropagates() throws Exception {
class BadRecords extends scala.collection.AbstractIterator<Product2<Object, Object>> {
@Override public boolean hasNext() {
throw new PandaException();
}
@Override public Product2<Object, Object> next() {
return null;
}
}
final UnsafeShuffleWriter<Object, Object> writer = createWriter(true);
writer.write(new BadRecords());
}

@Test
public void writeEmptyIterator() throws Exception {
final UnsafeShuffleWriter<Object, Object> writer = createWriter(true);
Expand Down