[ https://issues.apache.org/jira/browse/SPARK-17573?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15498958#comment-15498958 ]
Jianfei Wang commented on SPARK-17573: -------------------------------------- if some exceptions happen the fileStream may never be closed {code} private def nextBatchStream(): DeserializationStream = { // Note that batchOffsets.length = numBatches + 1 since we did a scan above; check whether // we're still in a valid batch. if (batchIndex < batchOffsets.length - 1) { if (deserializeStream != null) { deserializeStream.close() fileStream.close() deserializeStream = null fileStream = null } val start = batchOffsets(batchIndex) fileStream = new FileInputStream(file) fileStream.getChannel.position(start) batchIndex += 1 val end = batchOffsets(batchIndex) assert(end >= start, "start = " + start + ", end = " + end + ", batchOffsets = " + batchOffsets.mkString("[", ", ", "]")) val bufferedStream = new BufferedInputStream(ByteStreams.limit(fileStream, end - start)) val compressedStream = serializerManager.wrapForCompression(blockId, bufferedStream) ser.deserializeStream(compressedStream) } else { // No more batches left cleanup() null } } {code} > The FileInputStream may be uncloseed when some exceptions occurs > ---------------------------------------------------------------- > > Key: SPARK-17573 > URL: https://issues.apache.org/jira/browse/SPARK-17573 > Project: Spark > Issue Type: Improvement > Components: SQL > Affects Versions: 2.0.0 > Reporter: Jianfei Wang > Priority: Trivial > Labels: performance > > I think that the InputStream may never be closed when some exceptions occur, > we should surround this with try catch. > {code} > private def addFilesToZipStream(parent: String, source: File, output: > ZipOutputStream): Unit = { > if (source.isDirectory()) { > output.putNextEntry(new ZipEntry(parent + source.getName())) > for (file <- source.listFiles()) { > addFilesToZipStream(parent + source.getName() + File.separator, file, > output) > } > } else { > val in = new FileInputStream(source) > output.putNextEntry(new ZipEntry(parent + source.getName())) > val buf = new Array[Byte](8192) > var n = 0 > while (n != -1) { > n = in.read(buf) > if (n != -1) { > output.write(buf, 0, n) > } > } > output.closeEntry() > in.close() > } > } > some code in TestUtils.scala > val in = new FileInputStream(file) > ByteStreams.copy(in, jarStream) > in.close() > some code in IvyTestUtils.scala > val in = new FileInputStream(file._2) > ByteStreams.copy(in, jarStream) > in.close() > {code} -- This message was sent by Atlassian JIRA (v6.3.4#6332) --------------------------------------------------------------------- To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org For additional commands, e-mail: issues-h...@spark.apache.org