Fix data corrupted issue when loading from UFS #18500
Conversation
beinan
force-pushed
the
fix_data_corruption
branch
from
66c2662
to
8a89d03
Compare
| try { | ||
| // Read the entire block, caching to block store will be handled internally in UFS block | ||
| // store when close the reader. | ||
| // Note that, we read from UFS with a smaller buffer to avoid high pressure on heap | ||
| // memory when concurrent async requests are received and thus trigger GC. | ||
| long offset = 0; | ||
| while (offset < blockSize) { | ||
| long bufferSize = Math.min(8L * Constants.MB, blockSize - offset); | ||
| reader.read(offset, bufferSize); | ||
| offset += bufferSize; | ||
| } | ||
| reader.commit(); | ||
| } catch (Exception e) { | ||
| reader.abort(); | ||
| if (e instanceof IOException) { | ||
| throw (IOException) e; | ||
| } | ||
| throw new RuntimeException(e); | ||
| } |
There was a problem hiding this comment.
Readability: don't need another try but we can just add catch clause to the original try (BlockReader reader = mBlockWorker.createUfsBlockReader(.
try (BlockReader reader = mBlockWorker.createUfsBlockReader(
Sessions.CACHE_UFS_SESSION_ID, blockId, 0, false, openUfsBlockOptions)) {
// keep the same
} catch (Exception e) {
// handle exception
}There was a problem hiding this comment.
Hmmm, I had tried that actually, but I cannot call reader.abort() inside the catch clause because the it's out the scope of reader's lifecycle.
| * Compute with some io functions. | ||
| * @throws IOException | ||
| */ | ||
| void call() throws IOException; |
There was a problem hiding this comment.
it looks throwing IOE here makes the catch clause tricky.
Actually we are not really handling the IOE when abort fails, e.g. in following code like
catch (Exception e) {
reader.abort();
if (e instanceof IOException) {
throw (IOException) e;
}
throw new RuntimeException(e);
}
}For simplicity, how about just use Runnable interface (without being able to throw IOE)?
There was a problem hiding this comment.
I had tried Runnable, the abort and commit method will wrap the IOE into RuntimeException, which might break the current exception handling chain. I'm not sure how the callers of the method I modified are handling the IOE. To be safe, I still prefer to keep current logic, if it's an IOE, I still throw an IOE.
| private void abortUfsBlock(long sessionId, long blockId) throws IOException { | ||
| mUnderFileSystemBlockStore.closeBlock(sessionId, blockId); | ||
| Optional<TempBlockMeta> tempBlockMeta = mLocalBlockStore.getTempBlockMeta(blockId); | ||
| if (tempBlockMeta.isPresent() && tempBlockMeta.get().getSessionId() == sessionId) { |
There was a problem hiding this comment.
shall we have a log warn if this condition does not meet?
There was a problem hiding this comment.
good call, just added
beinan
force-pushed
the
fix_data_corruption
branch
from
8a89d03
to
37a1892
Compare
| abortBlock(sessionId, blockId); | ||
| } | ||
| } else { | ||
| if (!tempBlockMeta.isPresent() || tempBlockMeta.get().getSessionId() != sessionId) { |
There was a problem hiding this comment.
nit: we can merge the if conditions
| } | ||
| } | ||
|
|
||
| private void commitUfsBlock(long sessionId, long blockId) throws IOException { |
There was a problem hiding this comment.
nit: naming seems a bit weird since we are commiting Alluxio blocks
|
replaced by #18525 |
What changes are proposed in this pull request?
A this fix a bug in corrupted data files.
Previously, #17497 attempt to solve this issue but only covers the case when creating a UFS reader
Why are the changes needed?
Please clarify why the changes are needed. For instance,
Does this PR introduce any user facing changes?
Please list the user-facing changes introduced by your change, including