[PDI-13759] excel input step when using apache POI streaming does not…

… release file xlsx in case of error

build.xml

@@ -758,6 +758,7 @@
             haltonerror="false"/>
       <test todir="${testreports.xml.dir}" name="org.pentaho.di.trans.steps.excelinput.PoiWorkBookTest"
             haltonerror="false"/>
+      <test todir="${testreports.xml.dir}" name="org.pentaho.di.trans.steps.excelinput.StaxWorkBookTest" haltonerror="false"/>
       <test todir="${testreports.xml.dir}" name="org.pentaho.di.trans.steps.execsqlrow.ExecSQLRowTest"
             haltonerror="false"/>
       <test todir="${testreports.xml.dir}" name="org.pentaho.di.trans.steps.getxmldata.GetXMLDataTest"

engine/src/org/pentaho/di/trans/steps/excelinput/poi/PoiWorkbook.java

@@ -33,12 +33,16 @@
 import org.apache.poi.ss.usermodel.Sheet;
 import org.apache.poi.ss.usermodel.Workbook;
 import org.pentaho.di.core.exception.KettleException;
+import org.pentaho.di.core.logging.KettleLogStore;
+import org.pentaho.di.core.logging.LogChannelInterface;
 import org.pentaho.di.core.spreadsheet.KSheet;
 import org.pentaho.di.core.spreadsheet.KWorkbook;
 import org.pentaho.di.core.vfs.KettleVFS;
 
 public class PoiWorkbook implements KWorkbook {
 
+  private LogChannelInterface log;
+
   private Workbook workbook;
   private String filename;
   private String encoding;
@@ -50,7 +54,7 @@ public class PoiWorkbook implements KWorkbook {
   public PoiWorkbook( String filename, String encoding ) throws KettleException {
     this.filename = filename;
     this.encoding = encoding;
-
+    this.log = KettleLogStore.getLogChannelInterfaceFactory().create( this );
     try {
       FileObject fileObject = KettleVFS.getFileObject( filename );
       if ( fileObject instanceof LocalFile ) {
@@ -101,7 +105,7 @@ public void close() {
         opcpkg.revert();
       }
     } catch ( IOException ex ) {
-      // Ignore errors
+      log.logError( "Could not close workbook", ex );
     }
   }
 

engine/src/org/pentaho/di/trans/steps/excelinput/staxpoi/StaxPoiSheet.java

@@ -112,9 +112,6 @@ public StaxPoiSheet( XSSFReader reader, String sheetName, String sheetID ) {
 
   @Override
   public KCell[] getRow( int rownr ) {
-
-    // convert 0 based index to 1 based
-    rownr += 1;
     try {
       while ( sheetReader.hasNext() ) {
         int event = sheetReader.next();
@@ -126,14 +123,23 @@ public KCell[] getRow( int rownr ) {
           }
 
           KCell[] cells = new StaxPoiCell[numCols];
+          boolean richedRowEnd = false;
           for ( int i = 0; i < numCols; i++ ) {
             // go to the "c" <cell> tag
             while ( sheetReader.hasNext() ) {
               if ( event == XMLStreamConstants.START_ELEMENT && sheetReader.getLocalName().equals( "c" ) ) {
                 break;
               }
+              //if we have empty cell than we could reach and of row before fill all cells, so break all cycle
+              if ( event == XMLStreamConstants.END_ELEMENT && sheetReader.getLocalName().equals( "row" ) ) {
+                richedRowEnd = true;
+                break;
+              }
               event = sheetReader.next();
             }
+            if ( richedRowEnd ) {
+              break;
+            }
             String cellLocation = sheetReader.getAttributeValue( null, "r" );
             int columnIndex = StaxUtil.extractColumnNumber( cellLocation ) - 1;
             String cellType = sheetReader.getAttributeValue( null, "t" );
@@ -191,7 +197,7 @@ public KCell getCell( int colnr, int rownr ) {
   }
 
   public void close() throws IOException, XMLStreamException {
-    sheetStream.close();
     sheetReader.close();
+    sheetStream.close();
   }
 }

engine/src/org/pentaho/di/trans/steps/excelinput/staxpoi/StaxPoiWorkbook.java

@@ -26,24 +26,28 @@
 
 package org.pentaho.di.trans.steps.excelinput.staxpoi;
 
+import java.io.IOException;
 import java.io.InputStream;
-import java.util.ArrayList;
 import java.util.HashMap;
-import java.util.List;
 import java.util.Map;
 
 import javax.xml.stream.XMLInputFactory;
 import javax.xml.stream.XMLStreamConstants;
+import javax.xml.stream.XMLStreamException;
 import javax.xml.stream.XMLStreamReader;
 
 import org.apache.poi.openxml4j.opc.OPCPackage;
 import org.apache.poi.xssf.eventusermodel.XSSFReader;
 import org.pentaho.di.core.exception.KettleException;
+import org.pentaho.di.core.logging.KettleLogStore;
+import org.pentaho.di.core.logging.LogChannelInterface;
 import org.pentaho.di.core.spreadsheet.KSheet;
 import org.pentaho.di.core.spreadsheet.KWorkbook;
 
 public class StaxPoiWorkbook implements KWorkbook {
 
+  private LogChannelInterface log;
+
   private XSSFReader reader;
 
   // maintain the mapping of the sheet name to its ID
@@ -52,54 +56,66 @@ public class StaxPoiWorkbook implements KWorkbook {
   // mapping of the sheet object with its ID/Name
   private Map<String, StaxPoiSheet> openSheetsMap;
 
+  private OPCPackage opcpkg;
+
   public StaxPoiWorkbook() {
     openSheetsMap = new HashMap<String, StaxPoiSheet>();
+    this.log = KettleLogStore.getLogChannelInterfaceFactory().create( this );
   }
 
   public StaxPoiWorkbook( String filename, String encoding ) throws KettleException {
     this();
     try {
-      OPCPackage pkg = OPCPackage.open( filename );
-      openFile( pkg, encoding );
+      opcpkg = OPCPackage.open( filename );
+      openFile( opcpkg, encoding );
     } catch ( Exception e ) {
       throw new KettleException( e );
     }
-
   }
 
   public StaxPoiWorkbook( InputStream inputStream, String encoding ) throws KettleException {
     this();
     try {
-      OPCPackage pkg = OPCPackage.open( inputStream );
-      openFile( pkg, encoding );
+      opcpkg = OPCPackage.open( inputStream );
+      openFile( opcpkg, encoding );
     } catch ( Exception e ) {
       throw new KettleException( e );
     }
   }
 
   private void openFile( OPCPackage pkg, String encoding ) throws KettleException {
+    InputStream workbookData = null;
+    XMLStreamReader workbookReader = null;
     try {
       reader = new XSSFReader( pkg );
       sheetNameIDMap = new HashMap<String, String>();
-      List<String> sheetList = new ArrayList<String>();
-      InputStream workbookData;
       workbookData = reader.getWorkbookData();
       XMLInputFactory factory = XMLInputFactory.newInstance();
-      XMLStreamReader workbookReader = factory.createXMLStreamReader( workbookData );
+      workbookReader = factory.createXMLStreamReader( workbookData );
       while ( workbookReader.hasNext() ) {
-        if ( workbookReader.next() == XMLStreamConstants.START_ELEMENT
-          && workbookReader.getLocalName().equals( "sheet" ) ) {
+        if ( workbookReader.next() == XMLStreamConstants.START_ELEMENT && workbookReader.getLocalName().equals( "sheet" ) ) {
           String sheetName = workbookReader.getAttributeValue( null, "name" );
-          String sheetID =
-            workbookReader.getAttributeValue(
-              "http://schemas.openxmlformats.org/officeDocument/2006/relationships", "id" );
-          sheetList.add( sheetName );
+          String sheetID = workbookReader.getAttributeValue( "http://schemas.openxmlformats.org/officeDocument/2006/relationships", "id" );
           sheetNameIDMap.put( sheetName, sheetID );
         }
       }
-      workbookData.close();
     } catch ( Exception e ) {
       throw new KettleException( e );
+    } finally {
+      if ( workbookReader != null ) {
+        try {
+          workbookReader.close();
+        } catch ( XMLStreamException e ) {
+          throw new KettleException( e );
+        }
+      }
+      if ( workbookData != null ) {
+        try {
+          workbookData.close();
+        } catch ( IOException e ) {
+          throw new KettleException( e );
+        }
+      }
     }
   }
 
@@ -129,10 +145,16 @@ public void close() {
     for ( StaxPoiSheet sheet : openSheetsMap.values() ) {
       try {
         sheet.close();
-      } catch ( Exception e ) {
-        e.printStackTrace();
+      } catch ( IOException e ) {
+        log.logError( "Could not close workbook", e );
+      } catch ( XMLStreamException e ) {
+        log.logError( "Could not close xmlstream", e );
       }
     }
+    if ( opcpkg != null ) {
+      //We should not save change in xlsx because it is input step.
+      opcpkg.revert();
+    }
   }
 
   @Override

test/org/pentaho/di/trans/steps/excelinput/PoiWorkBookTest.java

@@ -2,7 +2,7 @@
  *
  * Pentaho Data Integration
  *
- * Copyright (C) 2002-2013 by Pentaho : http://www.pentaho.com
+ * Copyright (C) 2002-2015 by Pentaho : http://www.pentaho.com
  *
  *******************************************************************************
  *
@@ -22,21 +22,61 @@
 
 package org.pentaho.di.trans.steps.excelinput;
 
-import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.*;
 
 import java.io.File;
+import java.io.IOException;
+import java.io.RandomAccessFile;
+import java.nio.channels.FileChannel;
+import java.nio.channels.FileLock;
 import java.util.Date;
 
-import junit.framework.TestCase;
-
+import org.junit.Test;
+import org.pentaho.di.core.exception.KettleException;
 import org.pentaho.di.core.spreadsheet.KCell;
 import org.pentaho.di.core.spreadsheet.KCellType;
 import org.pentaho.di.core.spreadsheet.KSheet;
 import org.pentaho.di.core.spreadsheet.KWorkbook;
+import org.apache.commons.io.FileUtils;
+
+public class PoiWorkBookTest {
+
+  @Test
+  public void testReadData() throws KettleException {
+    readData();
+  }
 
-public class PoiWorkBookTest extends TestCase {
-  public void testRead() throws Exception {
+  @Test
+  public void testFileDoesNotChange() throws KettleException, IOException {
     File fileBeforeRead = new File( "testfiles/sample-file.xlsx" );
+    readData();
+    File fileAfterRead = new File( "testfiles/sample-file.xlsx" );
+    assertTrue( FileUtils.contentEquals(fileBeforeRead, fileAfterRead ) );
+  }
+
+  @Test
+  public void testResourceFree() throws Exception {
+    FileLock lock = null;
+    RandomAccessFile randomAccessFile = null;
+    try {
+      readData();
+      File fileAfterRead = new File( "testfiles/sample-file.xlsx" );
+      randomAccessFile = new RandomAccessFile( fileAfterRead, "rw" );
+      FileChannel fileChannel = randomAccessFile.getChannel();
+      lock = fileChannel.tryLock();
+      // check that we could lock file
+      assertTrue( lock.isValid() );
+    } finally {
+      if ( lock != null ) {
+        lock.release();
+      }
+      if ( randomAccessFile != null ) {
+        randomAccessFile.close();
+      }
+    }
+  }
+
+  private void readData() throws KettleException {
     KWorkbook workbook = WorkbookFactory.getWorkbook( SpreadSheetType.POI, "testfiles/sample-file.xlsx", null );
     int numberOfSheets = workbook.getNumberOfSheets();
     assertEquals( 3, numberOfSheets );
@@ -85,13 +125,10 @@ public void testRead() throws Exception {
 
     try {
       sheet1.getRow( 5 );
-      throw new Exception( "No out of bounds exception thrown when expected" );
+      fail( "No out of bounds exception thrown when expected" );
     } catch ( ArrayIndexOutOfBoundsException e ) {
       // OK!
     }
-
     workbook.close();
-    File fileAfterRead = new File( "testfiles/sample-file.xlsx" );
-    assertEquals( fileBeforeRead, fileAfterRead );
   }
 }