GH-3205: Make HadoopPositionOutputStream.close() safe to call even if closed

[dominicso]

Rationale for this change

HadoopPositionOutputStream implements Closable, so its close method should have no effect if already closed.

What changes are included in this PR?

Add code to HadoopPositionOutputStream.close() to check if already closed. If yes, do nothing.

Are these changes tested?

Expecting test runner to do this.

Are there any user-facing changes?

No

Closes#3205

[wgtmac]

If #3204 is merged, do we still need this fix?

cc @steveloughran

[dominicso]

While I believe if #3204 is merged, we likely won't see the problem I reported #3205, I think it's still good to have this fix because it will prevent a 2nd call to HadoopPositionOutputStream.close() from calling flush() on a closed OutputStream.

[steveloughran]

hadoop streams are meant to have close() idempotent, it is hflush() which is failing here as that is the strict call. Even with a switch to flush() things may have problems.

I think this needs a test, even just for local file output, calling close() twice, ideally on separate threads.

Is there one?

[steveloughran]

safest to use an atomic bool here

if (closed.getAndSet(true) { return;
}

streams are nominally not thread safe, but cross stream use happens. Making close() itself thread safe is low cost and avoids problems on cleanup

[dominicso]

Changed to use AtomicBoolean

[steveloughran]

just looked at flush() implementations

s3a: tells you off

 @Override
public synchronized void flush() throws IOException {
try {
checkOpen();
} catch (ClosedIOException e) {
LOG.warn("Stream closed: {}", e.getMessage());
return;
}
S3ADataBlocks.DataBlock dataBlock = getActiveBlock();
if (dataBlock != null) {
dataBlock.flush();
}
}

abfs: probably raises IllegalStateException
file: whatever java.io.FileOutputStream does
hdfs: does seem to queue stuff

overall then: dangerous to call flush() on a closed stream. Wrapping it is safer

[wgtmac]

Thanks @dominicso@steveloughran! #3204 is merged. Let's rebase this PR.

[dominicso]

Rebased and changed to use AtomicBoolean.

I think this needs a test, even just for local file output, calling close() twice, ideally on separate threads.

@steveloughran If a test is added to use local file output, wouldn't it use LocalPositionOutputStream instead of HadoopPositionOutputStream? Perhaps a unit test should be added just to call HadoopPositionOutputStream.close()?

Looking at LocalPositionOutputStream.close(), it looks like it also needs similar changes to make repeat calls to its close safe. Let me know if this should be done as part of this PR as well.

[wgtmac]

+1

[steveloughran]

LGTM

[wgtmac]

Merged. Thanks @dominicso and @steveloughran!