AvroInput extends BaseStep { @Override public boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException { meta = (AvroInputMeta) smi; data = (AvroInputData) sdi; do { try { if ( data.input == null || data.reader == null || data.rowIterator == null ) { if ( !initializeSource() ) { break; } } if ( data.rowIterator.hasNext() ) { updateVariableSpaceWithLookupFields( getInputRowMeta() ); RowMetaAndData row = data.rowIterator.next(); putRow( row.getRowMeta(), row.getData() ); return true; } fileFinishedHousekeeping(); } catch ( KettleException ex ) { throw ex; } catch ( Exception ex ) { throw new KettleException( ex ); } } while ( ( meta.getDataLocationType() == AvroInputMetaBase.LocationDescriptor.FIELD_NAME ) ); setOutputDone(); return false; } AvroInput( StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta,
                    Trans trans ); @Override boolean processRow( StepMetaInterface smi, StepDataInterface sdi ); static List<? extends IAvroInputField> getLeafFields( NamedClusterServiceLocator namedClusterServiceLocator,
                                                               NamedCluster namedCluster, String schemaPath,
                                                               String dataPath, VariableSpace variableSpace ); void checkForLegacyFieldNames( String schemaFileName, String avroFileName ); @Override boolean init( StepMetaInterface smi, StepDataInterface sdi );  }
@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 = avroInput.processRow( avroInputMeta, sdi ); if ( result == true ) { rowsProcessed++; } } while ( result == true ); assertEquals( 4, rowsProcessed ); verify( mockRowHandler, times( 4 ) ).putRow( rowMetaCaptor.capture(), dataCaptor.capture() ); List<RowMeta> rowMeta = rowMetaCaptor.getAllValues(); List<Object[]> dataCaptured = dataCaptor.getAllValues(); for ( int rowNum = 0; rowNum < 4; rowNum++ ) { assertEquals( 0, rowMeta.get( rowNum ).indexOfValue( "str" ) ); assertEquals( "string" + ( rowNum % 2 + 1 ), dataCaptured.get( rowNum )[ 0 ] ); } }