public boolean init( StepMetaInterface smi, StepDataInterface sdi ) {
    meta = (TableOutputMeta) smi;
    data = (TableOutputData) sdi;

    if ( super.init( smi, sdi ) ) {
      try {
        data.commitSize = Integer.parseInt( environmentSubstitute( meta.getCommitSize() ) );

        data.databaseMeta = meta.getDatabaseMeta();
        DatabaseInterface dbInterface = data.databaseMeta.getDatabaseInterface();

        // Batch updates are not supported on PostgreSQL (and look-a-likes)
        // together with error handling (PDI-366).
        // For these situations we can use savepoints to help out.
        data.useSafePoints =
          data.databaseMeta.getDatabaseInterface().useSafePoints() && getStepMeta().isDoingErrorHandling();

        // Get the boolean that indicates whether or not we can/should release
        // savepoints during data load.
        data.releaseSavepoint = dbInterface.releaseSavepoint();

        // Disable batch mode in case
        // - we use an unlimited commit size
        // - if we need to pick up auto-generated keys
        // - if you are running the transformation as a single database transaction (unique connections)
        // - if we are reverting to save-points
        data.batchMode =
          meta.useBatchUpdate()
            && data.commitSize > 0 && !meta.isReturningGeneratedKeys()
            && !getTransMeta().isUsingUniqueConnections() && !data.useSafePoints;

        // Per PDI-6211 : give a warning that batch mode operation in combination with step error handling can lead to
        // incorrectly processed rows.
        if ( getStepMeta().isDoingErrorHandling() && !dbInterface.supportsErrorHandlingOnBatchUpdates() ) {
          log.logMinimal( BaseMessages.getString(
            PKG, "TableOutput.Warning.ErrorHandlingIsNotFullySupportedWithBatchProcessing" ) );
        }

        if ( !dbInterface.supportsStandardTableOutput() ) {
          throw new KettleException( dbInterface.getUnsupportedTableOutputMessage() );
        }

        if ( log.isBasic() ) {
          logBasic( "Connected to database [" + meta.getDatabaseMeta() + "] (commit=" + data.commitSize + ")" );
        }

        // Postpone commit as long as possible. PDI-2091
        if ( data.commitSize == 0 ) {
          data.commitSize = Integer.MAX_VALUE;
        }
        data.db.setCommitSize( data.commitSize );

        if ( !meta.isPartitioningEnabled() && !meta.isTableNameInField() ) {
          data.tableName = environmentSubstitute( meta.getTableName() );
        }

        return true;
      } catch ( KettleException e ) {
        logError( "An error occurred intialising this step: " + e.getMessage() );
        stopAll();
        setErrors( 1 );
      }
    }
    return false;
  }
@Test
  public void testInit_unsupportedConnection() {

    TableOutputMeta meta =  mock( TableOutputMeta.class );
    TableOutputData data = mock( TableOutputData.class );

    DatabaseInterface dbInterface = mock( DatabaseInterface.class );

    doNothing().when( tableOutputSpy ).logError( anyString() );

    when( meta.getCommitSize() ).thenReturn( "1" );
    when( meta.getDatabaseMeta() ).thenReturn( databaseMeta );
    when( databaseMeta.getDatabaseInterface() ).thenReturn( dbInterface );

    String unsupportedTableOutputMessage = "unsupported exception";
    when( dbInterface.getUnsupportedTableOutputMessage() ).thenReturn( unsupportedTableOutputMessage );

    //Will cause the Kettle Exception
    when( dbInterface.supportsStandardTableOutput() ).thenReturn( false );

    tableOutputSpy.init( meta, data );

    KettleException ke = new KettleException( unsupportedTableOutputMessage );
    verify( tableOutputSpy, times( 1 ) ).logError( "An error occurred intialising this step: " + ke.getMessage() );
  }