@Test public void testProcessRow() throws Exception { boolean result; int rowsProcessed = 0; ArgumentCaptor<RowMeta> rowMetaCaptor = ArgumentCaptor.forClass( RowMeta.class ); ArgumentCaptor<Object[]> dataCaptor = ArgumentCaptor.forClass( Object[].class ); do { result = parquetOutput.processRow( parquetOutputMeta, parquetOutputData ); if ( result ) { rowsProcessed++; } } while ( result ); assertEquals( 3, rowsProcessed ); verify( mockRowHandler, times( 3 ) ).putRow( rowMetaCaptor.capture(), dataCaptor.capture() ); List<RowMeta> rowMetaCaptured = rowMetaCaptor.getAllValues(); List<Object[]> dataCaptured = dataCaptor.getAllValues(); for ( int rowNum = 0; rowNum < 3; rowNum++ ) { assertEquals( 0, rowMetaCaptured.get( rowNum ).indexOfValue( "StringName" ) ); assertEquals( "string" + ( rowNum % 3 + 1 ), dataCaptured.get( rowNum )[ 0 ] ); } }
@Override public synchronized boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException { try { if ( data.output == null ) { init( getInputRowMeta() ); } Object[] currentRow = getRow(); if ( currentRow != null ) { RowMetaAndData row = new RowMetaAndData( getInputRowMeta(), currentRow ); data.writer.write( row ); putRow( row.getRowMeta(), row.getData() ); return true; } else { closeWriter(); pvfsFileAliaser.copyFileToFinalDestination(); pvfsFileAliaser.deleteTempFileAndFolder(); setOutputDone(); return false; } } catch ( KettleException ex ) { try { closeWriter(); pvfsFileAliaser.deleteTempFileAndFolder(); } catch ( Exception ex2 ) { } throw ex; } catch ( IllegalStateException e ) { getLogChannel().logError( e.getMessage() ); setErrors( 1 ); pvfsFileAliaser.deleteTempFileAndFolder(); setOutputDone(); return false; } catch ( Exception ex ) { try { closeWriter(); pvfsFileAliaser.deleteTempFileAndFolder(); } catch ( Exception ex2 ) { } throw new KettleException( ex ); } }
ParquetOutput extends BaseStep implements StepInterface { @Override public synchronized boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException { try { if ( data.output == null ) { init( getInputRowMeta() ); } Object[] currentRow = getRow(); if ( currentRow != null ) { RowMetaAndData row = new RowMetaAndData( getInputRowMeta(), currentRow ); data.writer.write( row ); putRow( row.getRowMeta(), row.getData() ); return true; } else { closeWriter(); pvfsFileAliaser.copyFileToFinalDestination(); pvfsFileAliaser.deleteTempFileAndFolder(); setOutputDone(); return false; } } catch ( KettleException ex ) { try { closeWriter(); pvfsFileAliaser.deleteTempFileAndFolder(); } catch ( Exception ex2 ) { } throw ex; } catch ( IllegalStateException e ) { getLogChannel().logError( e.getMessage() ); setErrors( 1 ); pvfsFileAliaser.deleteTempFileAndFolder(); setOutputDone(); return false; } catch ( Exception ex ) { try { closeWriter(); pvfsFileAliaser.deleteTempFileAndFolder(); } catch ( Exception ex2 ) { } throw new KettleException( ex ); } } }
ParquetOutput extends BaseStep implements StepInterface { @Override public synchronized boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException { try { if ( data.output == null ) { init( getInputRowMeta() ); } Object[] currentRow = getRow(); if ( currentRow != null ) { RowMetaAndData row = new RowMetaAndData( getInputRowMeta(), currentRow ); data.writer.write( row ); putRow( row.getRowMeta(), row.getData() ); return true; } else { closeWriter(); pvfsFileAliaser.copyFileToFinalDestination(); pvfsFileAliaser.deleteTempFileAndFolder(); setOutputDone(); return false; } } catch ( KettleException ex ) { try { closeWriter(); pvfsFileAliaser.deleteTempFileAndFolder(); } catch ( Exception ex2 ) { } throw ex; } catch ( IllegalStateException e ) { getLogChannel().logError( e.getMessage() ); setErrors( 1 ); pvfsFileAliaser.deleteTempFileAndFolder(); setOutputDone(); return false; } catch ( Exception ex ) { try { closeWriter(); pvfsFileAliaser.deleteTempFileAndFolder(); } catch ( Exception ex2 ) { } throw new KettleException( ex ); } } ParquetOutput( StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta,
                        Trans trans ); }
ParquetOutput extends BaseStep implements StepInterface { @Override public synchronized boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException { try { if ( data.output == null ) { init( getInputRowMeta() ); } Object[] currentRow = getRow(); if ( currentRow != null ) { RowMetaAndData row = new RowMetaAndData( getInputRowMeta(), currentRow ); data.writer.write( row ); putRow( row.getRowMeta(), row.getData() ); return true; } else { closeWriter(); pvfsFileAliaser.copyFileToFinalDestination(); pvfsFileAliaser.deleteTempFileAndFolder(); setOutputDone(); return false; } } catch ( KettleException ex ) { try { closeWriter(); pvfsFileAliaser.deleteTempFileAndFolder(); } catch ( Exception ex2 ) { } throw ex; } catch ( IllegalStateException e ) { getLogChannel().logError( e.getMessage() ); setErrors( 1 ); pvfsFileAliaser.deleteTempFileAndFolder(); setOutputDone(); return false; } catch ( Exception ex ) { try { closeWriter(); pvfsFileAliaser.deleteTempFileAndFolder(); } catch ( Exception ex2 ) { } throw new KettleException( ex ); } } ParquetOutput( StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta,
                        Trans trans ); @Override synchronized boolean processRow( StepMetaInterface smi, StepDataInterface sdi ); void init( RowMetaInterface rowMeta ); void closeWriter(); boolean init( StepMetaInterface smi, StepDataInterface sdi ); }
ParquetOutput extends BaseStep implements StepInterface { @Override public synchronized boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException { try { if ( data.output == null ) { init( getInputRowMeta() ); } Object[] currentRow = getRow(); if ( currentRow != null ) { RowMetaAndData row = new RowMetaAndData( getInputRowMeta(), currentRow ); data.writer.write( row ); putRow( row.getRowMeta(), row.getData() ); return true; } else { closeWriter(); pvfsFileAliaser.copyFileToFinalDestination(); pvfsFileAliaser.deleteTempFileAndFolder(); setOutputDone(); return false; } } catch ( KettleException ex ) { try { closeWriter(); pvfsFileAliaser.deleteTempFileAndFolder(); } catch ( Exception ex2 ) { } throw ex; } catch ( IllegalStateException e ) { getLogChannel().logError( e.getMessage() ); setErrors( 1 ); pvfsFileAliaser.deleteTempFileAndFolder(); setOutputDone(); return false; } catch ( Exception ex ) { try { closeWriter(); pvfsFileAliaser.deleteTempFileAndFolder(); } catch ( Exception ex2 ) { } throw new KettleException( ex ); } } ParquetOutput( StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta,
                        Trans trans ); @Override synchronized boolean processRow( StepMetaInterface smi, StepDataInterface sdi ); void init( RowMetaInterface rowMeta ); void closeWriter(); boolean init( StepMetaInterface smi, StepDataInterface sdi );  }