Github user NicoK commented on a diff in the pull request:
https://github.com/apache/flink/pull/4238#discussion_r125430786
--- Diff:
flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobServerConnection.java
---
@@ -477,97 +406,96 @@ else if (contentAddressable == CONTENT_ADDRESSABLE) {
clientSocket.close();
}
finally {
- if (fos != null) {
- try {
- fos.close();
- } catch (Throwable t) {
- LOG.warn("Cannot close stream to BLOB
staging file", t);
- }
- }
if (incomingFile != null) {
- if (!incomingFile.delete()) {
+ if (!incomingFile.delete() &&
incomingFile.exists()) {
LOG.warn("Cannot delete BLOB server
staging file " + incomingFile.getAbsolutePath());
}
}
}
}
/**
- * Handles an incoming DELETE request from a BLOB client.
- *
- * @param inputStream The input stream to read the request from.
- * @param outputStream The output stream to write the response to.
- * @throws java.io.IOException Thrown if an I/O error occurs while
reading the request data from the input stream.
+ * Reads a full file from <tt>inputStream</tt> into
<tt>incomingFile</tt> returning its checksum.
+ *
+ * @param inputStream
+ * stream to read from
+ * @param incomingFile
+ * file to write to
+ * @param buf
+ * An auxiliary buffer for data
serialization/deserialization
+ *
+ * @return the received file's content hash as a BLOB key
+ *
+ * @throws IOException
+ * thrown if an I/O error occurs while reading/writing
data from/to the respective streams
*/
- private void delete(InputStream inputStream, OutputStream outputStream,
byte[] buf) throws IOException {
+ private static BlobKey readFileFully(
+ final InputStream inputStream, final File incomingFile,
final byte[] buf)
+ throws IOException {
+ MessageDigest md = BlobUtils.createMessageDigest();
+ FileOutputStream fos = new FileOutputStream(incomingFile);
try {
- int type = inputStream.read();
- if (type < 0) {
- throw new EOFException("Premature end of DELETE
request");
- }
-
- if (type == CONTENT_ADDRESSABLE) {
- BlobKey key =
BlobKey.readFromInputStream(inputStream);
- File blobFile =
blobServer.getStorageLocation(key);
-
- writeLock.lock();
-
- try {
- // we should make the local and remote
file deletion atomic, otherwise we might risk not
- // removing the remote file in case of
a concurrent put operation
- if (blobFile.exists() &&
!blobFile.delete()) {
- throw new IOException("Cannot
delete BLOB file " + blobFile.getAbsolutePath());
- }
-
- blobStore.delete(key);
- } finally {
- writeLock.unlock();
+ while (true) {
+ final int bytesExpected =
readLength(inputStream);
+ if (bytesExpected == -1) {
+ // done
+ break;
+ }
+ if (bytesExpected > BUFFER_SIZE) {
+ throw new IOException(
+ "Unexpected number of incoming
bytes: " + bytesExpected);
}
- }
- else if (type == NAME_ADDRESSABLE) {
- byte[] jidBytes = new byte[JobID.SIZE];
- readFully(inputStream, jidBytes, 0, JobID.SIZE,
"JobID");
- JobID jobID = JobID.fromByteArray(jidBytes);
- String key = readKey(buf, inputStream);
+ readFully(inputStream, buf, 0, bytesExpected,
"buffer");
+ fos.write(buf, 0, bytesExpected);
- File blobFile =
this.blobServer.getStorageLocation(jobID, key);
+ md.update(buf, 0, bytesExpected);
+ }
+ return new BlobKey(md.digest());
+ } finally {
--- End diff --
...and throw an exception if closing the stream fails? (the current and
previous take on this was to swallow the exception and log the error message)
How grave would you consider the failure of closing an output stream to a
temporary file which was completely written just fine?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
