public boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException {
    meta = (PGBulkLoaderMeta) smi;
    data = (PGBulkLoaderData) sdi;

    try {
      Object[] r = getRow(); // Get row from input rowset & set row busy!

      if ( r == null ) { // no more input to be expected...

        setOutputDone();

        // Close the output stream...
        // will be null if no records (empty stream)
        if ( data != null && pgCopyOut != null ) {
          pgCopyOut.flush();
          pgCopyOut.endCopy();

        }

        return false;
      }

      if ( first ) {
        first = false;

        // Cache field indexes.
        //
        data.keynrs = new int[meta.getFieldStream().length];
        for ( int i = 0; i < data.keynrs.length; i++ ) {
          data.keynrs[i] = getInputRowMeta().indexOfValue( meta.getFieldStream()[i] );
        }

        // execute the copy statement... pgCopyOut is setup there
        //
        do_copy( meta, true );


        // Write rows of data hereafter...
        //
      }

      writeRowToPostgres( getInputRowMeta(), r );

      putRow( getInputRowMeta(), r );
      incrementLinesOutput();

      return true;
    } catch ( Exception e ) {
      logError( BaseMessages.getString( PKG, "GPBulkLoader.Log.ErrorInStep" ), e );
      setErrors( 1 );
      stopAll();
      setOutputDone(); // signal end to receiver(s)
      return false;
    }
  }
@Test
  public void testProcessRow_StreamIsNull() throws Exception {
    PGBulkLoader pgBulkLoaderStreamIsNull = mock( PGBulkLoader.class );
    doReturn( null ).when( pgBulkLoaderStreamIsNull ).getRow();
    PGBulkLoaderMeta meta = mock( PGBulkLoaderMeta.class );
    PGBulkLoaderData data = mock( PGBulkLoaderData.class );
    assertEquals( false, pgBulkLoaderStreamIsNull.processRow( meta, data ) );
  }