jnp commented on a change in pull request #729: HDDS-1373. KeyOutputStream,
close after write request fails after retries, runs into
IllegalArgumentException
URL: https://github.com/apache/hadoop/pull/729#discussion_r275481162
##########
File path:
hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/KeyOutputStream.java
##########
@@ -295,60 +295,66 @@ private void handleWrite(byte[] b, int off, long len,
boolean retry)
throws IOException {
int succeededAllocates = 0;
while (len > 0) {
- if (streamEntries.size() <= currentStreamIndex) {
- Preconditions.checkNotNull(omClient);
- // allocate a new block, if a exception happens, log an error and
- // throw exception to the caller directly, and the write fails.
+ try {
+ if (streamEntries.size() <= currentStreamIndex) {
+ Preconditions.checkNotNull(omClient);
+ // allocate a new block, if a exception happens, log an error and
+ // throw exception to the caller directly, and the write fails.
+ try {
+ allocateNewBlock(currentStreamIndex);
+ succeededAllocates += 1;
+ } catch (IOException ioe) {
+ LOG.error("Try to allocate more blocks for write failed, already "
+ + "allocated " + succeededAllocates
+ + " blocks for this write.");
+ throw ioe;
+ }
+ }
+ // in theory, this condition should never violate due the check above
+ // still do a sanity check.
+ Preconditions.checkArgument(currentStreamIndex < streamEntries.size());
+ BlockOutputStreamEntry current = streamEntries.get(currentStreamIndex);
+
+ // length(len) will be in int range if the call is happening through
+ // write API of blockOutputStream. Length can be in long range if it
comes
+ // via Exception path.
+ int writeLen = Math.min((int) len, (int) current.getRemaining());
+ long currentPos = current.getWrittenDataLength();
try {
- allocateNewBlock(currentStreamIndex);
- succeededAllocates += 1;
+ if (retry) {
+ current.writeOnRetry(len);
+ } else {
+ current.write(b, off, writeLen);
+ offset += writeLen;
+ }
} catch (IOException ioe) {
- LOG.error("Try to allocate more blocks for write failed, already "
- + "allocated " + succeededAllocates + " blocks for this write.");
- throw ioe;
- }
- }
- // in theory, this condition should never violate due the check above
- // still do a sanity check.
- Preconditions.checkArgument(currentStreamIndex < streamEntries.size());
- BlockOutputStreamEntry current = streamEntries.get(currentStreamIndex);
-
- // length(len) will be in int range if the call is happening through
- // write API of blockOutputStream. Length can be in long range if it
comes
- // via Exception path.
- int writeLen = Math.min((int)len, (int) current.getRemaining());
- long currentPos = current.getWrittenDataLength();
- try {
- if (retry) {
- current.writeOnRetry(len);
- } else {
- current.write(b, off, writeLen);
- offset += writeLen;
+ // for the current iteration, totalDataWritten - currentPos gives the
+ // amount of data already written to the buffer
+
+ // In the retryPath, the total data to be written will always be
equal
+ // to or less than the max length of the buffer allocated.
+ // The len specified here is the combined sum of the data length of
+ // the buffers
+ Preconditions.checkState(!retry || len <= streamBufferMaxSize);
+ int dataWritten = (int) (current.getWrittenDataLength() -
currentPos);
+ writeLen = retry ? (int) len : dataWritten;
+ // In retry path, the data written is already accounted in offset.
+ if (!retry) {
+ offset += writeLen;
+ }
+ LOG.debug("writeLen {}, total len {}", writeLen, len);
+ handleException(current, currentStreamIndex, ioe);
}
- } catch (IOException ioe) {
- // for the current iteration, totalDataWritten - currentPos gives the
- // amount of data already written to the buffer
-
- // In the retryPath, the total data to be written will always be equal
- // to or less than the max length of the buffer allocated.
- // The len specified here is the combined sum of the data length of
- // the buffers
- Preconditions.checkState(!retry || len <= streamBufferMaxSize);
- int dataWritten = (int) (current.getWrittenDataLength() - currentPos);
- writeLen = retry ? (int) len : dataWritten;
- // In retry path, the data written is already accounted in offset.
- if (!retry) {
- offset += writeLen;
+ if (current.getRemaining() <= 0) {
+ // since the current block is already written close the stream.
+ handleFlushOrClose(StreamAction.FULL);
}
- LOG.debug("writeLen {}, total len {}", writeLen, len);
- handleException(current, currentStreamIndex, ioe);
- }
- if (current.getRemaining() <= 0) {
- // since the current block is already written close the stream.
- handleFlushOrClose(StreamAction.FULL);
+ len -= writeLen;
+ off += writeLen;
+ } catch (Exception e) {
+ markStreamClosed();
Review comment:
The stream is being closed only if the 'handleRetry' throws an exception. I
think, the stream should be marked closed, even without a retry, because the
retry is for the next stream.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
