public boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException {

    meta = (WriteToLogMeta) smi;
    data = (WriteToLogData) sdi;

    Object[] r = getRow(); // get row, set busy!
    if ( r == null ) { // no more input to be expected...

      setOutputDone();
      return false;
    }

    // Limit hit? skip
    if ( rowCounterLimitHit ) {
      putRow( getInputRowMeta(), r ); // copy row to output
      return true;
    }

    if ( first ) {
      first = false;

      if ( meta.getFieldName() != null && meta.getFieldName().length > 0 ) {
        data.fieldnrs = new int[meta.getFieldName().length];

        for ( int i = 0; i < data.fieldnrs.length; i++ ) {
          data.fieldnrs[i] = getInputRowMeta().indexOfValue( meta.getFieldName()[i] );
          if ( data.fieldnrs[i] < 0 ) {
            logError( BaseMessages.getString( PKG, "WriteToLog.Log.CanNotFindField", meta.getFieldName()[i] ) );
            throw new KettleException( BaseMessages.getString( PKG, "WriteToLog.Log.CanNotFindField", meta
              .getFieldName()[i] ) );
          }
        }
      } else {
        data.fieldnrs = new int[getInputRowMeta().size()];
        for ( int i = 0; i < data.fieldnrs.length; i++ ) {
          data.fieldnrs[i] = i;
        }
      }
      data.fieldnr = data.fieldnrs.length;
      data.loglevel = meta.getLogLevelByDesc();
      data.logmessage = Const.NVL( this.environmentSubstitute( meta.getLogMessage() ), "" );
      if ( !Utils.isEmpty( data.logmessage ) ) {
        data.logmessage += Const.CR + Const.CR;
      }
    } // end if first

    // We don't need to calculate if step log level is lower than the run log level
    if ( getLogLevel().getLevel() >= data.loglevel.getLevel() ) {
      StringBuilder out = new StringBuilder();
      out.append( Const.CR ).append( "------------> " )
        .append( BaseMessages.getString(
          PKG, "WriteToLog.Log.NLigne", "" + getLinesRead() ) )
        .append( "------------------------------" )
        .append( Const.CR );

      out.append( getRealLogMessage() );

      String[] fieldNames = {};

      // Obtaining the field name list is a heavy operation, as so, it was removed from the loop.
      // And, as it's only needed if the header is to be printed, I conditioned the calculation to that scenario.
      if ( meta.isDisplayHeader() ) {
        fieldNames = getInputRowMeta().getFieldNames();
      }

      // Loop through fields
      for ( int i = 0; i < data.fieldnr; i++ ) {
        String fieldValue = getInputRowMeta().getString( r, data.fieldnrs[ i ] );

        if ( meta.isDisplayHeader() ) {
          out.append( fieldNames[ data.fieldnrs[ i ] ] ).append( " = " );
        }

        out.append( fieldValue ).append( Const.CR );
      }

      out.append( Const.CR ).append( "====================" );

      setLog( data.loglevel, out );
    }

    // Increment counter
    if ( meta.isLimitRows() && ++rowCounter >= meta.getLimitRowsNumber() ) {
      rowCounterLimitHit = true;
    }

    putRow( getInputRowMeta(), r ); // copy row to output

    return true;
  }
@Test
  public void processRow_NullRow() throws Exception {
    WriteToLog writeToLogSpy = spy( writeToLog );
    doReturn( null ).when( writeToLogSpy ).getRow();
    WriteToLogMeta meta = mock( WriteToLogMeta.class );
    WriteToLogData data = mock( WriteToLogData.class );

    assertFalse( writeToLogSpy.processRow( meta, data ) );

    verify( writeToLogSpy, times( 0 ) ).getInputRowMeta();
    verify( writeToLogSpy, times( 0 ) ).getLogLevel();
  }