@Test public void testCleanup() throws IOException { BaseStep baseStep = new BaseStep( mockHelper.stepMeta, mockHelper.stepDataInterface, 0, mockHelper.transMeta, mockHelper.trans ); ServerSocket serverSocketMock = mock( ServerSocket.class ); doReturn( 0 ).when( serverSocketMock ).getLocalPort(); baseStep.setServerSockets( Collections.singletonList( serverSocketMock ) ); SocketRepository socketRepositoryMock = mock( SocketRepository.class ); baseStep.setSocketRepository( socketRepositoryMock ); baseStep.cleanup(); verify( socketRepositoryMock ).releaseSocket( 0 ); }
@Override public void cleanup() { for ( ServerSocket serverSocket : serverSockets ) { try { socketRepository.releaseSocket( serverSocket.getLocalPort() ); logDetailed( BaseMessages.getString( PKG, "BaseStep.Log.ReleasedServerSocketOnPort", serverSocket.getLocalPort() ) ); } catch ( IOException e ) { logError( "Cleanup: Unable to release server socket (" + serverSocket.getLocalPort() + ")", e ); } } List<RemoteStep> remoteInputSteps = getRemoteInputSteps(); if ( remoteInputSteps != null ) { cleanupRemoteSteps( remoteInputSteps ); } List<RemoteStep> remoteOutputSteps = getRemoteOutputSteps(); if ( remoteOutputSteps != null ) { cleanupRemoteSteps( remoteOutputSteps ); } }
BaseStep implements VariableSpace, StepInterface, LoggingObjectInterface, ExtensionDataInterface { @Override public void cleanup() { for ( ServerSocket serverSocket : serverSockets ) { try { socketRepository.releaseSocket( serverSocket.getLocalPort() ); logDetailed( BaseMessages.getString( PKG, "BaseStep.Log.ReleasedServerSocketOnPort", serverSocket.getLocalPort() ) ); } catch ( IOException e ) { logError( "Cleanup: Unable to release server socket (" + serverSocket.getLocalPort() + ")", e ); } } List<RemoteStep> remoteInputSteps = getRemoteInputSteps(); if ( remoteInputSteps != null ) { cleanupRemoteSteps( remoteInputSteps ); } List<RemoteStep> remoteOutputSteps = getRemoteOutputSteps(); if ( remoteOutputSteps != null ) { cleanupRemoteSteps( remoteOutputSteps ); } } }
BaseStep implements VariableSpace, StepInterface, LoggingObjectInterface, ExtensionDataInterface { @Override public void cleanup() { for ( ServerSocket serverSocket : serverSockets ) { try { socketRepository.releaseSocket( serverSocket.getLocalPort() ); logDetailed( BaseMessages.getString( PKG, "BaseStep.Log.ReleasedServerSocketOnPort", serverSocket.getLocalPort() ) ); } catch ( IOException e ) { logError( "Cleanup: Unable to release server socket (" + serverSocket.getLocalPort() + ")", e ); } } List<RemoteStep> remoteInputSteps = getRemoteInputSteps(); if ( remoteInputSteps != null ) { cleanupRemoteSteps( remoteInputSteps ); } List<RemoteStep> remoteOutputSteps = getRemoteOutputSteps(); if ( remoteOutputSteps != null ) { cleanupRemoteSteps( remoteOutputSteps ); } } BaseStep( StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta,
                   Trans trans ); }
BaseStep implements VariableSpace, StepInterface, LoggingObjectInterface, ExtensionDataInterface { @Override public void cleanup() { for ( ServerSocket serverSocket : serverSockets ) { try { socketRepository.releaseSocket( serverSocket.getLocalPort() ); logDetailed( BaseMessages.getString( PKG, "BaseStep.Log.ReleasedServerSocketOnPort", serverSocket.getLocalPort() ) ); } catch ( IOException e ) { logError( "Cleanup: Unable to release server socket (" + serverSocket.getLocalPort() + ")", e ); } } List<RemoteStep> remoteInputSteps = getRemoteInputSteps(); if ( remoteInputSteps != null ) { cleanupRemoteSteps( remoteInputSteps ); } List<RemoteStep> remoteOutputSteps = getRemoteOutputSteps(); if ( remoteOutputSteps != null ) { cleanupRemoteSteps( remoteOutputSteps ); } } BaseStep( StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta,
                   Trans trans ); @Override boolean init( StepMetaInterface smi, StepDataInterface sdi ); @Override void dispose( StepMetaInterface smi, StepDataInterface sdi ); @Override void cleanup(); @Override long getProcessed(); void setCopy( int cop ); @Override int getCopy(); @Override long getErrors(); @Override void setErrors( long e ); @Override long getLinesRead(); long incrementLinesRead(); long decrementLinesRead(); void setLinesRead( long newLinesReadValue ); @Override long getLinesInput(); long incrementLinesInput(); void setLinesInput( long newLinesInputValue ); @Override long getLinesOutput(); long incrementLinesOutput(); void setLinesOutput( long newLinesOutputValue ); @Override long getLinesWritten(); long incrementLinesWritten(); long decrementLinesWritten(); void setLinesWritten( long newLinesWrittenValue ); @Override long getLinesUpdated(); long incrementLinesUpdated(); void setLinesUpdated( long newLinesUpdatedValue ); @Override long getLinesRejected(); long incrementLinesRejected(); @Override void setLinesRejected( long newLinesRejectedValue ); long getLinesSkipped(); long incrementLinesSkipped(); void setLinesSkipped( long newLinesSkippedValue ); @Override String getStepname(); void setStepname( String stepname ); Trans getDispatcher(); String getStatusDescription(); StepMetaInterface getStepMetaInterface(); void setStepMetaInterface( StepMetaInterface stepMetaInterface ); StepDataInterface getStepDataInterface(); void setStepDataInterface( StepDataInterface stepDataInterface ); @Override StepMeta getStepMeta(); void setStepMeta( StepMeta stepMeta ); TransMeta getTransMeta(); void setTransMeta( TransMeta transMeta ); @Override Trans getTrans(); @Override void putRow( RowMetaInterface rowMeta, Object[] row ); void putRowTo( RowMetaInterface rowMeta, Object[] row, RowSet rowSet ); void handlePutRowTo( RowMetaInterface rowMeta, Object[] row, RowSet rowSet ); void putError( RowMetaInterface rowMeta, Object[] row, long nrErrors, String errorDescriptions,
                        String fieldNames, String errorCodes ); @Override Object[] getRow(); void setRowHandler( RowHandler rowHandler ); RowHandler getRowHandler(); @Override void identifyErrorOutput(); static void safeModeChecking( RowMetaInterface referenceRowMeta, RowMetaInterface rowMeta ); Object[] getRowFrom( RowSet rowSet ); Object[] handleGetRowFrom( RowSet rowSet ); RowSet findInputRowSet( String sourceStep ); RowSet findInputRowSet( String from, int fromcopy, String to, int tocopy ); RowSet findOutputRowSet( String targetStep ); RowSet findOutputRowSet( String from, int fromcopy, String to, int tocopy ); @Override void setOutputDone(); void dispatch(); boolean isBasic(); boolean isDetailed(); boolean isDebug(); boolean isRowLevel(); void logMinimal( String message ); void logMinimal( String message, Object... arguments ); void logBasic( String message ); void logBasic( String message, Object... arguments ); void logDetailed( String message ); void logDetailed( String message, Object... arguments ); void logDebug( String message ); void logDebug( String message, Object... arguments ); void logRowlevel( String message ); void logRowlevel( String message, Object... arguments ); void logError( String message ); void logError( String message, Throwable e ); void logError( String message, Object... arguments ); int getNextClassNr(); boolean outputIsDone(); @Override void stopAll(); @Override boolean isStopped(); @Override boolean isRunning(); @Override boolean isPaused(); @Override void setStopped( boolean stopped ); @Override void setRunning( boolean running ); @Override void pauseRunning(); @Override void resumeRunning(); void setPaused( boolean paused ); void setPaused( AtomicBoolean paused ); boolean isInitialising(); @Override void markStart(); void setInternalVariables(); @Override void markStop(); @Override long getRuntime(); RowMetaAndData buildLog( String sname, int copynr, long lines_read, long lines_written,
                                  long lines_updated, long lines_skipped, long errors, Date start_date,
                                  Date end_date ); static final RowMetaInterface getLogFields( String comm ); @Override String toString(); @Override int rowsetOutputSize(); @Override int rowsetInputSize(); @Override void stopRunning( StepMetaInterface stepMetaInterface, StepDataInterface stepDataInterface ); @Deprecated void stopRunning(); void logSummary(); @Override String getStepID(); @Override List<RowSet> getInputRowSets(); void setInputRowSets( List<RowSet> inputRowSets ); @Override List<RowSet> getOutputRowSets(); void setOutputRowSets( List<RowSet> outputRowSets ); boolean isDistributed(); void setDistributed( boolean distributed ); @Override void addRowListener( RowListener rowListener ); @Override void removeRowListener( RowListener rowListener ); @Override List<RowListener> getRowListeners(); void addResultFile( ResultFile resultFile ); @Override Map<String, ResultFile> getResultFiles(); @Override StepExecutionStatus getStatus(); @Override String getPartitionID(); @Override void setPartitionID( String partitionID ); Map<String, BlockingRowSet> getPartitionTargets(); void setPartitionTargets( Map<String, BlockingRowSet> partitionTargets ); int getRepartitioning(); @Override void setRepartitioning( int repartitioning ); @Override boolean isPartitioned(); @Override void setPartitioned( boolean partitioned ); RowMetaInterface getInputRowMeta(); void setInputRowMeta( RowMetaInterface rowMeta ); RowMetaInterface getErrorRowMeta(); void setErrorRowMeta( RowMetaInterface errorRowMeta ); RowMetaInterface getPreviewRowMeta(); void setPreviewRowMeta( RowMetaInterface previewRowMeta ); @Override void copyVariablesFrom( VariableSpace space ); @Override String environmentSubstitute( String aString ); @Override String[] environmentSubstitute( String[] aString ); @Override String fieldSubstitute( String aString, RowMetaInterface rowMeta, Object[] rowData ); @Override VariableSpace getParentVariableSpace(); @Override void setParentVariableSpace( VariableSpace parent ); @Override String getVariable( String variableName, String defaultValue ); @Override String getVariable( String variableName ); @Override boolean getBooleanValueOfVariable( String variableName, boolean defaultValue ); @Override void initializeVariablesFrom( VariableSpace parent ); @Override String[] listVariables(); @Override void setVariable( String variableName, String variableValue ); @Override void shareVariablesWith( VariableSpace space ); @Override void injectVariables( Map<String, String> prop ); String getTypeId(); int getSlaveNr(); int getClusterSize(); int getUniqueStepNrAcrossSlaves(); int getUniqueStepCountAcrossSlaves(); List<ServerSocket> getServerSockets(); void setServerSockets( List<ServerSocket> serverSockets ); @Override void setUsingThreadPriorityManagment( boolean usingThreadPriorityManagment ); @Override boolean isUsingThreadPriorityManagment(); @Override void initBeforeStart(); List<StepListener> getStepListeners(); void setStepListeners( List<StepListener> stepListeners ); @Override boolean processRow( StepMetaInterface smi, StepDataInterface sdi ); @Override boolean canProcessOneRow(); @Override void addStepListener( StepListener stepListener ); @Override boolean isMapping(); SocketRepository getSocketRepository(); void setSocketRepository( SocketRepository socketRepository ); @Override String getObjectName(); @Override LogChannelInterface getLogChannel(); @Override String getFilename(); @Override String getLogChannelId(); @Override ObjectId getObjectId(); @Override ObjectRevision getObjectRevision(); @Override LoggingObjectType getObjectType(); @Override LoggingObjectInterface getParent(); @Override RepositoryDirectory getRepositoryDirectory(); @Override String getObjectCopy(); @Override LogLevel getLogLevel(); void setLogLevel( LogLevel logLevel ); static void closeQuietly( Closeable cl ); @Override String getContainerObjectId(); void setCarteObjectId( String containerObjectId ); @Override void batchComplete(); List<RemoteStep> getRemoteInputSteps(); List<RemoteStep> getRemoteOutputSteps(); @Override Date getRegistrationDate(); @Override boolean isGatheringMetrics(); @Override void setGatheringMetrics( boolean gatheringMetrics ); @Override boolean isForcingSeparateLogging(); @Override void setForcingSeparateLogging( boolean forcingSeparateLogging ); @Override Repository getRepository(); @Override void setRepository( Repository repository ); @Override IMetaStore getMetaStore(); @Override void setMetaStore( IMetaStore metaStore ); @Override int getCurrentOutputRowSetNr(); @Override void setCurrentOutputRowSetNr( int index ); @Override int getCurrentInputRowSetNr(); @Override void setCurrentInputRowSetNr( int index ); @Override Map<String, Object> getExtensionDataMap(); }
BaseStep implements VariableSpace, StepInterface, LoggingObjectInterface, ExtensionDataInterface { @Override public void cleanup() { for ( ServerSocket serverSocket : serverSockets ) { try { socketRepository.releaseSocket( serverSocket.getLocalPort() ); logDetailed( BaseMessages.getString( PKG, "BaseStep.Log.ReleasedServerSocketOnPort", serverSocket.getLocalPort() ) ); } catch ( IOException e ) { logError( "Cleanup: Unable to release server socket (" + serverSocket.getLocalPort() + ")", e ); } } List<RemoteStep> remoteInputSteps = getRemoteInputSteps(); if ( remoteInputSteps != null ) { cleanupRemoteSteps( remoteInputSteps ); } List<RemoteStep> remoteOutputSteps = getRemoteOutputSteps(); if ( remoteOutputSteps != null ) { cleanupRemoteSteps( remoteOutputSteps ); } } BaseStep( StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta,
                   Trans trans ); @Override boolean init( StepMetaInterface smi, StepDataInterface sdi ); @Override void dispose( StepMetaInterface smi, StepDataInterface sdi ); @Override void cleanup(); @Override long getProcessed(); void setCopy( int cop ); @Override int getCopy(); @Override long getErrors(); @Override void setErrors( long e ); @Override long getLinesRead(); long incrementLinesRead(); long decrementLinesRead(); void setLinesRead( long newLinesReadValue ); @Override long getLinesInput(); long incrementLinesInput(); void setLinesInput( long newLinesInputValue ); @Override long getLinesOutput(); long incrementLinesOutput(); void setLinesOutput( long newLinesOutputValue ); @Override long getLinesWritten(); long incrementLinesWritten(); long decrementLinesWritten(); void setLinesWritten( long newLinesWrittenValue ); @Override long getLinesUpdated(); long incrementLinesUpdated(); void setLinesUpdated( long newLinesUpdatedValue ); @Override long getLinesRejected(); long incrementLinesRejected(); @Override void setLinesRejected( long newLinesRejectedValue ); long getLinesSkipped(); long incrementLinesSkipped(); void setLinesSkipped( long newLinesSkippedValue ); @Override String getStepname(); void setStepname( String stepname ); Trans getDispatcher(); String getStatusDescription(); StepMetaInterface getStepMetaInterface(); void setStepMetaInterface( StepMetaInterface stepMetaInterface ); StepDataInterface getStepDataInterface(); void setStepDataInterface( StepDataInterface stepDataInterface ); @Override StepMeta getStepMeta(); void setStepMeta( StepMeta stepMeta ); TransMeta getTransMeta(); void setTransMeta( TransMeta transMeta ); @Override Trans getTrans(); @Override void putRow( RowMetaInterface rowMeta, Object[] row ); void putRowTo( RowMetaInterface rowMeta, Object[] row, RowSet rowSet ); void handlePutRowTo( RowMetaInterface rowMeta, Object[] row, RowSet rowSet ); void putError( RowMetaInterface rowMeta, Object[] row, long nrErrors, String errorDescriptions,
                        String fieldNames, String errorCodes ); @Override Object[] getRow(); void setRowHandler( RowHandler rowHandler ); RowHandler getRowHandler(); @Override void identifyErrorOutput(); static void safeModeChecking( RowMetaInterface referenceRowMeta, RowMetaInterface rowMeta ); Object[] getRowFrom( RowSet rowSet ); Object[] handleGetRowFrom( RowSet rowSet ); RowSet findInputRowSet( String sourceStep ); RowSet findInputRowSet( String from, int fromcopy, String to, int tocopy ); RowSet findOutputRowSet( String targetStep ); RowSet findOutputRowSet( String from, int fromcopy, String to, int tocopy ); @Override void setOutputDone(); void dispatch(); boolean isBasic(); boolean isDetailed(); boolean isDebug(); boolean isRowLevel(); void logMinimal( String message ); void logMinimal( String message, Object... arguments ); void logBasic( String message ); void logBasic( String message, Object... arguments ); void logDetailed( String message ); void logDetailed( String message, Object... arguments ); void logDebug( String message ); void logDebug( String message, Object... arguments ); void logRowlevel( String message ); void logRowlevel( String message, Object... arguments ); void logError( String message ); void logError( String message, Throwable e ); void logError( String message, Object... arguments ); int getNextClassNr(); boolean outputIsDone(); @Override void stopAll(); @Override boolean isStopped(); @Override boolean isRunning(); @Override boolean isPaused(); @Override void setStopped( boolean stopped ); @Override void setRunning( boolean running ); @Override void pauseRunning(); @Override void resumeRunning(); void setPaused( boolean paused ); void setPaused( AtomicBoolean paused ); boolean isInitialising(); @Override void markStart(); void setInternalVariables(); @Override void markStop(); @Override long getRuntime(); RowMetaAndData buildLog( String sname, int copynr, long lines_read, long lines_written,
                                  long lines_updated, long lines_skipped, long errors, Date start_date,
                                  Date end_date ); static final RowMetaInterface getLogFields( String comm ); @Override String toString(); @Override int rowsetOutputSize(); @Override int rowsetInputSize(); @Override void stopRunning( StepMetaInterface stepMetaInterface, StepDataInterface stepDataInterface ); @Deprecated void stopRunning(); void logSummary(); @Override String getStepID(); @Override List<RowSet> getInputRowSets(); void setInputRowSets( List<RowSet> inputRowSets ); @Override List<RowSet> getOutputRowSets(); void setOutputRowSets( List<RowSet> outputRowSets ); boolean isDistributed(); void setDistributed( boolean distributed ); @Override void addRowListener( RowListener rowListener ); @Override void removeRowListener( RowListener rowListener ); @Override List<RowListener> getRowListeners(); void addResultFile( ResultFile resultFile ); @Override Map<String, ResultFile> getResultFiles(); @Override StepExecutionStatus getStatus(); @Override String getPartitionID(); @Override void setPartitionID( String partitionID ); Map<String, BlockingRowSet> getPartitionTargets(); void setPartitionTargets( Map<String, BlockingRowSet> partitionTargets ); int getRepartitioning(); @Override void setRepartitioning( int repartitioning ); @Override boolean isPartitioned(); @Override void setPartitioned( boolean partitioned ); RowMetaInterface getInputRowMeta(); void setInputRowMeta( RowMetaInterface rowMeta ); RowMetaInterface getErrorRowMeta(); void setErrorRowMeta( RowMetaInterface errorRowMeta ); RowMetaInterface getPreviewRowMeta(); void setPreviewRowMeta( RowMetaInterface previewRowMeta ); @Override void copyVariablesFrom( VariableSpace space ); @Override String environmentSubstitute( String aString ); @Override String[] environmentSubstitute( String[] aString ); @Override String fieldSubstitute( String aString, RowMetaInterface rowMeta, Object[] rowData ); @Override VariableSpace getParentVariableSpace(); @Override void setParentVariableSpace( VariableSpace parent ); @Override String getVariable( String variableName, String defaultValue ); @Override String getVariable( String variableName ); @Override boolean getBooleanValueOfVariable( String variableName, boolean defaultValue ); @Override void initializeVariablesFrom( VariableSpace parent ); @Override String[] listVariables(); @Override void setVariable( String variableName, String variableValue ); @Override void shareVariablesWith( VariableSpace space ); @Override void injectVariables( Map<String, String> prop ); String getTypeId(); int getSlaveNr(); int getClusterSize(); int getUniqueStepNrAcrossSlaves(); int getUniqueStepCountAcrossSlaves(); List<ServerSocket> getServerSockets(); void setServerSockets( List<ServerSocket> serverSockets ); @Override void setUsingThreadPriorityManagment( boolean usingThreadPriorityManagment ); @Override boolean isUsingThreadPriorityManagment(); @Override void initBeforeStart(); List<StepListener> getStepListeners(); void setStepListeners( List<StepListener> stepListeners ); @Override boolean processRow( StepMetaInterface smi, StepDataInterface sdi ); @Override boolean canProcessOneRow(); @Override void addStepListener( StepListener stepListener ); @Override boolean isMapping(); SocketRepository getSocketRepository(); void setSocketRepository( SocketRepository socketRepository ); @Override String getObjectName(); @Override LogChannelInterface getLogChannel(); @Override String getFilename(); @Override String getLogChannelId(); @Override ObjectId getObjectId(); @Override ObjectRevision getObjectRevision(); @Override LoggingObjectType getObjectType(); @Override LoggingObjectInterface getParent(); @Override RepositoryDirectory getRepositoryDirectory(); @Override String getObjectCopy(); @Override LogLevel getLogLevel(); void setLogLevel( LogLevel logLevel ); static void closeQuietly( Closeable cl ); @Override String getContainerObjectId(); void setCarteObjectId( String containerObjectId ); @Override void batchComplete(); List<RemoteStep> getRemoteInputSteps(); List<RemoteStep> getRemoteOutputSteps(); @Override Date getRegistrationDate(); @Override boolean isGatheringMetrics(); @Override void setGatheringMetrics( boolean gatheringMetrics ); @Override boolean isForcingSeparateLogging(); @Override void setForcingSeparateLogging( boolean forcingSeparateLogging ); @Override Repository getRepository(); @Override void setRepository( Repository repository ); @Override IMetaStore getMetaStore(); @Override void setMetaStore( IMetaStore metaStore ); @Override int getCurrentOutputRowSetNr(); @Override void setCurrentOutputRowSetNr( int index ); @Override int getCurrentInputRowSetNr(); @Override void setCurrentInputRowSetNr( int index ); @Override Map<String, Object> getExtensionDataMap(); @Deprecated
  public long linesRead; @Deprecated
  public long linesWritten; @Deprecated
  public long linesInput; @Deprecated
  public long linesOutput; @Deprecated
  public long linesUpdated; @Deprecated
  public long linesSkipped; @Deprecated
  public long linesRejected; public boolean first; public boolean terminator; public List<Object[]> terminator_rows; }