#780 Ensure streams are closed after index generation.

Author: yruslan

cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/processor/CobolProcessor.scala

@@ -50,12 +50,13 @@ object CobolProcessor {
     private val caseInsensitiveOptions = new mutable.HashMap[String, String]()
 
     def build(): CobolProcessor = {
+      val readerParameters = getReaderParameters
+      val cobolSchema = getCobolSchema(readerParameters)
+
       new CobolProcessor {
         override def process(inputStream: SimpleStream,
                              outputStream: OutputStream)
                             (rawRecordProcessor: RawRecordProcessor): Unit = {
-          val readerParameters = getReaderParameters
-          val cobolSchema = getCobolSchema(readerParameters)
           val recordExtractor = getRecordExtractor(readerParameters, inputStream)
 
           val dataStream = inputStream.copyStream()

spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/index/IndexBuilder.scala

@@ -163,7 +163,12 @@ private[source] object IndexBuilder extends Logging {
     logger.info(s"Going to generate index for the file: $filePath")
 
     val (inputStream, headerStream, maximumBytes) = getStreams(filePath, startOffset, endOffset, config)
-    val index = reader.generateIndex(inputStream, headerStream, fileOrder, reader.isRdwBigEndian)
+    val index = try {
+      reader.generateIndex(inputStream, headerStream, fileOrder, reader.isRdwBigEndian)
+    } finally {
+      inputStream.close()
+      headerStream.close()
+    }
 
     val indexWithEndOffset = if (maximumBytes > 0 ){
       index.map(entry => if (entry.offsetTo == -1) entry.copy(offsetTo = startOffset + maximumBytes) else entry)