HADOOP-17657: implement StreamCapabilities in SequenceFile.Writer and fall back to flush, if hflush is not supported #2949
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
added 3 commits
April 23, 2021 14:04
…ther they want to use hflush or flush
Contributor
|
LGTM. Is there an easy way to test this? |
Contributor
|
checkstyle |
added 2 commits
April 26, 2021 11:20
steveloughran
requested changes
Apr 27, 2021
Contributor
steveloughran
left a comment
steveloughran
left a comment
There was a problem hiding this comment.
thanks for the test, I actually think you've done it too well and that mockito can be avoided. This is important as mockito tests are painfully brittle and a large source of false failures
| } | ||
|
|
||
| @Test | ||
| public void testSequenceFileWriter() throws Exception { |
Contributor
There was a problem hiding this comment.
I see what you are trying to do here and think it's actually overkill. And as mockito backporting is an utter nightmare (believe me), I don't want mockito code unless need be.
how about just creating a writer on the localfs (no mocking), verify that you can verify that the stream supports "HSYNC"; call hflush and hsync on it.
Now, a full end to end test would be to write data, flush it and verify that the file length is now > 0.
writer.write()
writer.hflush();
writer.hsync();
Assertions.assertThat(fs.getFileStatus(p).getLen()).isGreaterThan(0);
that should be enough. Probe passthrough and the write goes through.
Contributor
Author
There was a problem hiding this comment.
@steveloughran Thank you for the feedback. I have addressed all your concerns. Please take a look.
| writer.hflush(); | ||
| Assert.assertFalse(writer.hasCapability(StreamCapabilities.HFLUSH)); | ||
|
|
||
| writer.close(); |
Contributor
There was a problem hiding this comment.
better to use try-with-resources so even when an assertion is thrown the writer is closed.
| return new CompressionOption(value, codec); | ||
| } | ||
|
|
||
| @org.apache.hadoop.thirdparty.com.google.common.annotations.VisibleForTesting |
Contributor
There was a problem hiding this comment.
I'm proposing elsewhere to not use mockito, which avoids exposing this operation...I'd be worried about exposing something into a broadly used API, as inevitably someone will use it in production
added 2 commits
April 28, 2021 12:01
|
💔 -1 overall
This message was automatically generated. |
Contributor
|
I do like the new test, now there's just the little detail that it's not quite working yet
|
steveloughran
approved these changes
Apr 29, 2021
Contributor
steveloughran
left a comment
steveloughran
left a comment
There was a problem hiding this comment.
LGTM
+1 pending Yetus being happy.
|
💔 -1 overall
This message was automatically generated. |
|
💔 -1 overall
This message was automatically generated. |
|
💔 -1 overall
This message was automatically generated. |
asfgit
pushed a commit
that referenced
this pull request
May 4, 2021
… fall back to flush, if hflush is not supported (#2949) Co-authored-by: Kishen Das <[email protected]> Reviewed-by: Steve Loughran <[email protected]> (cherry picked from commit e571025)
kiran-maturi
pushed a commit
to kiran-maturi/hadoop
that referenced
this pull request
Nov 24, 2021
… fall back to flush, if hflush is not supported (apache#2949) Co-authored-by: Kishen Das <[email protected]> Reviewed-by: Steve Loughran <[email protected]>
jojochuang
pushed a commit
to jojochuang/hadoop
that referenced
this pull request
May 23, 2023
…e.Writer and fall back to flush, if hflush is not supported (apache#2949) Contributed by: Kishen Das <[email protected]> (cherry picked from commit 14a9fa0d48de210c0d40e40e35ac912a2619e038) Signed-off-by: Arpit Agarwal <[email protected]> Change-Id: I310ff5f4be9673b8970a89eddf99765d33ec752c
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Following exception is thrown whenever we invoke ProtoMessageWriter.hflush on S3 from Tez, which internally calls org.apache.hadoop.io.SequenceFile$Writer.hflush -> org.apache.hadoop.fs.FS DataOutputStream.hflush -> S3ABlockOutputStream.hflush which is not implemented and throws java.lang.UnsupportedOperationException.
bdffe22d96ae [mdc@18060 class="yarn.YarnUncaughtExceptionHandler" level="ERROR" thread="HistoryEventHandlingThread"] Thread Thread[HistoryEventHandlingThread, 5,main] threw an Exception.^Mjava.lang.UnsupportedOperationException: S3A streams are not Syncable^M at org.apache.hadoop.fs.s3a.S3ABlockOutputStream.hflush(S3ABlockOutputStream.java:657)^M at org.apache.hadoop.fs.FS DataOutputStream.hflush(FSDataOutputStream.java:136)^M at org.apache.hadoop.io.SequenceFile$Writer.hflush(SequenceFile.java:1367)^M at org.apache.tez.dag.history.logging.proto.ProtoMessageWriter.hflush(ProtoMessageWr iter.java:64)^M at org.apache.tez.dag.history.logging.proto.ProtoHistoryLoggingService.finishCurrentDag(ProtoHistoryLoggingService.java:239)^M at org.apache.tez.dag.history.logging.proto.ProtoHistoryLoggingService.han dleEvent(ProtoHistoryLoggingService.java:198)^M at org.apache.tez.dag.history.logging.proto.ProtoHistoryLoggingService.loop(ProtoHistoryLoggingService.java:153)^M at java.lang.Thread.run(Thread.java:748)^M
In order to fix this issue we should implement StreamCapabilities in SequenceFile.Writer. Also, we should fall back to flush(), if hflush() is not supported.