signature:
public class RemoteDataCollector { /** * Returns info about remote pipelines that have changed since the last sending of events */
@ Override public List < PipelineAndValidationStatus > getRemotePipelinesWithChanges ( ) throws PipelineException { } }

implementation: List < PipelineAndValidationStatus > pipelineAndValidationStatuses = new ArrayList < > ( ) ; for ( Pair < PipelineState , Map < String , String > > pipelineStateAndOffset : stateEventListener . getPipelineStateEvents ( ) ) { PipelineState pipelineState = pipelineStateAndOffset . getLeft ( ) ; Map < String , String > offset = pipelineStateAndOffset . getRight ( ) ; String name = pipelineState . getPipelineId ( ) ; String rev = pipelineState . getRev ( ) ; boolean isClusterMode = ( pipelineState . getExecutionMode ( ) != ExecutionMode . STANDALONE ) ? true : false ; List < WorkerInfo > workerInfos = new ArrayList < > ( ) ; String title ; int runnerCount = 0 ; if ( pipelineStore . hasPipeline ( name ) ) { title = pipelineStore . getInfo ( name ) . getTitle ( ) ; Runner runner = manager . getRunner ( name , rev ) ; if ( isClusterMode ) { workerInfos = getWorkers ( runner . getSlaveCallbackList ( CallbackObjectType . METRICS ) ) ; } runnerCount = runner . getRunnerCount ( ) ; } else { title = null ; } pipelineAndValidationStatuses . add ( new PipelineAndValidationStatus ( getSchGeneratedPipelineName ( name , rev ) , title , rev , pipelineState . getTimeStamp ( ) , true , pipelineState . getStatus ( ) , pipelineState . getMessage ( ) , workerInfos , isClusterMode , getSourceOffset ( name , offset ) , null , runnerCount ) ) ; } return pipelineAndValidationStatuses ;


signature:
public class RemoteDataCollectorResult { /** * Creates a future ack result from the given parameter . It is expected that the caller will eventually wait for the
 * future ack result that is returned .
 * @ param futureResult the future ack
 * @ return an instance of this class with the given future ack */
public static RemoteDataCollectorResult futureAck ( Future < AckEvent > futureResult ) { } }

implementation: return new RemoteDataCollectorResult ( futureResult , null , false , null ) ;


signature:
public class JdbcOracleCDCRecordReader { /** * Get an operation code from record .
 * First , look for sdc . operation . code from record header .
 * If not set , look for " _ _ $ operation " in record . It is a specific field that MS SQL CDC origin set .
 * If either of them is set , check if the value is supported operation .
 * If not supported , let it handle by UnsupportedAction
 * check if the value is supported .
 * @ param record
 * @ return
 * @ throws StageException */
@ Override @ VisibleForTesting int getOperationFromRecord ( Record record , int defaultOpCode , UnsupportedOperationAction unsupportedAction , List < OnRecordErrorException > errorRecords ) { } }

implementation:
int opCode = - 1 ; // -1 is invalid and not used in OperationType .
String op = null ; try { // Try sdc . operation . type first
op = record . getHeader ( ) . getAttribute ( OperationType . SDC_OPERATION_TYPE ) ; // If not set , look for oracle . cdc . operation in record header .
if ( StringUtils . isBlank ( op ) ) { op = record . getHeader ( ) . getAttribute ( OracleCDCOperationCode . OPERATION ) ; if ( op != null ) { // Convert the Oracle specific operation code to SDC standard operation code
opCode = OracleCDCOperationCode . convertFromOracleToSDCCode ( op ) ; } } else { opCode = JDBCOperationType . convertToIntCode ( op ) ; } if ( opCode == - 1 ) { opCode = defaultOpCode ; } } catch ( NumberFormatException | UnsupportedOperationException ex ) { LOG . debug ( "Operation obtained from record is not supported: {}. Handle by UnsupportedOpertaionAction {}. {}" , ex . getMessage ( ) , unsupportedAction . getLabel ( ) , ex ) ; switch ( unsupportedAction ) { case DISCARD : LOG . debug ( "Discarding record with unsupported operation {}" , op ) ; break ; case SEND_TO_ERROR : LOG . debug ( "Sending record to error due to unsupported operation {}" , op ) ; errorRecords . add ( new OnRecordErrorException ( record , JdbcErrors . JDBC_70 , op ) ) ; break ; case USE_DEFAULT : opCode = defaultOpCode ; break ; default : // unknown action
LOG . debug ( "Sending record to error due to unknown operation: {}" , op ) ; } } return opCode ;


signature:
public class MetricsConfigurator { /** * Kept for backward compatibility with runtime stats , to be removed in future */
public static Gauge < Map < String , Object > > createGauge ( MetricRegistry metrics , String name , Gauge gauge , final String pipelineName , final String pipelineRev ) { } }

implementation: return create ( metrics , gauge , metricName ( name , GAUGE_SUFFIX ) , pipelineName , pipelineRev ) ;


signature:
public class MetricsConfigurator { /** * Remove metric object ( regardless of it ' s type ) */
private static boolean remove ( final MetricRegistry metrics , final String name , String pipelineName , String pipelineRev ) { } }

implementation: final String jmxNamePrefix = jmxPipelinePrefix ( pipelineName , pipelineRev ) ; final MetricRegistry metricRegistry = sdcMetrics ; if ( metricRegistry != null ) { AccessController . doPrivileged ( new PrivilegedAction < Void > ( ) { @ Override public Void run ( ) { metricRegistry . remove ( jmxNamePrefix + name ) ; return null ; } } ) ; } return metrics . remove ( name ) ;


signature:
public class JdbcMongoDBOplogRecordReader { /** * Records from MongoDB Oplog origin have a bit unique structure .
 * Records for Insert have all data in the field " o " , which is a map and contains all data for columns
 * Records for Delete have objectId in the field " o " , which is a map and contains only objectId .
 * Records for Update have a field " o2 " , which is a map and contains only objectId ,
 * and a field " o " where it stores data for updating columns and values .
 * @ param record
 * @ param op
 * @ param parameters
 * @ param columnsToFields
 * @ return */
@ Override public SortedMap < String , String > getColumnsToParameters ( final Record record , int op , Map < String , String > parameters , Map < String , String > columnsToFields ) { } }

implementation: SortedMap < String , String > columnsToParameters = new TreeMap < > ( ) ; for ( Map . Entry < String , String > entry : columnsToFields . entrySet ( ) ) { String columnName = entry . getKey ( ) ; String fieldPath = getFieldPath ( columnName , columnsToFields , op ) ; if ( record . has ( fieldPath ) ) { columnsToParameters . put ( columnName , parameters . get ( columnName ) ) ; } else { LOG . trace ( "Record is missing a field for column {} for the operation code {}" , columnName , op ) ; } } return columnsToParameters ;


signature:
public class JdbcMongoDBOplogRecordReader { /** * Return a fieldpath for the corresponding column name . The columnToField Map
 * contains column - to - field mapping , so simply get fieldpath for the column name
 * if operation is INSERT or DELETE .
 * We replace / o with / o2 to get _ id field , and change / o to / o / $ set to get
 * updating column & value fields .
 * @ param columnName column name
 * @ param columnsToField mapping of column name to field path
 * @ param op Not used here but is used in subclass .
 * @ return */
@ Override String getFieldPath ( String columnName , Map < String , String > columnsToField , int op ) { } }

implementation:
if ( op == OperationType . UPDATE_CODE ) { String fieldPath = columnsToField . get ( columnName ) ; if ( fieldPath == null ) { LOG . error ( "Column name {} is not defined in column-filed mapping" , columnName ) ; return null ; } if ( fieldPath . contains ( ID_FIELD ) ) { // _ id is stored in " / o2 / column _ name for update records . Need to change the fieldpath "
return fieldPath . replace ( OP_FIELD , OP2_FIELD ) ; } else { // column and values are stored in " / o / $ set / column _ name " . Need to change the fieldpath
return fieldPath . replaceFirst ( OP_FIELD , String . format ( "%s/%s" , OP_FIELD , SET_FIELD ) ) ; } } return columnsToField . get ( columnName ) ;


signature:
public class WholeFileTransformerProcessor { /** * Generate the Header attributes
 * @ param file the { @ link java . nio . file . Path } temporary parquet file path */
private Map < String , Object > generateHeaderAttrs ( Path file ) throws StageException { } }

implementation: try { Map < String , Object > recordHeaderAttr = new HashMap < > ( ) ; recordHeaderAttr . put ( HeaderAttributeConstants . FILE , file . toAbsolutePath ( ) ) ; recordHeaderAttr . put ( HeaderAttributeConstants . FILE_NAME , file . getFileName ( ) ) ; recordHeaderAttr . put ( HeaderAttributeConstants . SIZE , Files . size ( file ) ) ; recordHeaderAttr . put ( HeaderAttributeConstants . LAST_MODIFIED_TIME , Files . getLastModifiedTime ( file ) ) ; return recordHeaderAttr ; } catch ( IOException e ) { throw new TransformerStageCheckedException ( Errors . CONVERT_09 , e . toString ( ) , e ) ; }


signature:
public class WholeFileTransformerProcessor { /** * Validate the record is a whole file record
 * @ param record the { @ link com . streamsets . pipeline . api . Record } whole file record */
private void validateRecord ( Record record ) throws StageException { } }

implementation: try { FileRefUtil . validateWholeFileRecord ( record ) ; } catch ( IllegalArgumentException e ) { throw new TransformerStageCheckedException ( Errors . CONVERT_01 , e . toString ( ) , e ) ; }


signature:
public class WholeFileTransformerProcessor { /** * Return the temporary parquet file path and validate the path
 * @ param record the { @ link com . streamsets . pipeline . api . Record } whole file record
 * @ param sourceFileName the source Avro file name */
@ VisibleForTesting Path getAndValidateTempFilePath ( Record record , String sourceFileName ) throws StageException { } }

implementation:
RecordEL . setRecordInContext ( variables , record ) ; String dirPath ; try { dirPath = resolveEL ( tempDirElEval , variables , jobConfig . tempDir , String . class ) ; } catch ( ELEvalException ex ) { throw new TransformerStageCheckedException ( Errors . CONVERT_04 , jobConfig . tempDir ) ; } if ( Strings . isNullOrEmpty ( dirPath ) ) { throw new TransformerStageCheckedException ( Errors . CONVERT_02 , jobConfig . tempDir ) ; } if ( Strings . isNullOrEmpty ( sourceFileName ) ) { throw new TransformerStageCheckedException ( Errors . CONVERT_03 , FILENAME ) ; } String fileName = jobConfig . uniquePrefix + sourceFileName + jobConfig . fileNameSuffix ; Path tempParquetFile = Paths . get ( dirPath , fileName ) ; if ( ! tempParquetFile . isAbsolute ( ) ) { throw new TransformerStageCheckedException ( Errors . CONVERT_05 , tempParquetFile ) ; } try { if ( ! Files . exists ( tempParquetFile . getParent ( ) ) ) { Files . createDirectories ( Paths . get ( dirPath ) ) ; } } catch ( IOException ex ) { throw new TransformerStageCheckedException ( Errors . CONVERT_10 , tempParquetFile . toString ( ) , ex ) ; } // handle old temp files
try { handleOldTempFiles ( tempParquetFile ) ; } catch ( IOException ex ) { throw new TransformerStageCheckedException ( Errors . CONVERT_06 , tempParquetFile . toString ( ) , ex ) ; } return tempParquetFile ;


signature:
public class WholeFileTransformerProcessor { /** * Delete temporary parquet file
 * @ param tempParquetFile the { @ link java . nio . file . Path } temporary parquet file path */
private void handleOldTempFiles ( Path tempParquetFile ) throws IOException { } }

implementation: if ( tempParquetFile == null ) { LOG . warn ( "temporary parquet file is empty" ) ; return ; } Files . deleteIfExists ( tempParquetFile ) ;


signature:
public class WholeFileTransformerProcessor { /** * Return the Avro file input stream
 * @ param record the { @ link com . streamsets . pipeline . api . Record } whole file record */
private InputStream getAvroInputStream ( Record record ) throws StageException { } }

implementation:
try { FileRef fileRef = record . get ( FileRefUtil . FILE_REF_FIELD_PATH ) . getValueAsFileRef ( ) ; // get avro reader
final boolean includeChecksumInTheEvents = false ; InputStream is = FileRefUtil . getReadableStream ( getContext ( ) , fileRef , InputStream . class , includeChecksumInTheEvents , null , null ) ; return is ; } catch ( IOException ex ) { throw new TransformerStageCheckedException ( Errors . CONVERT_07 , ex . toString ( ) , ex ) ; }


signature:
public class WholeFileTransformerProcessor { /** * Return the Avro file reader
 * @ param is the { @ link java . io . InputStream } input stream of the source Avro file
 * @ param sourceFileName the source Avro file name */
private DataFileStream < GenericRecord > getFileReader ( InputStream is , String sourceFileName ) throws StageException { } }

implementation: try { DatumReader < GenericRecord > reader = new GenericDatumReader < > ( ) ; DataFileStream < GenericRecord > fileReader = new DataFileStream < > ( is , reader ) ; return fileReader ; } catch ( IOException ex ) { throw new TransformerStageCheckedException ( Errors . CONVERT_11 , sourceFileName , ex ) ; }


signature:
public class WholeFileTransformerProcessor { /** * Convert Avro record to Parquet
 * @ param sourceFileName the source Avro file name
 * @ param fileReader the { @ link org . apache . avro . file . DataFileStream } Avro file reader
 * @ param tempParquetFile the { @ link java . nio . file . Path } temporary parquet file path */
private void writeParquet ( String sourceFileName , DataFileStream < GenericRecord > fileReader , Path tempParquetFile ) throws StageException { } }

implementation:
long recordCount = 0 ; GenericRecord avroRecord ; Schema schema = fileReader . getSchema ( ) ; LOG . debug ( "Start reading input file : {}" , sourceFileName ) ; try { // initialize parquet writer
Configuration jobConfiguration = new Configuration ( ) ; String compressionCodecName = compressionElEval . eval ( variables , jobConfig . avroParquetConfig . compressionCodec , String . class ) ; jobConfiguration . set ( AvroParquetConstants . COMPRESSION_CODEC_NAME , compressionCodecName ) ; jobConfiguration . setInt ( AvroParquetConstants . ROW_GROUP_SIZE , jobConfig . avroParquetConfig . rowGroupSize ) ; jobConfiguration . setInt ( AvroParquetConstants . PAGE_SIZE , jobConfig . avroParquetConfig . pageSize ) ; jobConfiguration . setInt ( AvroParquetConstants . DICTIONARY_PAGE_SIZE , jobConfig . avroParquetConfig . dictionaryPageSize ) ; jobConfiguration . setInt ( AvroParquetConstants . MAX_PADDING_SIZE , jobConfig . avroParquetConfig . maxPaddingSize ) ; // Parquet writer
ParquetWriter . Builder builder = AvroToParquetConverterUtil . initializeWriter ( new org . apache . hadoop . fs . Path ( tempParquetFile . toString ( ) ) , schema , jobConfiguration ) ; parquetWriter = builder . build ( ) ; while ( fileReader . hasNext ( ) ) { avroRecord = fileReader . next ( ) ; parquetWriter . write ( avroRecord ) ; recordCount ++ ; } parquetWriter . close ( ) ; } catch ( IOException ex ) { throw new TransformerStageCheckedException ( Errors . CONVERT_08 , sourceFileName , recordCount , ex ) ; } LOG . debug ( "Finished writing {} records to {}" , recordCount , tempParquetFile . getFileName ( ) ) ;


signature:
public class BaseRedisSource { /** * { @ inheritDoc } */
@ Override public void destroy ( ) { } }

implementation:
// Clean up any open resources .
if ( null != redisClient ) { redisClient . disconnect ( ) ; redisClient . close ( ) ; redisClient = null ; } super . destroy ( ) ;


signature:
public class BlackListURLClassLoader { /** * Visible for testing only */
void validateClass ( String name ) { } }

implementation:
// python scripting engine generates _ _ * _ _ classes under all packages ( including SDC api package )
if ( ! ( name . endsWith ( "__" ) ) ) { for ( String blacklistedPackage : blacklistedPackages ) { if ( name . startsWith ( blacklistedPackage ) ) { throw new IllegalArgumentException ( String . format ( "Class '%s' cannot be present in %s" , name , toString ( ) ) ) ; } } }


signature:
public class BlackListURLClassLoader { /** * Visible for testing only */
void validateResource ( String name ) { } }

implementation: for ( String blacklistedPackage : blacklistedDirs ) { if ( name . startsWith ( blacklistedPackage ) ) { throw new IllegalArgumentException ( String . format ( "Resource '%s' cannot be present in %s" , name , toString ( ) ) ) ; } }


signature:
public class FieldEncrypter { /** * Checks that the encryption input is a supported type , otherwise sends the
 * record to error .
 * @ param record { @ link Record }
 * @ param field { @ link Field }
 * @ return empty { @ link Optional } if the record was sent to error */
public Optional < Field > checkInputEncrypt ( Record record , Field field ) { } }

implementation: if ( UNSUPPORTED_TYPES . contains ( field . getType ( ) ) ) { getContext ( ) . toError ( record , CRYPTO_03 , field . getType ( ) ) ; return Optional . empty ( ) ; } return Optional . of ( field ) ;


signature:
public class FieldEncrypter { /** * Checks that the encryption input is a supported type , otherwise sends the
 * record to StageException .
 * @ param field { @ link Field }
 * @ return empty { @ link Optional } if the record was sent to error */
public Optional < Field > checkInputEncrypt ( Field field ) throws StageException { } }

implementation: if ( UNSUPPORTED_TYPES . contains ( field . getType ( ) ) ) { throw new StageException ( CRYPTO_03 , field . getType ( ) ) ; } return Optional . of ( field ) ;


signature:
public class FieldEncrypter { /** * Checks that the decryption input is a valid type , otherwise sends the
 * record to error .
 * @ param record { @ link Record }
 * @ param field { @ link Field }
 * @ return empty { @ link Optional } if the record was sent to error */
public Optional < Field > checkInputDecrypt ( Record record , Field field ) { } }

implementation: if ( field . getType ( ) != Field . Type . BYTE_ARRAY ) { getContext ( ) . toError ( record , CRYPTO_02 , field . getType ( ) ) ; return Optional . empty ( ) ; } return Optional . of ( field ) ;


signature:
public class FieldEncrypter { /** * Checks that the decryption input is a valid type , otherwise sends the
 * record to StageException .
 * @ param field { @ link Field }
 * @ return empty { @ link Optional } if the record was sent to error */
public Optional < Field > checkInputDecrypt ( Field field ) throws StageException { } }

implementation: if ( field . getType ( ) != Field . Type . BYTE_ARRAY ) { throw new StageException ( CRYPTO_02 , field . getType ( ) ) ; } return Optional . of ( field ) ;


signature:
public class FieldEncrypter { /** * Does data type conversions in preparation for encryption .
 * @ param field { @ link Field } to encrypt
 * @ param context ( AAD )
 * @ return byte array to encrypt */
public byte [ ] prepareEncrypt ( Field field , Map < String , String > context ) { } }

implementation:
context . put ( SDC_FIELD_TYPE , field . getType ( ) . name ( ) ) ; if ( field . getType ( ) == Field . Type . BYTE_ARRAY ) { return field . getValueAsByteArray ( ) ; } else { // Treat all other data as strings
return field . getValueAsString ( ) . getBytes ( Charsets . UTF_8 ) ; }


signature:
public class FieldEncrypter { /** * Returns a decrypted { @ link Field } with its original type preserved when type information
 * has been preserved in the AAD .
 * @ param result of the decryption operation
 * @ return { @ link Field } */
public Field createResultFieldDecrypt ( CryptoResult < byte [ ] , ? > result ) { } }

implementation:
Field . Type fieldType = Field . Type . valueOf ( result . getEncryptionContext ( ) . getOrDefault ( SDC_FIELD_TYPE , Field . Type . BYTE_ARRAY . name ( ) ) ) ; // Field API prohibits STRING to BYTE _ ARRAY conversion so this is a special case
if ( fieldType == Field . Type . BYTE_ARRAY ) { return Field . create ( result . getResult ( ) ) ; } // Field API supports STRING to other primitive types .
return Field . create ( fieldType , new String ( result . getResult ( ) ) ) ;


signature:
public class SpoolDirRunnable { /** * Processes a batch from the specified file and offset up to a maximum batch size . If the file is fully processed
 * it must return - 1 , otherwise it must return the offset to continue from next invocation . */
public String generateBatch ( WrappedFile file , String offset , int maxBatchSize , BatchMaker batchMaker ) throws StageException , BadSpoolFileException { } }

implementation:
if ( offset == null ) { offset = "0" ; } String sourceFile = file . getFileName ( ) ; try { if ( parser == null ) { parser = SpoolDirUtil . getParser ( fs , file , conf . dataFormat , parserFactory , offset , conf . dataFormatConfig . wholeFileMaxObjectLen , rateLimitElEval , rateLimitElVars , conf . dataFormatConfig . rateLimit ) ; } Map < String , Object > recordHeaderAttr = generateHeaderAttrs ( file ) ; for ( int i = 0 ; i < maxBatchSize ; i ++ ) { try { Record record ; try { record = parser . parse ( ) ; } catch ( RecoverableDataParserException ex ) { // Propagate partially parsed record to error stream
record = ex . getUnparsedRecord ( ) ; recordHeaderAttr . put ( HeaderAttributeConstants . OFFSET , offset ) ; setHeaders ( record , recordHeaderAttr ) ; errorRecordHandler . onError ( new OnRecordErrorException ( record , ex . getErrorCode ( ) , ex . getParams ( ) ) ) ; perFileErrorCount ++ ; noMoreDataErrorCount ++ ; // We ' ll simply continue reading once this
continue ; } if ( record != null ) { recordHeaderAttr . put ( HeaderAttributeConstants . OFFSET , offset ) ; setHeaders ( record , recordHeaderAttr ) ; batchMaker . addRecord ( record ) ; offset = parser . getOffset ( ) ; if ( offset == null ) { offset = "0" ; } noMoreDataRecordCount ++ ; perFileRecordCount ++ ; } else { parser . close ( ) ; parser = null ; offset = MINUS_ONE ; break ; } } catch ( ObjectLengthException ex ) { String exOffset = offset ; offset = ( parser != null ) ? parser . getOffset ( ) : MINUS_ONE ; if ( offset == null ) { offset = "0" ; } errorRecordHandler . onError ( Errors . SPOOLDIR_02 , sourceFile , exOffset , ex ) ; perFileErrorCount ++ ; noMoreDataErrorCount ++ ; } } } catch ( IOException | DataParserException ex ) { if ( ex instanceof ClosedByInterruptException || ex . getCause ( ) instanceof ClosedByInterruptException ) { // If the pipeline was stopped , we may get a ClosedByInterruptException while reading avro data .
// This is because the thread is interrupted when the pipeline is stopped .
// Instead of sending the file to error , publish batch and move one .
} else { offset = MINUS_ONE ; String exOffset ; if ( ex instanceof OverrunException ) { exOffset = String . valueOf ( ( ( OverrunException ) ex ) . getStreamOffset ( ) ) ; } else { try { exOffset = ( parser != null ) ? parser . getOffset ( ) : MINUS_ONE ; } catch ( IOException ex1 ) { LOG . warn ( "Could not get the file offset to report with error, reason: {}" , ex1 . toString ( ) , ex ) ; exOffset = MINUS_ONE ; } } switch ( context . getOnErrorRecord ( ) ) { case DISCARD : break ; case TO_ERROR : // we failed to produce a record , which leaves the input file in an unknown state . all we can do here is
// throw an exception .
throw new BadSpoolFileException ( file . getAbsolutePath ( ) , exOffset , ex ) ; case STOP_PIPELINE : context . reportError ( Errors . SPOOLDIR_04 , sourceFile , exOffset , ex . toString ( ) , ex ) ; throw new StageException ( Errors . SPOOLDIR_04 , sourceFile , exOffset , ex . toString ( ) ) ; default : throw new IllegalStateException ( Utils . format ( "Unknown OnError value '{}'" , context . getOnErrorRecord ( ) , ex ) ) ; } } } finally { if ( MINUS_ONE . equals ( offset ) ) { if ( parser != null ) { try { parser . close ( ) ; parser = null ; } catch ( IOException ex ) { // NOP
} } } } return offset ;


signature:
public class SpoolDirRunnable { /** * Initialize the gauge with needed information */
private void initGaugeIfNeeded ( ) { } }

implementation: gaugeMap . put ( THREAD_NAME , Thread . currentThread ( ) . getName ( ) ) ; gaugeMap . put ( STATUS , "" ) ; gaugeMap . put ( CURRENT_FILE , "" ) ;


signature:
public class SpoolDirRunnable { /** * Handle Exception */
private void handleStageError ( ErrorCode errorCode , Exception e ) { } }

implementation:
final String errorMessage = "Failure Happened" ; LOG . error ( errorMessage , e ) ; try { errorRecordHandler . onError ( errorCode , e ) ; } catch ( StageException se ) { LOG . error ( "Error when routing to stage error" , se ) ; // Way to throw stage exception from runnable to main source thread
Throwables . propagate ( se ) ; }


signature:
public class CycleDetector { /** * Find cycles in the { @ link # directedGraph } .
 * @ return Set of list of vertices involved in the cycle . */
boolean isGraphCyclic ( ) { } }

implementation: for ( V vertex : directedGraph . vertices ( ) ) { boolean areThereCycles = findCycles ( new LinkedHashSet < V > ( ) , vertex ) ; if ( areThereCycles ) { return true ; } } return false ;


signature:
public class Matcher { /** * Changes the Pattern that this Matcher uses to find matches with
 * @ param newPattern the new pattern
 * @ return this Matcher */
public Matcher usePattern ( Pattern newPattern ) { } }

implementation: if ( newPattern == null ) { throw new IllegalArgumentException ( "newPattern cannot be null" ) ; } this . parentPattern = newPattern ; matcher . usePattern ( newPattern . pattern ( ) ) ; return this ;


signature:
public class Matcher { /** * Implements a non - terminal append - and - replace step .
 * @ param sb The target string buffer
 * @ param replacement The replacement string
 * @ return The target string buffer */
public Matcher appendReplacement ( StringBuffer sb , String replacement ) { } }

implementation: matcher . appendReplacement ( sb , parentPattern . replaceProperties ( replacement ) ) ; return this ;


signature:
public class Matcher { /** * Finds all named groups that exist in the input string . This resets the
 * matcher and attempts to match the input against the pre - specified
 * pattern .
 * @ return a map of the group named and matched values
 * ( empty if no match found ) */
@ Override public Map < String , String > namedGroups ( ) { } }

implementation: Map < String , String > result = new LinkedHashMap < String , String > ( ) ; if ( matcher . find ( 0 ) ) { for ( String groupName : parentPattern . groupNames ( ) ) { String groupValue = matcher . group ( groupIndex ( groupName ) ) ; result . put ( groupName , groupValue ) ; } } return result ;


signature:
public class Matcher { /** * Replaces every subsequence of the input sequence that matches the pattern
 * with the given replacement string .
 * @ param replacement The replacement string
 * @ return The string constructed by replacing each matching subsequence by
 * the replacement string , substituting captured subsequences as needed */
public String replaceAll ( String replacement ) { } }

implementation: String r = parentPattern . replaceProperties ( replacement ) ; return matcher . replaceAll ( r ) ;


signature:
public class MapRDBCDCSource { /** * no trespassing . . . */
private Properties getKafkaProperties ( ) { } }

implementation: Properties props = new Properties ( ) ; props . putAll ( conf . streamsOptions ) ; props . setProperty ( "group.id" , conf . consumerGroup ) ; props . setProperty ( "max.poll.records" , String . valueOf ( batchSize ) ) ; props . setProperty ( "enable.auto.commit" , "true" ) ; props . setProperty ( "auto.commit.interval.ms" , "1000" ) ; props . setProperty ( "key.deserializer" , "org.apache.kafka.common.serialization.ByteArrayDeserializer" ) ; props . setProperty ( "value.deserializer" , "com.mapr.db.cdc.ChangeDataRecordDeserializer" ) ; return props ;


signature:
public class EscapeUtil { /** * This method is used during deserializer and sqpath ( Single quote escaped path ) is passed to determine the last field name . */
public static String getLastFieldNameFromPath ( String path ) { } }

implementation:
String [ ] pathSplit = ( path != null ) ? path . split ( "/" ) : null ; if ( pathSplit != null && pathSplit . length > 0 ) { String lastFieldName = pathSplit [ pathSplit . length - 1 ] ; // handle special case field name containing slash eg . / ' foo / bar '
boolean singleQuoted = lastFieldName . charAt ( 0 ) == QUOTE_CHAR && lastFieldName . charAt ( lastFieldName . length ( ) - 1 ) == QUOTE_CHAR ; if ( lastFieldName . contains ( "'" ) && ! singleQuoted ) { // If path contains slash inside name , split it by " / ' "
pathSplit = path . split ( "/'" ) ; if ( pathSplit . length > 0 ) { lastFieldName = "'" + pathSplit [ pathSplit . length - 1 ] ; singleQuoted = true ; } } if ( singleQuoted ) { return singleQuoteUnescape ( lastFieldName ) ; } else { return lastFieldName ; } } return path ;


signature:
public class EscapeUtil { /** * This method escapes backslash , double quotes and single quotes ( keeping replacement of ' to \ \ \ \ \ '
 * as is so as to maintain backward compatibility any serialization / deserialization ) */
private static String escapeQuotesAndBackSlash ( String path , boolean isSingleQuoteEscape ) { } }

implementation: String quoteChar = isSingleQuoteEscape ? "'" : "\"" ; StringBuilder sb = new StringBuilder ( path . length ( ) * 2 ) . append ( quoteChar ) ; char [ ] chars = path . toCharArray ( ) ; for ( char c : chars ) { if ( c == '\\' ) { sb . append ( "\\\\" ) ; } else if ( c == '"' ) { sb . append ( isSingleQuoteEscape ? "\\\"" : "\\\\\"" ) ; } else if ( c == '\'' ) { sb . append ( isSingleQuoteEscape ? "\\\\\'" : "\\\'" ) ; } else { sb . append ( c ) ; } } return sb . append ( quoteChar ) . toString ( ) ;


signature:
public class EscapeUtil { /** * This method un escapes backslash , double quotes and single quotes ( keeping replacement of \ \ \ \ \ ' to '
 * as is so as to maintain backward compatibility any serialization / deserialization ) */
private static String unescapeQuotesAndBackSlash ( String path , boolean isSingleQuoteUnescape ) { } }


signature:
public class EscapeUtil { /** * This method un escapes backslash and un escapes extra escapes before double quotes and single quotes
 * ( appended by { @ link # escapeQuotesAndBackSlash ( String , boolean ) } ) .
 * This method should be used internally and not during for any serialization / deserialization */
public static String standardizePathForParse ( String path , boolean isSingleQuoteEscape ) { } }

implementation: path = isSingleQuoteEscape ? path . replace ( "\\\\\'" , "\\'" ) : path . replace ( "\\\\\"" , "\\\"" ) ; return path . replace ( "\\\\" , "\\" ) ;


signature:
public class DataFormatUpgradeHelper { /** * There is a problem with some older stages that originally were not using our data parser library as the
 * upgrade on those stages does not create all the properties that were introduced by the data parser library .
 * For example File Tail origin doesn ' t support avro , so the upgrade procedure doesn ' t use create the avro
 * specific properties . This is normally not a problem as post - upgrade SDC will create all missing properties
 * with default values . However this particular upgrade will fail if the property avroSchema is missing .
 * Hence for such stages , this method will ensure that the property avroSchema is properly present . The migration
 * to data parser library happened for majority of stages in 1.2 release and hence this really affects only people
 * upgrading from 1.1 . x all the way to 2.1 or above . If the user upgraded from 1.1 . x and then to any other release
 * below 2.1 first , they would not hit this issue as the property avroSchema would be added with default value . */
public static void ensureAvroSchemaExists ( List < Config > configs , String prefix ) { } }

implementation: Optional < Config > avroSchema = findByName ( configs , "avroSchema" ) ; if ( ! avroSchema . isPresent ( ) ) { configs . add ( new Config ( prefix + ".avroSchema" , null ) ) ; }


signature:
public class Util { /** * Get global variable value .
 * @ param dataSource
 * @ param variable
 * @ return global variable value of empty string .
 * @ throws SQLException */
public static String getGlobalVariable ( DataSource dataSource , String variable ) throws SQLException { } }

implementation: try ( Connection conn = dataSource . getConnection ( ) ) { try ( Statement stmt = conn . createStatement ( ) ; ResultSet rs = stmt . executeQuery ( String . format ( "show global variables like '%s'" , variable ) ) ; ) { if ( rs . next ( ) ) { return rs . getString ( 2 ) ; } else { return "" ; } } }


signature: public class BeanHelper { public static StatusJson wrapState ( PipelineStatus status ) { } }

implementation: if ( status == null ) { return null ; } switch ( status ) { case STOPPED : return StatusJson . STOPPED ; case STOPPING : return StatusJson . STOPPING ; case RUNNING : return StatusJson . RUNNING ; case RUN_ERROR : return StatusJson . RUN_ERROR ; case FINISHED : return StatusJson . FINISHED ; case CONNECTING : return StatusJson . CONNECTING ; case CONNECT_ERROR : return StatusJson . CONNECT_ERROR ; case DISCONNECTED : return StatusJson . DISCONNECTED ; case DISCONNECTING : return StatusJson . DISCONNECTING ; case EDITED : return StatusJson . EDITED ; case FINISHING : return StatusJson . FINISHING ; case KILLED : return StatusJson . KILLED ; case RUNNING_ERROR : return StatusJson . RUNNING_ERROR ; case STARTING : return StatusJson . STARTING ; case STARTING_ERROR : return StatusJson . STARTING_ERROR ; case START_ERROR : return StatusJson . START_ERROR ; case RETRY : return StatusJson . RETRY ; case STOP_ERROR : return StatusJson . STOP_ERROR ; case STOPPING_ERROR : return StatusJson . STOPPING_ERROR ; case DELETED : return StatusJson . DELETED ; default : throw new IllegalArgumentException ( "Unrecognized state" + status ) ; }


signature:
public class Pipeline { /** * Create pipeline start event . */
private Record createStartEvent ( ) { } }

implementation:
Preconditions . checkState ( startEventStage != null , "Start Event Stage is not set!" ) ; EventRecord eventRecord = new EventRecordImpl ( "pipeline-start" , 1 , startEventStage . getInfo ( ) . getInstanceName ( ) , "" , null , null ) ; Map < String , Field > rootField = new LinkedHashMap < > ( ) ; rootField . put ( "user" , Field . create ( Field . Type . STRING , userContext . getUser ( ) ) ) ; rootField . put ( "pipelineId" , Field . create ( Field . Type . STRING , name ) ) ; rootField . put ( "pipelineTitle" , Field . create ( Field . Type . STRING , pipelineConf . getTitle ( ) ) ) ; // Pipeline parameters
Map < String , Field > parameters = new LinkedHashMap < > ( ) ; if ( runtimeParameters != null ) { for ( Map . Entry < String , Object > entry : runtimeParameters . entrySet ( ) ) { parameters . put ( entry . getKey ( ) , Field . create ( Field . Type . STRING , entry . getValue ( ) . toString ( ) ) ) ; } } rootField . put ( "parameters" , Field . create ( parameters ) ) ; eventRecord . set ( Field . create ( rootField ) ) ; return eventRecord ;


signature:
public class Pipeline { /** * Create pipeline stop event . */
private Record createStopEvent ( PipelineStopReason stopReason ) { } }

implementation: Preconditions . checkState ( stopEventStage != null , "Stop Event Stage is not set!" ) ; EventRecord eventRecord = new EventRecordImpl ( "pipeline-stop" , 1 , stopEventStage . getInfo ( ) . getInstanceName ( ) , "" , null , null ) ; Map < String , Field > rootField = new LinkedHashMap < > ( ) ; rootField . put ( "reason" , Field . create ( Field . Type . STRING , stopReason . name ( ) ) ) ; rootField . put ( "pipelineId" , Field . create ( Field . Type . STRING , name ) ) ; rootField . put ( "pipelineTitle" , Field . create ( Field . Type . STRING , pipelineConf . getTitle ( ) ) ) ; eventRecord . set ( Field . create ( rootField ) ) ; return eventRecord ;


signature:
public class SobjectRecordCreator { /** * infinite recursion */
private void getAllReferences ( PartnerConnection partnerConnection , Map < String , ObjectMetadata > metadataMap , List < List < Pair < String , String > > > references , String [ ] allTypes , int depth ) throws ConnectionException { } }

implementation:
if ( depth < 0 ) { return ; } List < String > next = new ArrayList < > ( ) ; for ( int typeIndex = 0 ; typeIndex < allTypes . length ; typeIndex += MAX_METADATA_TYPES ) { int copyTo = Math . min ( typeIndex + MAX_METADATA_TYPES , allTypes . length ) ; String [ ] types = Arrays . copyOfRange ( allTypes , typeIndex , copyTo ) ; // Special case - we prepopulate the cache with the root sobject type - don ' t repeat
// ourselves
if ( types . length > 1 || ! metadataMap . containsKey ( types [ 0 ] ) ) { for ( DescribeSObjectResult result : partnerConnection . describeSObjects ( types ) ) { Map < String , Field > fieldMap = new LinkedHashMap < > ( ) ; Map < String , Field > relationshipMap = new LinkedHashMap < > ( ) ; for ( Field field : result . getFields ( ) ) { fieldMap . put ( field . getName ( ) . toLowerCase ( ) , field ) ; String relationshipName = field . getRelationshipName ( ) ; if ( relationshipName != null ) { relationshipMap . put ( relationshipName . toLowerCase ( ) , field ) ; } } Map < String , String > childRelationships = new LinkedHashMap < > ( ) ; for ( ChildRelationship child : result . getChildRelationships ( ) ) { if ( child . getRelationshipName ( ) != null ) { childRelationships . put ( child . getRelationshipName ( ) . toLowerCase ( ) , child . getChildSObject ( ) . toLowerCase ( ) ) ; } } metadataMap . put ( result . getName ( ) . toLowerCase ( ) , new ObjectMetadata ( fieldMap , relationshipMap , childRelationships ) ) ; } } if ( references != null ) { for ( List < Pair < String , String > > path : references ) { // Top field name in the path should be in the metadata now
if ( ! path . isEmpty ( ) ) { Pair < String , String > top = path . get ( 0 ) ; Field field = metadataMap . get ( top . getLeft ( ) ) . getFieldFromRelationship ( top . getRight ( ) ) ; Set < String > sobjectNames = metadataMap . keySet ( ) ; for ( String ref : field . getReferenceTo ( ) ) { ref = ref . toLowerCase ( ) ; if ( ! sobjectNames . contains ( ref ) && ! next . contains ( ref ) ) { next . add ( ref ) ; } if ( path . size ( ) > 1 ) { path . set ( 1 , Pair . of ( ref , path . get ( 1 ) . getRight ( ) ) ) ; } } // SDC - 10422 Polymorphic references have an implicit reference to the Name object type
if ( field . isPolymorphicForeignKey ( ) ) { next . add ( NAME ) ; } path . remove ( 0 ) ; } } } } if ( ! next . isEmpty ( ) ) { getAllReferences ( partnerConnection , metadataMap , references , next . toArray ( new String [ 0 ] ) , depth - 1 ) ; }


signature:
public class SobjectRecordCreator { /** * since Salesforce doesn ' t like scientific notation in queries */
protected String fixOffset ( String offsetColumn , String offset ) { } }

implementation:
com . sforce . soap . partner . Field sfdcField = getFieldMetadata ( sobjectType , offsetColumn ) ; if ( SobjectRecordCreator . DECIMAL_TYPES . contains ( sfdcField . getType ( ) . toString ( ) ) && offset . contains ( "E" ) ) { BigDecimal val = new BigDecimal ( offset ) ; offset = val . toPlainString ( ) ; if ( val . compareTo ( MAX_OFFSET_INT ) > 0 && ! offset . contains ( "." ) ) { // We need the " . 0 " suffix since Salesforce doesn ' t like integer
// bigger than 2147483647
offset += ".0" ; } } return offset ;


signature:
public class ConfigValueExtractor { /** * RUNTIME supports only Numeric types and String at the moment */
private Object extractAsRuntime ( Field field , String valueStr ) { } }

implementation: if ( field . getType ( ) == Byte . TYPE || field . getType ( ) == Byte . class || field . getType ( ) == Short . TYPE || field . getType ( ) == Short . class || field . getType ( ) == Integer . TYPE || field . getType ( ) == Integer . class || field . getType ( ) == Long . TYPE || field . getType ( ) == Long . class || field . getType ( ) == Float . TYPE || field . getType ( ) == Float . class || field . getType ( ) == Double . TYPE || field . getType ( ) == Double . class ) { return extractAsNumber ( field , valueStr ) ; } else if ( String . class . isAssignableFrom ( field . getType ( ) ) ) { return valueStr ; } throw new IllegalArgumentException ( Utils . format ( "Invalid type for RUNTIME type: {}" , field . getType ( ) ) ) ;


signature:
public class HiveQueryExecutor { /** * Add a new partition to the given table , with optional custom location .
 * @ param qualifiedTableName Table name in the form ` database ` . ` table ` .
 * @ param partitionNameValueMap Map between partition column names and their values .
 * @ param partitionTypeMap Map between partition column names and their types .
 * @ param partitionPath Location in the Hadoop filesystem where the partition will be created . If null , the default
 * location will be used . */
public void executeAlterTableAddPartitionQuery ( String qualifiedTableName , LinkedHashMap < String , String > partitionNameValueMap , Map < String , HiveTypeInfo > partitionTypeMap , String partitionPath ) throws StageException { } }

implementation: String sql = buildPartitionAdditionQuery ( qualifiedTableName , partitionNameValueMap , partitionTypeMap , partitionPath ) ; execute ( sql ) ;


signature:
public class HiveQueryExecutor { /** * Execute Alter Table set Table Properties
 * @ param qualifiedTableName qualified table name .
 * @ param partitionPath parition path .
 * @ throws StageException in case of any { @ link SQLException } */
public void executeAlterTableSetTblPropertiesQuery ( String qualifiedTableName , String partitionPath ) throws StageException { } }

implementation: String sql = buildSetTablePropertiesQuery ( qualifiedTableName , partitionPath ) ; execute ( sql ) ;


signature:
public class HiveQueryExecutor { /** * Returns { @ link Set } of partitions
 * @ param qualifiedTableName qualified table name
 * @ return { @ link Set } of partitions
 * @ throws StageException in case of any { @ link SQLException } */
public Set < PartitionInfoCacheSupport . PartitionValues > executeShowPartitionsQuery ( String qualifiedTableName ) throws StageException { } }

implementation: String sql = buildShowPartitionsQuery ( qualifiedTableName ) ; return executeQuery ( sql , new WithResultSet < Set < PartitionInfoCacheSupport . PartitionValues > > ( ) { @ Override public Set < PartitionInfoCacheSupport . PartitionValues > run ( ResultSet rs ) throws SQLException , StageException { Set < PartitionInfoCacheSupport . PartitionValues > partitionValuesSet = new HashSet < > ( ) ; while ( rs . next ( ) ) { String partitionInfoString = rs . getString ( 1 ) ; String [ ] partitionInfoSplit = partitionInfoString . split ( HiveMetastoreUtil . SEP ) ; LinkedHashMap < String , String > vals = new LinkedHashMap < > ( ) ; for ( String partitionValInfo : partitionInfoSplit ) { String [ ] partitionNameVal = partitionValInfo . split ( "=" ) ; vals . put ( partitionNameVal [ 0 ] , partitionNameVal [ 1 ] ) ; } partitionValuesSet . add ( new PartitionInfoCacheSupport . PartitionValues ( vals ) ) ; } return partitionValuesSet ; } } ) ;


signature:
public class HiveQueryExecutor { /** * Returns { @ link Pair } of Column Type Info and Partition Type Info .
 * @ param qualifiedTableName qualified table name .
 * @ return { @ link Pair } of Column Type Info and Partition Type Info .
 * @ throws StageException in case of any { @ link SQLException } */
public Pair < LinkedHashMap < String , HiveTypeInfo > , LinkedHashMap < String , HiveTypeInfo > > executeDescTableQuery ( String qualifiedTableName ) throws StageException { } }

implementation:
String sql = buildDescTableQuery ( qualifiedTableName ) ; return executeQuery ( sql , new WithResultSet < Pair < LinkedHashMap < String , HiveTypeInfo > , LinkedHashMap < String , HiveTypeInfo > > > ( ) { @ Override public Pair < LinkedHashMap < String , HiveTypeInfo > , LinkedHashMap < String , HiveTypeInfo > > run ( ResultSet rs ) throws SQLException , StageException { LinkedHashMap < String , HiveTypeInfo > columnTypeInfo = extractTypeInfo ( rs ) ; processDelimiter ( rs , "#" ) ; processDelimiter ( rs , "#" ) ; processDelimiter ( rs , "" ) ; LinkedHashMap < String , HiveTypeInfo > partitionTypeInfo = extractTypeInfo ( rs ) ; // Remove partition columns from the columns map .
for ( String partitionCol : partitionTypeInfo . keySet ( ) ) { columnTypeInfo . remove ( partitionCol ) ; } return Pair . of ( columnTypeInfo , partitionTypeInfo ) ; } } ) ;


signature:
public class HiveQueryExecutor { /** * Returns location for given database .
 * @ param dbName Database name
 * @ return Path where the database is stored
 * @ throws StageException in case of any { @ link SQLException } */
public String executeDescribeDatabase ( String dbName ) throws StageException { } }

implementation: String sql = buildDescribeDatabase ( dbName ) ; return executeQuery ( sql , rs -> { if ( ! rs . next ( ) ) { throw new HiveStageCheckedException ( Errors . HIVE_35 , "Database doesn't exists." ) ; } return HiveMetastoreUtil . stripHdfsHostAndPort ( rs . getString ( RESULT_SET_LOCATION ) ) ; } ) ;


signature:
public class HiveQueryExecutor { /** * Returns { @ link Pair } of IsExternal and storedAsAvro TBLProperties .
 * @ param qualifiedTableName qualified table name .
 * @ return { @ link Pair } of IsExternal and storedAsAvro .
 * @ throws StageException in case of any { @ link SQLException } */
public Pair < Boolean , Boolean > executeShowTBLPropertiesQuery ( String qualifiedTableName ) throws StageException { } }

implementation: String sql = String . format ( SHOW_TBLPROPERTIES , qualifiedTableName ) ; return executeQuery ( sql , new WithResultSet < Pair < Boolean , Boolean > > ( ) { @ Override public Pair < Boolean , Boolean > run ( ResultSet rs ) throws SQLException { boolean isExternal = false , useAsAvro = true ; while ( rs . next ( ) ) { String propName = rs . getString ( RESULT_SET_PROP_NAME ) ; String propValue = rs . getString ( RESULT_SET_PROP_VALUE ) ; if ( propName . toUpperCase ( ) . equals ( EXTERNAL ) ) { isExternal = Boolean . valueOf ( propValue ) ; } else if ( propName . equals ( AVRO_SCHEMA_URL ) ) { useAsAvro = false ; } } return Pair . of ( isExternal , useAsAvro ) ; } } ) ;


signature:
public class HiveQueryExecutor { /** * Execute given query .
 * With all required side effects such as catching exceptions , updating metrics , . . .
 * @ param query Query to execute
 * @ throws StageException */
private void execute ( String query ) throws StageException { } }

implementation: LOG . debug ( "Executing SQL: {}" , query ) ; Timer . Context t = updateTimer . time ( ) ; try ( Statement statement = hiveConfigBean . getHiveConnection ( ) . createStatement ( ) ) { statement . execute ( query ) ; } catch ( Exception e ) { LOG . error ( "Exception while processing query: {}" , query , e ) ; throw new HiveStageCheckedException ( Errors . HIVE_20 , query , e . getMessage ( ) ) ; } finally { long time = t . stop ( ) ; LOG . debug ( "Query '{}' took {} nanoseconds" , query , time ) ; updateMeter . mark ( ) ; }


signature:
public class HiveQueryExecutor { /** * Execute given query and process it ' s result set .
 * With all required side effects such as catching exceptions , updating metrics , . . .
 * @ param query Query to execute
 * @ param execution Action that should be carried away on the result set
 * @ throws StageException */
private < T > T executeQuery ( String query , WithResultSet < T > execution ) throws StageException { } }

implementation:
LOG . debug ( "Executing SQL:  {}" , query ) ; Timer . Context t = selectTimer . time ( ) ; try ( Statement statement = hiveConfigBean . getHiveConnection ( ) . createStatement ( ) ; ResultSet rs = statement . executeQuery ( query ) ; ) { // Stop timer immediately so that we ' re calculating only query execution time and not the processing time
long time = t . stop ( ) ; LOG . debug ( "Query '{}' took {} nanoseconds" , query , time ) ; t = null ; return execution . run ( rs ) ; } catch ( Exception e ) { LOG . error ( "Exception while processing query: {}" , query , e ) ; throw new HiveStageCheckedException ( Errors . HIVE_20 , query , e . getMessage ( ) ) ; } finally { // If the timer wasn ' t stopped due to exception yet , stop it now
if ( t != null ) { long time = t . stop ( ) ; LOG . debug ( "Query '{}' took {} nanoseconds" , query , time ) ; } selectMeter . mark ( ) ; }


signature:
public class SampleSource { /** * { @ inheritDoc } */
@ Override public String produce ( String lastSourceOffset , int maxBatchSize , BatchMaker batchMaker ) throws StageException { } }

implementation:
// Offsets can vary depending on the data source . Here we use an integer as an example only .
long nextSourceOffset = 0 ; if ( lastSourceOffset != null ) { nextSourceOffset = Long . parseLong ( lastSourceOffset ) ; } int numRecords = 0 ; // TODO : As the developer , implement your logic that reads from a data source in this method .
// Create records and add to batch . Records must have a string id . This can include the source offset
// or other metadata to help uniquely identify the record itself .
while ( numRecords < maxBatchSize ) { Record record = getContext ( ) . createRecord ( "some-id::" + nextSourceOffset ) ; Map < String , Field > map = new HashMap < > ( ) ; map . put ( "fieldName" , Field . create ( "Some Value" ) ) ; record . set ( Field . create ( map ) ) ; batchMaker . addRecord ( record ) ; ++ nextSourceOffset ; ++ numRecords ; } return String . valueOf ( nextSourceOffset ) ;


signature:
public class PipeRunner { /** * Run batch with given consumer for each pipe .
 * This method will also set the logger appropriately and calculate the runner specific metrics . */
public void executeBatch ( String offsetKey , String offsetValue , long batchStartTime , ThrowingConsumer < Pipe > consumer ) throws PipelineRuntimeException , StageException { } }

implementation:
MDC . put ( LogConstants . RUNNER , String . valueOf ( runnerId ) ) ; // Persist static information for the batch ( this won ' t change as the batch progresses )
this . runtimeMetricGauge . put ( METRIC_BATCH_START_TIME , batchStartTime ) ; this . runtimeMetricGauge . put ( METRIC_OFFSET_KEY , Optional . ofNullable ( offsetKey ) . orElse ( "" ) ) ; this . runtimeMetricGauge . put ( METRIC_OFFSET_VALUE , Optional . ofNullable ( offsetValue ) . orElse ( "" ) ) ; this . runtimeMetricGauge . put ( METRIC_STAGE_START_TIME , System . currentTimeMillis ( ) ) ; try { // Run one pipe at a time
for ( Pipe p : pipes ) { String instanceName = p . getStage ( ) . getInfo ( ) . getInstanceName ( ) ; this . runtimeMetricGauge . put ( METRIC_CURRENT_STAGE , instanceName ) ; MDC . put ( LogConstants . STAGE , instanceName ) ; if ( p instanceof StagePipe ) { this . runtimeMetricGauge . put ( METRIC_STAGE_START_TIME , System . currentTimeMillis ( ) ) ; } acceptConsumer ( consumer , p ) ; } // We ' ve successfully finished batch
this . runtimeMetricGauge . computeIfPresent ( METRIC_BATCH_COUNT , ( key , value ) -> ( ( long ) value ) + 1 ) ; } finally { resetBatchSpecificMetrics ( ) ; MDC . put ( LogConstants . RUNNER , "" ) ; MDC . put ( LogConstants . STAGE , "" ) ; }


signature:
public class PipeRunner { /** * Execute given consumer for each pipe , rethrowing usual exceptions as RuntimeException .
 * Suitable for consumer that is not suppose to throw PipelineException and StageException . This method will
 * not calculate usual stage metrics . */
public void forEach ( ThrowingConsumer < Pipe > consumer ) { } }

implementation: try { MDC . put ( LogConstants . RUNNER , String . valueOf ( runnerId ) ) ; try { for ( Pipe p : pipes ) { MDC . put ( LogConstants . STAGE , p . getStage ( ) . getInfo ( ) . getInstanceName ( ) ) ; acceptConsumer ( consumer , p ) ; } } finally { MDC . put ( LogConstants . RUNNER , "" ) ; MDC . put ( LogConstants . STAGE , "" ) ; } } catch ( PipelineException | StageException e ) { throw new RuntimeException ( e ) ; }


signature:
public class PipeRunner { /** * Retrieve OffsetCommitTrigger pipe .
 * If it exists , null otherwise . */
public OffsetCommitTrigger getOffsetCommitTrigger ( ) { } }

implementation: for ( Pipe pipe : pipes ) { Stage stage = pipe . getStage ( ) . getStage ( ) ; if ( stage instanceof Target && stage instanceof OffsetCommitTrigger ) { return ( OffsetCommitTrigger ) stage ; } } return null ;


signature:
public class PipeRunner { /** * Return true if at least one stage is configured with STOP _ PIPELINE for OnRecordError policy . */
public boolean onRecordErrorStopPipeline ( ) { } }

implementation: for ( Pipe pipe : pipes ) { StageContext stageContext = pipe . getStage ( ) . getContext ( ) ; if ( stageContext . getOnErrorRecord ( ) == OnRecordError . STOP_PIPELINE ) { return true ; } } return false ;


signature:
public class PipeRunner { /** * Accept given consumer and proper log context of any exception . */
private void acceptConsumer ( ThrowingConsumer < Pipe > consumer , Pipe p ) throws PipelineRuntimeException , StageException { } }

implementation:
try { // Process pipe
consumer . accept ( p ) ; } catch ( Throwable t ) { String instanceName = p . getStage ( ) . getInfo ( ) . getInstanceName ( ) ; LOG . error ( "Failed executing stage '{}': {}" , instanceName , t . toString ( ) , t ) ; Throwables . propagateIfInstanceOf ( t , PipelineRuntimeException . class ) ; Throwables . propagateIfInstanceOf ( t , StageException . class ) ; Throwables . propagate ( t ) ; }


signature:
public class BigQueryTarget { /** * Evaluate and obtain the row id if the expression is present or return null . */
private String getInsertIdForRecord ( ELVars elVars , Record record ) throws OnRecordErrorException { } }

implementation: String recordId = null ; RecordEL . setRecordInContext ( elVars , record ) ; try { if ( ! ( StringUtils . isEmpty ( conf . rowIdExpression ) ) ) { recordId = rowIdELEval . eval ( elVars , conf . rowIdExpression , String . class ) ; if ( StringUtils . isEmpty ( recordId ) ) { throw new OnRecordErrorException ( record , Errors . BIGQUERY_15 ) ; } } } catch ( ELEvalException e ) { LOG . error ( "Error evaluating Row Expression EL" , e ) ; throw new OnRecordErrorException ( record , Errors . BIGQUERY_10 , e ) ; } return recordId ;


signature:
public class BigQueryTarget { /** * Convert the root field to a java map object implicitly mapping each field to the column ( only non nested objects )
 * @ param record record to be converted
 * @ return Java row representation for the record */
private Map < String , Object > convertToRowObjectFromRecord ( Record record ) throws OnRecordErrorException { } }

implementation:
Field rootField = record . get ( ) ; Map < String , Object > rowObject = new LinkedHashMap < > ( ) ; if ( rootField . getType ( ) . isOneOf ( Field . Type . MAP , Field . Type . LIST_MAP ) ) { Map < String , Field > fieldMap = rootField . getValueAsMap ( ) ; for ( Map . Entry < String , Field > fieldEntry : fieldMap . entrySet ( ) ) { Field field = fieldEntry . getValue ( ) ; // Skip null value fields
if ( field . getValue ( ) != null ) { try { rowObject . put ( fieldEntry . getKey ( ) , getValueFromField ( "/" + fieldEntry . getKey ( ) , field ) ) ; } catch ( IllegalArgumentException e ) { throw new OnRecordErrorException ( record , Errors . BIGQUERY_13 , e . getMessage ( ) ) ; } } } } else { throw new OnRecordErrorException ( record , Errors . BIGQUERY_16 ) ; } return rowObject ;


signature:
public class BigQueryTarget { /** * Convert the sdc Field to an object for row content */
private Object getValueFromField ( String fieldPath , Field field ) { } }

implementation:
LOG . trace ( "Visiting Field Path '{}' of type '{}'" , fieldPath , field . getType ( ) ) ; switch ( field . getType ( ) ) { case LIST : // REPEATED
List < Field > listField = field . getValueAsList ( ) ; // Convert the list to map with indices as key and Field as value ( Map < Integer , Field > )
Map < Integer , Field > fields = IntStream . range ( 0 , listField . size ( ) ) . boxed ( ) . collect ( Collectors . toMap ( Function . identity ( ) , listField :: get ) ) ; // filter map to remove fields with null value
fields = fields . entrySet ( ) . stream ( ) . filter ( e -> e . getValue ( ) . getValue ( ) != null ) . collect ( Collectors . toMap ( Map . Entry :: getKey , Map . Entry :: getValue ) ) ; // now use the map index to generate field path and generate object for big query write
return fields . entrySet ( ) . stream ( ) . map ( e -> getValueFromField ( fieldPath + "[" + e . getKey ( ) + "]" , e . getValue ( ) ) ) . collect ( Collectors . toList ( ) ) ; case MAP : case LIST_MAP : // RECORD
return field . getValueAsMap ( ) . entrySet ( ) . stream ( ) . filter ( me -> me . getValue ( ) . getValue ( ) != null ) . collect ( Collectors . toMap ( Map . Entry :: getKey , e -> getValueFromField ( fieldPath + "/" + e . getKey ( ) , e . getValue ( ) ) ) ) ; case DATE : return dateFormat . format ( field . getValueAsDate ( ) ) ; case TIME : return timeFormat . format ( field . getValueAsTime ( ) ) ; case DATETIME : return dateTimeFormat . format ( field . getValueAsDatetime ( ) ) ; case BYTE_ARRAY : return Base64 . getEncoder ( ) . encodeToString ( field . getValueAsByteArray ( ) ) ; case DECIMAL : case BYTE : case CHAR : case FILE_REF : throw new IllegalArgumentException ( Utils . format ( Errors . BIGQUERY_12 . getMessage ( ) , fieldPath , field . getType ( ) ) ) ; default : // Boolean - > Map to Boolean in big query
// Float , Double - > Map to Float in big query
// String - > maps to String in big query
// Short , Integer , Long - > Map to integer in big query
return field . getValue ( ) ; }


signature:
public class WebServerTask { /** * port . */
private void checkValidPorts ( ) { } }

implementation: if ( ( conf . get ( HTTP_PORT_KEY , HTTP_PORT_DEFAULT ) == 0 && conf . get ( HTTPS_PORT_KEY , HTTPS_PORT_DEFAULT ) != - 1 ) || ( conf . get ( HTTPS_PORT_KEY , HTTPS_PORT_DEFAULT ) == 0 && conf . get ( HTTP_PORT_KEY , HTTP_PORT_DEFAULT ) != - 1 ) ) { throw new IllegalArgumentException ( "Invalid port combination for http and https, If http port is set to 0 (random), then https should be " + "set to -1 or vice versa" ) ; }


signature:
public class CouchbaseProcessor { /** * Aggregates errors that occur during the processing of a batch .
 * @ param record the record being written
 * @ param error the error encountered while writing the record
 * @ param ex the exception encountered while writing the record
 * @ param passable whether the record can be optionally passed down the pipeline
 * @ return an empty observable */
private < T > Observable < T > handleError ( Record record , Errors error , Throwable ex , boolean passable ) { } }

implementation: errors . put ( record , new ErrorRecord ( error , ex , passable ) ) ; return Observable . empty ( ) ;


signature:
public class CouchbaseProcessor { /** * Aggregates errors that occur during the processing of a batch .
 * @ param record the record being written
 * @ param error the error encountered while writing the record
 * @ param passable whether the record can be optionally passed down the pipeline
 * @ return an empty observable */
private < T > Observable < T > handleError ( Record record , Errors error , boolean passable ) { } }

implementation: return handleError ( record , error , new RuntimeException ( ) , passable ) ;


signature:
public class CouchbaseProcessor { /** * Writes sub - document lookup values to the record
 * @ param record the record being processed
 * @ param frag the sub - document fragment from Couchbase
 * @ return an Observable of the updated record */
private Observable < Record > setFragmentInRecord ( Record record , DocumentFragment < Lookup > frag ) { } }

implementation: if ( frag . content ( 0 ) == null ) { LOG . debug ( "Sub-document path not found" ) ; return handleError ( record , Errors . COUCHBASE_25 , true ) ; } for ( SubdocMappingConfig subdocMapping : config . subdocMappingConfigs ) { Object fragJson = frag . content ( subdocMapping . subdocPath ) ; if ( fragJson == null ) { return handleError ( record , Errors . COUCHBASE_25 , true ) ; } try { record . set ( subdocMapping . sdcField , jsonToField ( fragJson ) ) ; record . getHeader ( ) . setAttribute ( config . CAS_HEADER_ATTRIBUTE , String . valueOf ( frag . cas ( ) ) ) ; } catch ( IOException e ) { try { record . set ( subdocMapping . sdcField , jsonToField ( JsonObject . fromJson ( fragJson . toString ( ) ) . toMap ( ) ) ) ; record . getHeader ( ) . setAttribute ( config . CAS_HEADER_ATTRIBUTE , String . valueOf ( frag . cas ( ) ) ) ; } catch ( IOException ex ) { return handleError ( record , Errors . COUCHBASE_19 , ex , false ) ; } } } return Observable . just ( record ) ;


signature:
public class CouchbaseProcessor { /** * Writes full document lookup values to the record
 * @ param record the record being processed
 * @ param doc the JsonDocument from Couchbase
 * @ return an Observable of the updated record */
private Observable < Record > setDocumentInRecord ( Record record , JsonDocument doc ) { } }

implementation: if ( doc . content ( ) == null ) { LOG . debug ( "Document does not exist: {}" , doc . id ( ) ) ; return handleError ( record , Errors . COUCHBASE_26 , true ) ; } try { record . set ( config . outputField , jsonToField ( doc . content ( ) . toMap ( ) ) ) ; record . getHeader ( ) . setAttribute ( config . CAS_HEADER_ATTRIBUTE , String . valueOf ( doc . cas ( ) ) ) ; return Observable . just ( record ) ; } catch ( IOException e ) { LOG . debug ( "Unable to set KV lookup in record for: {}" , doc . id ( ) ) ; return handleError ( record , Errors . COUCHBASE_19 , e , false ) ; }


signature:
public class CouchbaseProcessor { /** * Writes N1QL query result rows to the record
 * @ param record the record being processed
 * @ param row the current row from the N1QL query result
 * @ return an Observable of the updated record */
private Observable < Record > setN1QLRowInRecord ( Record record , AsyncN1qlQueryRow row ) { } }

implementation: for ( N1QLMappingConfig n1qlMapping : config . n1qlMappingConfigs ) { if ( config . multipleValueOperation == MultipleValueType . FIRST && record . get ( n1qlMapping . sdcField ) != null ) { LOG . debug ( "Only populating output field with first record. Skipping additional result." ) ; return Observable . empty ( ) ; } Object property = row . value ( ) . get ( n1qlMapping . property ) ; if ( property == null ) { LOG . debug ( "Requested property not returned: {}" , n1qlMapping . property ) ; return handleError ( record , Errors . COUCHBASE_27 , true ) ; } try { record . set ( n1qlMapping . sdcField , jsonToField ( property ) ) ; } catch ( IOException e ) { try { record . set ( n1qlMapping . sdcField , jsonToField ( JsonObject . fromJson ( property . toString ( ) ) . toMap ( ) ) ) ; } catch ( IOException ex ) { LOG . debug ( "Unable to set N1QL property in record" ) ; return handleError ( record , Errors . COUCHBASE_19 , ex , false ) ; } } } return Observable . just ( record ) ;


signature:
public class MSOperationCode { /** * Take an numeric operation code and check if the number is
 * valid operation code .
 * The operation code must be numeric : 1 ( insert ) , 2 ( update ) , 3 ( delete ) , etc ,
 * @ param op Numeric operation code in String
 * @ return Operation code in int , - 1 if invalid number */
public static int convertToJDBCCode ( int op ) { } }

implementation: if ( CRUD_MAP . containsKey ( op ) ) { return CRUD_MAP . get ( op ) ; } throw new UnsupportedOperationException ( Utils . format ( "Operation code {} is not supported" , op ) ) ;


signature:
public class AmazonS3Runnable { /** * as the record is just the metadata along with file ref . */
private void handleWholeFileDataFormat ( S3ObjectSummary s3ObjectSummary , String recordId ) throws StageException { } }

implementation:
S3Object partialS3ObjectForMetadata ; // partialObject with fetchSize 1 byte .
// This is mostly used for extracting metadata and such .
partialS3ObjectForMetadata = AmazonS3Util . getObjectRange ( s3Client , s3ConfigBean . s3Config . bucket , s3ObjectSummary . getKey ( ) , 1 , s3ConfigBean . sseConfig . useCustomerSSEKey , s3ConfigBean . sseConfig . customerKey , s3ConfigBean . sseConfig . customerKeyMd5 ) ; S3FileRef . Builder s3FileRefBuilder = new S3FileRef . Builder ( ) . s3Client ( s3Client ) . s3ObjectSummary ( s3ObjectSummary ) . useSSE ( s3ConfigBean . sseConfig . useCustomerSSEKey ) . customerKey ( s3ConfigBean . sseConfig . customerKey ) . customerKeyMd5 ( s3ConfigBean . sseConfig . customerKeyMd5 ) . bufferSize ( ( int ) dataParser . suggestedWholeFileBufferSize ( ) ) . createMetrics ( true ) . totalSizeInBytes ( s3ObjectSummary . getSize ( ) ) . rateLimit ( dataParser . wholeFileRateLimit ( ) ) ; if ( dataParser . isWholeFileChecksumRequired ( ) ) { s3FileRefBuilder . verifyChecksum ( true ) . checksumAlgorithm ( HashingUtil . HashType . MD5 ) // 128 bit hex encoded md5 checksum .
. checksum ( partialS3ObjectForMetadata . getObjectMetadata ( ) . getETag ( ) ) ; } Map < String , Object > metadata = AmazonS3Util . getMetaData ( partialS3ObjectForMetadata ) ; metadata . put ( S3Constants . BUCKET , s3ObjectSummary . getBucketName ( ) ) ; metadata . put ( S3Constants . OBJECT_KEY , s3ObjectSummary . getKey ( ) ) ; metadata . put ( S3Constants . OWNER , s3ObjectSummary . getOwner ( ) ) ; metadata . put ( S3Constants . SIZE , s3ObjectSummary . getSize ( ) ) ; metadata . put ( HeaderAttributeConstants . FILE_NAME , s3ObjectSummary . getKey ( ) ) ; metadata . remove ( S3Constants . CONTENT_LENGTH ) ; parser = dataParser . getParser ( recordId , metadata , s3FileRefBuilder . build ( ) ) ; // Object is assigned so that setHeaders ( ) function can use this to get metadata
// information about the object
object = partialS3ObjectForMetadata ;


signature:
public class AmazonS3Runnable { /** * Initialize the gauge with needed information */
private void initGaugeIfNeeded ( ) { } }

implementation: gaugeMap . put ( S3Constants . THREAD_NAME , Thread . currentThread ( ) . getName ( ) ) ; gaugeMap . put ( S3Constants . STATUS , "" ) ; gaugeMap . put ( S3Constants . BUCKET , "" ) ; gaugeMap . put ( S3Constants . OBJECT_KEY , "" ) ;


signature:
public class GtidSourceOffset { /** * Check if given gtid + seqNo pair is contained in this incomplete transactions .
 * @ param gtid
 * @ param seqNo
 * @ return true if incomplete transactions have given gtid and
 * its associated seqNo is greater or equal to given seqNo . */
public boolean incompleteTransactionsContain ( String gtid , long seqNo ) { } }

implementation: Long s = incompleteTransactions . get ( gtid ) ; return s != null && s >= seqNo ;


signature:
public class LambdaUtil { /** * Runs a Supplier within the context of a specified ClassLoader .
 * @ param classLoader the ClassLoader to run the Supplier .
 * @ param supplier the Supplier to run within the context of a specified ClassLoader . */
public static < T > T withClassLoader ( ClassLoader classLoader , ExceptionSupplier < T > supplier ) { } }

implementation: try { return withClassLoaderInternal ( classLoader , supplier ) ; } catch ( Exception e ) { Throwables . propagate ( e ) ; } return null ;


signature:
public class LambdaUtil { /** * Runs a Supplier within the context of a specified ClassLoader .
 * @ param classLoader the ClassLoader to run the Supplier .
 * @ param supplier the Supplier to run within the context of a specified ClassLoader . */
public static < T , E1 extends Exception > T withClassLoader ( ClassLoader classLoader , Class < E1 > e1 , ExceptionSupplier < T > supplier ) throws E1 { } }

implementation: try { return withClassLoaderInternal ( classLoader , supplier ) ; } catch ( Exception e ) { Throwables . propagateIfPossible ( e , e1 ) ; Throwables . propagate ( e ) ; } return null ;


signature:
public class LambdaUtil { /** * Runs a Supplier within the context of a specified ClassLoader and in priviledged mode .
 * @ param classLoader the ClassLoader to run the Supplier .
 * @ param supplier the Supplier to run within the context of a specified ClassLoader . */
public static < T > T privilegedWithClassLoader ( ClassLoader classLoader , ExceptionSupplier < T > supplier ) { } }

implementation: try { return AccessController . doPrivileged ( ( PrivilegedExceptionAction < T > ) ( ) -> withClassLoaderInternal ( classLoader , supplier ) ) ; } catch ( PrivilegedActionException e ) { Throwables . propagate ( e ) ; } return null ;


signature:
public class LambdaUtil { /** * Runs a Supplier within the context of a specified ClassLoader and in priviledged mode .
 * @ param classLoader the ClassLoader to run the Supplier .
 * @ param e1 Exception class that should be propagated as - is
 * @ param supplier the Supplier to run within the context of a specified ClassLoader . */
public static < T , E1 extends Exception > T privilegedWithClassLoader ( ClassLoader classLoader , Class < E1 > e1 , ExceptionSupplier < T > supplier ) throws E1 { } }

implementation: try { return AccessController . doPrivileged ( ( PrivilegedExceptionAction < T > ) ( ) -> withClassLoaderInternal ( classLoader , supplier ) ) ; } catch ( PrivilegedActionException e ) { Throwables . propagateIfPossible ( e . getCause ( ) , e1 ) ; Throwables . propagate ( e ) ; } return null ;


signature:
public class LambdaUtil { /** * Internal version of the wrapping function that will simply propagate all exceptions up . */
private static < T > T withClassLoaderInternal ( ClassLoader classLoader , ExceptionSupplier < T > supplier ) throws Exception { } }

implementation: ClassLoader previousClassLoader = Thread . currentThread ( ) . getContextClassLoader ( ) ; try { Thread . currentThread ( ) . setContextClassLoader ( classLoader ) ; return supplier . get ( ) ; } finally { Thread . currentThread ( ) . setContextClassLoader ( previousClassLoader ) ; }


signature:
public class HeaderImpl { /** * HeaderImpl setter methods */
public void setStageCreator ( String stateCreator ) { } }

implementation: Preconditions . checkNotNull ( stateCreator , "stateCreator cannot be null" ) ; map . put ( STAGE_CREATOR_INSTANCE_ATTR , stateCreator ) ;


signature:
public class HeaderImpl { /** * To be removed */
public Map < String , Object > getUserAttributes ( ) { } }

implementation: return map . entrySet ( ) . stream ( ) . filter ( map -> ! map . getKey ( ) . startsWith ( RESERVED_PREFIX ) ) . collect ( Collectors . toMap ( map -> map . getKey ( ) , map -> map . getValue ( ) ) ) ;


signature:
public class HeaderImpl { /** * To be removed */
public Map < String , Object > setUserAttributes ( Map < String , Object > newAttributes ) { } }

implementation:
// ImmutableMap can ' t have null values and our map could have , so use unmodifiable map
Map < String , Object > old = Collections . unmodifiableMap ( getUserAttributes ( ) ) ; // Set current map to just the Reserved System Attributes
map = getSystemAttributes ( ) ; // Add and validate each of the new user attributes
newAttributes . forEach ( ( k , v ) -> setAttribute ( k , v . toString ( ) ) ) ; return old ;


signature:
public class SecurityContext { /** * Logs in . If Kerberos is enabled it logs in against the KDC , otherwise is a NOP . */
public synchronized void login ( ) { } }

implementation:
if ( subject != null ) { throw new IllegalStateException ( Utils . format ( "Service already login, Principal '{}'" , subject . getPrincipals ( ) ) ) ; } if ( securityConfiguration . isKerberosEnabled ( ) ) { try { loginContext = createLoginContext ( ) ; subject = loginContext . getSubject ( ) ; } catch ( Exception ex ) { throw new RuntimeException ( Utils . format ( "Could not get Kerberos credentials: {}" , ex . toString ( ) ) , ex ) ; } if ( renewalThread == null ) { renewalThread = new Thread ( ) { @ Override public void run ( ) { LOG . debug ( "Starting renewal thread" ) ; if ( ! SecurityContext . this . sleep ( THIRTY_SECONDS_MS ) ) { LOG . info ( "Interrupted, exiting renewal thread" ) ; return ; } while ( true ) { LOG . trace ( "Renewal check starts" ) ; try { KerberosTicket lastExpiringTGT = getNewestTGT ( ) ; if ( lastExpiringTGT == null ) { LOG . warn ( "Could not obtain kerberos ticket, it may have expired already or it was logged out, will wait" + "30 secs to attempt a relogin" ) ; LOG . trace ( "Ticket not found, sleeping 30 secs and trying to login" ) ; if ( ! SecurityContext . this . sleep ( THIRTY_SECONDS_MS ) ) { LOG . info ( "Interrupted, exiting renewal thread" ) ; return ; } } else { long renewalTimeMs = calculateRenewalTime ( lastExpiringTGT ) - THIRTY_SECONDS_MS ; LOG . trace ( "Ticket found time to renewal '{}ms', sleeping that time" , renewalTimeMs ) ; if ( renewalTimeMs > 0 ) { if ( ! SecurityContext . this . sleep ( renewalTimeMs ) ) { LOG . info ( "Interrupted, exiting renewal thread" ) ; return ; } } } LOG . debug ( "Triggering relogin" ) ; Set < KerberosTicket > oldTickets = getSubject ( ) . getPrivateCredentials ( KerberosTicket . class ) ; relogin ( 3 ) ; // Remove all old private credentials , since we only need the new one we just added
getSubject ( ) . getPrivateCredentials ( ) . removeAll ( oldTickets ) ; } catch ( Exception exception ) { LOG . error ( "Stopping renewal thread because of exception: " + exception , exception ) ; return ; } catch ( Throwable throwable ) { LOG . error ( "Error in renewal thread: " + throwable , throwable ) ; return ; } } } } ; List < String > principals = new ArrayList < > ( ) ; for ( Principal p : subject . getPrincipals ( ) ) { principals . add ( p . getName ( ) ) ; } renewalThread . setName ( "Kerberos-Renewal-Thread-" + Joiner . on ( "," ) . join ( principals ) ) ; renewalThread . setContextClassLoader ( Thread . currentThread ( ) . getContextClassLoader ( ) ) ; renewalThread . setDaemon ( true ) ; renewalThread . start ( ) ; } } else { subject = new Subject ( ) ; } LOG . debug ( "Login. Kerberos enabled '{}', Principal '{}'" , securityConfiguration . isKerberosEnabled ( ) , subject . getPrincipals ( ) ) ;


signature:
public class SecurityContext { /** * Logs out . If Keberos is enabled it logs out from the KDC , otherwise is a NOP . */
public synchronized void logout ( ) { } }

implementation: if ( subject != null ) { LOG . debug ( "Logout. Kerberos enabled '{}', Principal '{}'" , securityConfiguration . isKerberosEnabled ( ) , subject . getPrincipals ( ) ) ; if ( loginContext != null ) { try { loginContext . logout ( ) ; } catch ( LoginException ex ) { LOG . warn ( "Error while doing logout from Kerberos: {}" , ex . toString ( ) , ex ) ; } finally { loginContext = null ; } } subject = null ; }


signature:
public class SdcSecurityManager { /** * This method should be called only once and before any stages are loaded . */
private void setExceptions ( Configuration configuration ) { } }

implementation:
this . exceptions . clear ( ) ; this . stageLibExceptions . clear ( ) ; // Load general exceptions
for ( String path : configuration . get ( PROPERTY_EXCEPTIONS , "" ) . split ( "," ) ) { this . exceptions . add ( replaceVariables ( path ) ) ; } // Load Stage library specific exceptions
Configuration stageSpecific = configuration . getSubSetConfiguration ( PROPERTY_STAGE_EXCEPTIONS , true ) ; for ( Map . Entry < String , String > entry : stageSpecific . getValues ( ) . entrySet ( ) ) { Set < String > stageExceptions = new HashSet < > ( ) ; for ( String path : entry . getValue ( ) . split ( "," ) ) { stageExceptions . add ( replaceVariables ( path ) ) ; } this . stageLibExceptions . put ( entry . getKey ( ) , stageExceptions ) ; }


signature:
public class SdcSecurityManager { /** * Replace variables to internal SDC directories so that users don ' t have to be entering FQDN . */
private String replaceVariables ( String path ) { } }

implementation: return path . replace ( "$SDC_DATA" , dataDir ) . replace ( "$SDC_CONF" , configDir ) . replace ( "$SDC_RESOURCES" , resourcesDir ) ;


signature:
public class SdcSecurityManager { /** * Make sure that the active code have proper rights to access the file inside protected directory . */
private void ensureProperPermissions ( String path ) { } }

implementation:
ClassLoader cl = Thread . currentThread ( ) . getContextClassLoader ( ) ; // 1 ) Container can access anything
if ( cl instanceof ContainerClassLoader ) { return ; } // 2 . Some files are whitelisted globally for all stage libraries
if ( exceptions . contains ( path ) ) { return ; } // 3 . Some stage libraries have some files whitelisted globally
if ( cl instanceof SDCClassLoader ) { String libraryName = ( ( SDCClassLoader ) cl ) . getName ( ) ; if ( stageLibExceptions . containsKey ( libraryName ) && stageLibExceptions . get ( libraryName ) . contains ( path ) ) { return ; } } // No whitelist , no fun , go away
throw new SecurityException ( Utils . format ( "Classloader {} is not allowed access to Data Collector internal directories ({})." , cl . toString ( ) , path ) ) ;


signature:
public class JdbcTypeSupport { /** * Generate Hive Type Info Representation inside the Metadata Record .
 * @ param jdbcTypeInfo { @ link JdbcTypeInfo }
 * @ return { @ link Field } which is the field in the metadata record which represents { @ link JdbcTypeInfo } */
public Field generateJdbcTypeInfoFieldForMetadataRecord ( JdbcTypeInfo jdbcTypeInfo ) { } }

implementation: Map < String , Field > fields = new HashMap < > ( ) ; fields . put ( TYPE , Field . create ( jdbcTypeInfo . getJdbcType ( ) . name ( ) ) ) ; fields . put ( EXTRA_INFO , generateExtraInfoFieldForMetadataRecord ( jdbcTypeInfo ) ) ; return Field . create ( fields ) ;


signature:
public class JdbcTypeSupport { /** * Generate { @ link JdbcTypeInfo } from the Metadata Record < br > .
 * ( Reverse of { @ link # generateJdbcTypeInfoFieldForMetadataRecord ( JdbcTypeInfo ) } )
 * @ throws StageException if the metadata field is not valid .
 * @ param jdbcTypeInfoField
 * @ return { @ link JdbcTypeInfo }
 * @ throws StageException If the record has invalid */
@ SuppressWarnings ( "unchecked" ) public JdbcTypeInfo generateJdbcTypeInfoFromMetadataField ( Field jdbcTypeInfoField , JdbcSchemaWriter schemaWriter ) throws StageException { } }

implementation: if ( jdbcTypeInfoField . getType ( ) == Field . Type . MAP ) { Map < String , Field > fields = ( Map < String , Field > ) jdbcTypeInfoField . getValue ( ) ; if ( ! fields . containsKey ( TYPE ) || ! fields . containsKey ( EXTRA_INFO ) ) { throw new StageException ( JdbcErrors . JDBC_308 , TYPE_INFO ) ; } JdbcType jdbcType = JdbcType . getJdbcTypeFromString ( fields . get ( TYPE ) . getValueAsString ( ) ) ; return generateJdbcTypeInfoFromMetadataField ( jdbcType , fields . get ( EXTRA_INFO ) , schemaWriter ) ; } else { throw new StageException ( JdbcErrors . JDBC_308 , TYPE_INFO ) ; }


signature:
public class JdbcTypeSupport { /** * Generate Column Definition for create table / add columns
 * @ param jdbcTypeInfo { @ link JdbcTypeInfo }
 * @ param columnName Column Name
 * @ return Column definition */
public String generateColumnTypeDefinition ( JdbcTypeInfo jdbcTypeInfo , String columnName ) { } }

implementation: return String . format ( COLUMN_TYPE , columnName , jdbcTypeInfo . toString ( ) ) ;


signature:
public class BootstrapCluster { /** * Obtaining a reference on the dummy source which is used to feed a pipeline < br / >
 * Direction : Stage - > Container
 * @ param postBatchRunnable runnable to run after each batch is finished
 * @ return a source object associated with the newly created pipeline
 * @ throws Exception */
public static /* PipelineStartResult */
Object startPipeline ( Runnable postBatchRunnable ) throws Exception { } }

implementation: BootstrapCluster . initialize ( ) ; ClassLoader originalClassLoader = Thread . currentThread ( ) . getContextClassLoader ( ) ; try { Thread . currentThread ( ) . setContextClassLoader ( containerCL ) ; Class embeddedPipelineFactoryClz = Class . forName ( "com.streamsets.datacollector.EmbeddedDataCollectorFactory" , true , containerCL ) ; Method createPipelineMethod = embeddedPipelineFactoryClz . getMethod ( "startPipeline" , Runnable . class ) ; return createPipelineMethod . invoke ( null , postBatchRunnable ) ; } catch ( Exception ex ) { String msg = "Error trying to create pipeline: " + ex ; throw new IllegalStateException ( msg , ex ) ; } finally { Thread . currentThread ( ) . setContextClassLoader ( originalClassLoader ) ; }


signature:
public class BootstrapEmrBatch { /** * Bootstrapping the Driver which starts a EMR job on cluster */
public static void main ( String [ ] args ) throws Exception { } }

implementation:
EmrBinding binding = null ; try { binding = new EmrBinding ( args ) ; binding . init ( ) ; binding . awaitTermination ( ) ; // killed by ClusterProviderImpl before returning
} catch ( Exception ex ) { String msg = "Error trying to invoke BootstrapEmrBatch.main: " + ex ; throw new IllegalStateException ( msg , ex ) ; } finally { try { if ( binding != null ) { binding . close ( ) ; } } catch ( Exception ex ) { LOG . warn ( "Error on binding close: " + ex , ex ) ; } }


signature:
public class RecordWriterManager { /** * Returns directory path for given record and date . */
String getDirPath ( Date date , Record record ) throws StageException { } }

implementation:
if ( dirPathTemplateInHeader ) { // We ' re not validating if the header exists as that job is already done
return record . getHeader ( ) . getAttribute ( HdfsTarget . TARGET_DIRECTORY_HEADER ) ; } return pathResolver . resolvePath ( date , record ) ;


signature:
public class RecordWriterManager { /** * This method should be called every time we finish writing into a file and consider it " done " . */
Path renameToFinalName ( FileSystem fs , Path tempPath ) throws IOException , StageException { } }

implementation: return fsHelper . renameAndGetPath ( fs , tempPath ) ;


signature:
public class RecordWriterManager { /** * Produce events that were cached during the batch processing . */
public void issueCachedEvents ( ) throws IOException { } }

implementation: Path closedPath ; while ( ( closedPath = closedPaths . poll ( ) ) != null ) { produceCloseFileEvent ( fs , closedPath ) ; }


signature:
public class RecordWriterManager { /** * rename all _ tmp _ files under directory path ( dirPathTemplate )
 * return the number of _ tmp _ files */
public int handleAlreadyExistingFiles ( ) throws StageException , IOException { } }

implementation: int result = 0 ; String globPath = dirPathTemplate ; final String staticExpReg = "\\$\\{(sdc:|pipeline:|runtime:)[a-zA-Z0-9\\(\\)]*\\}" ; Pattern pattern = Pattern . compile ( staticExpReg ) ; Matcher matcher = pattern . matcher ( globPath ) ; ELEval eval = context . createELEval ( "dirPathTemplate" ) ; ELVars vars = context . createELVars ( ) ; while ( matcher . find ( ) ) { String expressionString = eval . eval ( vars , matcher . group ( ) , String . class ) ; globPath = globPath . replace ( matcher . group ( ) , expressionString ) ; } final String expReg = "\\$\\{[^}]*\\}" ; pattern = Pattern . compile ( expReg ) ; matcher = pattern . matcher ( globPath ) ; while ( matcher . find ( ) ) { globPath = globPath . replace ( matcher . group ( ) , "*" ) ; } globPath = globPath . replaceAll ( "\\*+" , "*" ) ; globPath = globPath + "/" + TMP_FILE_PREFIX + uniquePrefix + "*" ; LOG . info ( "Created the following glob path for file recovery: {}" , globPath ) ; FileStatus [ ] fileStatuses = fs . globStatus ( new Path ( globPath ) ) ; for ( FileStatus fileStatus : fileStatuses ) { fsHelper . handleAlreadyExistingFile ( fs , fileStatus . getPath ( ) ) ; result ++ ; } return result ;


signature:
public class RecordWriterManager { /** * This method must always be called after the closeLock ( ) method on the writer has been called . */
public Path commitWriter ( RecordWriter writer ) throws IOException , StageException { } }

implementation:
Path path = null ; if ( ( ! writer . isClosed ( ) || writer . isIdleClosed ( ) ) && ! writer . isRenamed ( ) ) { // Unset the interrupt flag before close ( ) . InterruptedIOException makes close ( ) fail
// resulting that the tmp file never gets renamed when stopping the pipeline .
boolean interrupted = Thread . interrupted ( ) ; try { // Since this method is always called from exactly one thread , and
// we checked to make sure that it was not closed or it was idle closed , this method either closes
// the file or pushes us into the catch block .
writer . close ( ) ; } catch ( IdleClosedException e ) { LOG . info ( "Writer for {} was idle closed, renaming.." , writer . getPath ( ) ) ; } LOG . debug ( "Path[{}] - Committing Writer" , writer . getPath ( ) ) ; path = renameToFinalName ( fs , writer . getPath ( ) ) ; writer . setRenamed ( true ) ; LOG . debug ( "Path[{}] - Committed Writer to '{}'" , writer . getPath ( ) , path ) ; // Reset the interrupt flag back .
if ( interrupted ) { Thread . currentThread ( ) . interrupt ( ) ; } } return path ;


signature:
public class RecordWriterManager { /** * Return true if this record should be written into a new file regardless whether we have a file for the record
 * currently opened or not . */
public boolean shouldRoll ( RecordWriter writer , Record record ) { } }

implementation: if ( rollIfHeader && record . getHeader ( ) . getAttribute ( rollHeaderName ) != null ) { LOG . debug ( "Path[{}] - will be rolled because of roll attribute '{}' set to '{}' in the record : '{}'" , writer . getPath ( ) , rollHeaderName , record . getHeader ( ) . getAttribute ( rollHeaderName ) , record . getHeader ( ) . getSourceId ( ) ) ; return true ; } return false ;


signature:
public class FileTailSource { /** * When we start with a file ( empty or not ) the file offset is zero .
 * If the file is a rolled file , the file will be EOF immediately triggering a close of the reader and setting the
 * offset to Long . MAX _ VALUE ( this happens in the MultiDirectoryReader class ) . This is the signal that in the next
 * read a directory scan should be triggered to get the next rolled file or the live file if we were scanning the last
 * rolled file .
 * If the file you are starting is the live file , we don ' t get an EOF as we expect data to be appended . We just return
 * null chunks while there is no data . If the file is rolled we ' ll detect that and then do what is described in the
 * previous paragraph .
 * When offset for file is " " it means we never processed things in the directory , at that point we start from the
 * first file ( according to the defined order ) in the directory , or if a ' first file ' as been set in the configuration ,
 * we start from that file .
 * We encode in lastSourceOffset the current file and offset from all directories in JSON . */
@ Override public String produce ( String lastSourceOffset , int maxBatchSize , BatchMaker batchMaker ) throws StageException { } }

implementation:
int recordCounter = 0 ; long startTime = System . currentTimeMillis ( ) ; maxBatchSize = Math . min ( conf . batchSize , maxBatchSize ) ; // deserializing offsets of all directories
Map < String , String > offsetMap = deserializeOffsetMap ( lastSourceOffset ) ; boolean offsetSet = false ; while ( ! offsetSet ) { try { multiDirReader . setOffsets ( offsetMap ) ; offsetSet = true ; } catch ( IOException ex ) { LOG . warn ( "Error while creating reading previous offset: {}" , ex . toString ( ) , ex ) ; multiDirReader . purge ( ) ; } } while ( recordCounter < maxBatchSize && ! isTimeout ( startTime ) ) { LiveFileChunk chunk = multiDirReader . next ( getRemainingWaitTime ( startTime ) ) ; if ( chunk != null ) { String tag = chunk . getTag ( ) ; tag = ( tag != null && tag . isEmpty ( ) ) ? null : tag ; String liveFileStr = chunk . getFile ( ) . serialize ( ) ; List < FileLine > lines = chunk . getLines ( ) ; int truncatedLine = chunk . isTruncated ( ) ? lines . size ( ) - 1 : - 1 ; for ( int i = 0 ; i < lines . size ( ) ; i ++ ) { FileLine line = lines . get ( i ) ; String sourceId = liveFileStr + "::" + line . getFileOffset ( ) ; try ( DataParser parser = parserFactory . getParser ( sourceId , line . getText ( ) ) ) { if ( i == truncatedLine ) { // set truncated
parser . setTruncated ( ) ; } Record record = parser . parse ( ) ; if ( record != null ) { if ( tag != null ) { record . getHeader ( ) . setAttribute ( "tag" , tag ) ; } record . getHeader ( ) . setAttribute ( HeaderAttributeConstants . FILE , chunk . getFile ( ) . getPath ( ) . toString ( ) ) ; record . getHeader ( ) . setAttribute ( HeaderAttributeConstants . FILE_NAME , chunk . getFile ( ) . getPath ( ) . getFileName ( ) . toString ( ) ) ; record . getHeader ( ) . setAttribute ( HeaderAttributeConstants . OFFSET , String . valueOf ( line . getFileOffset ( ) ) ) ; record . getHeader ( ) . setAttribute ( HeaderAttributeConstants . LAST_MODIFIED_TIME , String . valueOf ( Files . getLastModifiedTime ( chunk . getFile ( ) . getPath ( ) ) . toMillis ( ) ) ) ; batchMaker . addRecord ( record , outputLane ) ; recordCounter ++ ; } } catch ( IOException | DataParserException ex ) { errorRecordHandler . onError ( Errors . TAIL_12 , sourceId , ex . toString ( ) , ex ) ; } } } } boolean metadataGenerationFailure = false ; Date now = new Date ( startTime ) ; for ( FileEvent event : multiDirReader . getEvents ( ) ) { try { LiveFile file = event . getFile ( ) . refresh ( ) ; Record metadataRecord = getContext ( ) . createRecord ( "" ) ; Map < String , Field > map = new HashMap < > ( ) ; map . put ( "fileName" , Field . create ( file . getPath ( ) . toString ( ) ) ) ; map . put ( "inode" , Field . create ( file . getINode ( ) ) ) ; map . put ( "time" , Field . createDate ( now ) ) ; map . put ( "event" , Field . create ( ( event . getAction ( ) . name ( ) ) ) ) ; metadataRecord . set ( Field . create ( map ) ) ; batchMaker . addRecord ( metadataRecord , metadataLane ) ; // We ' re also sending the same information on event lane
String eventRecordSourceId = Utils . format ( "event:{}:{}:{}" , event . getAction ( ) . name ( ) , 1 , file . getPath ( ) . toString ( ) ) ; EventRecord eventRecord = getContext ( ) . createEventRecord ( event . getAction ( ) . name ( ) , 1 , eventRecordSourceId ) ; eventRecord . set ( Field . create ( map ) ) ; getContext ( ) . toEvent ( eventRecord ) ; } catch ( IOException ex ) { LOG . warn ( "Error while creating metadata records: {}" , ex . toString ( ) , ex ) ; metadataGenerationFailure = true ; } } if ( metadataGenerationFailure ) { multiDirReader . purge ( ) ; } boolean offsetExtracted = false ; while ( ! offsetExtracted ) { try { offsetMap = multiDirReader . getOffsets ( ) ; offsetExtracted = true ; } catch ( IOException ex ) { LOG . warn ( "Error while creating creating new offset: {}" , ex . toString ( ) , ex ) ; multiDirReader . purge ( ) ; } } // Calculate Offset lag Metric .
calculateOffsetLagMetric ( offsetMap ) ; // Calculate Pending Files Metric
calculatePendingFilesMetric ( ) ; // serializing offsets of all directories
return serializeOffsetMap ( offsetMap ) ;


signature:
public class AntPathMatcher { /** * Set the path separator to use for pattern parsing .
 * < p > Default is " / " , as in Ant . */
public void setPathSeparator ( String pathSeparator ) { } }

implementation: this . pathSeparator = ( pathSeparator != null ? pathSeparator : DEFAULT_PATH_SEPARATOR ) ; this . pathSeparatorPatternCache = new PathSeparatorPatternCache ( this . pathSeparator ) ;


signature:
public class AntPathMatcher { /** * Tokenize the given path pattern into parts , based on this matcher ' s settings .
 * < p > Performs caching based on { @ link # setCachePatterns } , delegating to
 * { @ link # tokenizePath ( String ) } for the actual tokenization algorithm .
 * @ param pattern the pattern to tokenize
 * @ return the tokenized pattern parts */
protected String [ ] tokenizePattern ( String pattern ) { } }

implementation:
String [ ] tokenized = null ; Boolean cachePatterns = this . cachePatterns ; if ( cachePatterns == null || cachePatterns . booleanValue ( ) ) { tokenized = this . tokenizedPatternCache . get ( pattern ) ; } if ( tokenized == null ) { tokenized = tokenizePath ( pattern ) ; if ( cachePatterns == null && this . tokenizedPatternCache . size ( ) >= CACHE_TURNOFF_THRESHOLD ) { // Try to adapt to the runtime situation that we ' re encountering :
// There are obviously too many different patterns coming in here . . .
// So let ' s turn off the cache since the patterns are unlikely to be reoccurring .
deactivatePatternCache ( ) ; return tokenized ; } if ( cachePatterns == null || cachePatterns . booleanValue ( ) ) { this . tokenizedPatternCache . put ( pattern , tokenized ) ; } } return tokenized ;


signature:
public class AntPathMatcher { /** * Copy the given { @ code Collection } into a { @ code String } array .
 * < p > The { @ code Collection } must contain { @ code String } elements only .
 * @ param collection the { @ code Collection } to copy
 * @ return the { @ code String } array ( { @ code null } if the supplied
 * { @ code Collection } was { @ code null } ) */
private static String [ ] toStringArray ( Collection < String > collection ) { } }

implementation: if ( collection == null ) { return null ; } return collection . toArray ( new String [ collection . size ( ) ] ) ;


signature:
public class AntPathMatcher { /** * Test whether or not a string matches against a pattern .
 * @ param pattern the pattern to match against ( never { @ code null } )
 * @ param str the String which must be matched against the pattern ( never { @ code null } )
 * @ return { @ code true } if the string matches against the pattern , or { @ code false } otherwise */
private boolean matchStrings ( String pattern , String str , Map < String , String > uriTemplateVariables ) { } }

implementation: return getStringMatcher ( pattern ) . matchStrings ( str , uriTemplateVariables ) ;


signature:
public class MultiFileReader { /** * Sets the file offsets to use for the next read . To work correctly , the last return offsets should be used or
 * an empty < code > Map < / code > if there is none .
 * If a reader is already live , the corresponding set offset is ignored as we cache all the contextual information
 * of live readers .
 * @ param offsets directory offsets .
 * @ throws IOException thrown if there was an IO error while preparing file offsets . */
public void setOffsets ( Map < String , String > offsets ) throws IOException { } }

implementation:
Utils . checkState ( open , "Not open" ) ; fileContextProvider . setOffsets ( offsets ) ; // we reset the events on every setOffsets ( ) .
events . clear ( ) ;


signature:
public class MultiFileReader { /** * Returns the current file offsets . The returned offsets should be set before the next read .
 * @ return the current file offsets .
 * @ throws IOException thrown if there was an IO error while preparing file offsets . */
public Map < String , String > getOffsets ( ) throws IOException { } }

implementation: Utils . checkState ( open , "Not open" ) ; return fileContextProvider . getOffsets ( ) ;


signature:
public class MultiFileReader { /** * remaining time till timeout , return zero if already in timeout */
private long getRemainingWaitTime ( long startTime , long maxWaitTimeMillis ) { } }

implementation: long remaining = maxWaitTimeMillis - ( System . currentTimeMillis ( ) - startTime ) ; return ( remaining > 0 ) ? remaining : 0 ;


signature:
public class MultiFileReader { /** * Reads the next { @ link LiveFileChunk } from the directories waiting the specified time for one .
 * @ param waitMillis number of milliseconds to block waiting for a chunk .
 * @ return the next chunk , or < code > null < / code > if there is no next chunk and the waiting time passed . */
public LiveFileChunk next ( long waitMillis ) { } }

implementation:
Utils . checkState ( open , "Not open" ) ; waitMillis = ( waitMillis > 0 ) ? waitMillis : 0 ; long startTime = System . currentTimeMillis ( ) ; LiveFileChunk chunk = null ; boolean exit = false ; fileContextProvider . startNewLoop ( ) ; while ( ! exit ) { if ( ! fileContextProvider . didFullLoop ( ) ) { FileContext fileContext = fileContextProvider . next ( ) ; try { LiveFileReader reader = fileContext . getReader ( ) ; if ( reader != null ) { if ( reader . hasNext ( ) ) { chunk = reader . next ( 0 ) ; if ( LOG . isTraceEnabled ( ) ) { LOG . trace ( "next(): directory '{}', file '{}', offset '{}' got data '{}'" , fileContext . getMultiFileInfo ( ) . getFileFullPath ( ) , reader . getLiveFile ( ) , reader . getOffset ( ) , chunk != null ) ; } } else { if ( LOG . isTraceEnabled ( ) ) { LOG . trace ( "next(): directory '{}', file '{}', offset '{}' EOF reached" , fileContext . getMultiFileInfo ( ) . getFileFullPath ( ) , reader . getLiveFile ( ) , reader . getOffset ( ) ) ; } } fileContext . releaseReader ( false ) ; } else { if ( LOG . isTraceEnabled ( ) ) { LOG . trace ( "next(): directory '{}', no reader available" , fileContext . getMultiFileInfo ( ) . getFileFullPath ( ) ) ; } } } catch ( IOException ex ) { LOG . error ( "Error while reading file: {}" , ex . toString ( ) , ex ) ; try { fileContext . releaseReader ( true ) ; } catch ( IOException ex1 ) { LOG . warn ( "Error while releasing reader in error: {}" , ex1 . toString ( ) , ex1 ) ; } } } // check exit conditions ( we have a chunk , or we timed - out waitMillis )
exit = chunk != null ; if ( ! exit ) { // if we looped thru all dir contexts in this call we yield CPU
if ( fileContextProvider . didFullLoop ( ) ) { exit = isTimeout ( startTime , waitMillis ) ; if ( ! exit && LOG . isTraceEnabled ( ) ) { LOG . trace ( "next(): looped through all directories, yielding CPU" ) ; } exit = exit || ! ThreadUtil . sleep ( Math . min ( getRemainingWaitTime ( startTime , waitMillis ) , MAX_YIELD_TIME ) ) ; fileContextProvider . startNewLoop ( ) ; } } } return chunk ;


signature:
public class MultiFileReader { /** * Determines the offset lag for each active file being read .
 * @ param offsetMap the current Offset for file keys .
 * @ return map of fileKey to offset . */
public Map < String , Long > getOffsetsLag ( Map < String , String > offsetMap ) throws IOException { } }

implementation: return fileContextProvider . getOffsetsLag ( offsetMap ) ;


signature:
public class StageLibraryDelegateCreator { /** * Create and initialize new delegate .
 * We usually have this functionality in a builder somewhere else ( like a Pipeline . Builder ) , but
 * we put it here for now for simplicity . Since this is container - only class , we will be able to
 * move this logic somewhere else at a later time without difficulties .
 * @ param stageLib Stage library task instance .
 * @ param configuration SDC Configuration
 * @ param stageLibraryName Name of the stage library from which the delegator should be loaded .
 * @ param exportedInterface Interface describing which delegator is needed
 * @ param < R > Interface describing which delegator is needed
 * @ return Wrapped new instance of the delegator or null on any issue */
public < R > R createAndInitialize ( StageLibraryTask stageLib , Configuration configuration , String stageLibraryName , Class < R > exportedInterface ) { } }

implementation:
StageLibraryDelegate instance = create ( stageLib , stageLibraryName , exportedInterface ) ; if ( instance == null ) { return null ; } // Create & set context
StageLibraryDelegateContext context = new StageLibraryDelegateContext ( configuration ) ; instance . setContext ( context ) ; return ( R ) new StageLibraryDelegateRuntime ( instance . getClass ( ) . getClassLoader ( ) , instance ) ;