signature:
public class KineticaTableUtils { /** * Get the table description from the ShowTableResponse */
private List < String > getTableDescription ( ) throws GPUdbException { } }

implementation: List < List < String > > descriptions = showTableResponse . getTableDescriptions ( ) ; if ( descriptions == null || descriptions . size ( ) != 1 ) { throw new GPUdbException ( "Error getting description for table " + tableName ) ; } return descriptions . get ( 0 ) ;


signature:
public class KineticaTableUtils { /** * if it is not */
private void validateTableAcceptsInserts ( ) throws GPUdbException { } }

implementation: for ( String s : tableDescription ) { if ( s . equalsIgnoreCase ( "COLLECTION" ) ) { throw new GPUdbException ( "Error: table " + tableName + " is a Collection" ) ; } else if ( s . equalsIgnoreCase ( "VIEW" ) ) { throw new GPUdbException ( "Error: table " + tableName + " is a View" ) ; } else if ( s . equalsIgnoreCase ( "JOIN" ) ) { throw new GPUdbException ( "Error: table " + tableName + " is a Join Table" ) ; } else if ( s . equalsIgnoreCase ( "RESULT_TABLE" ) ) { throw new GPUdbException ( "Error: table " + tableName + " is a Result Table" ) ; } }


signature:
public class KineticaTableUtils { /** * Get the Class for the column type */
private Class < ? > getColumnType ( JSONObject field ) throws GPUdbException { } }

implementation:
Class < ? > columnType = null ; // The Avro " type " element might be an array if the type is nullable
if ( field . get ( "type" ) instanceof JSONArray ) { JSONArray columnTypes = field . getJSONArray ( "type" ) ; for ( int j = 0 ; j < columnTypes . length ( ) ; j ++ ) { String ct = ( String ) columnTypes . get ( j ) ; if ( ! ct . equals ( "null" ) ) { columnType = getClassForType ( ct ) ; break ; } } } else { columnType = getClassForType ( field . getString ( "type" ) ) ; } if ( columnType == null ) { throw new GPUdbException ( "Error getting column type for field: " + field . toString ( ) ) ; } return columnType ;


signature:
public class KineticaTableUtils { /** * the Avro " type " field ) */
private boolean typeIsNullable ( JSONObject field ) throws GPUdbException { } }

implementation: if ( field . get ( "type" ) instanceof JSONArray ) { JSONArray columnTypes = field . getJSONArray ( "type" ) ; for ( int j = 0 ; j < columnTypes . length ( ) ; j ++ ) { String ct = ( String ) columnTypes . get ( j ) ; if ( ct . equals ( "null" ) ) { return true ; } } } return false ;


signature:
public class KineticaTableUtils { /** * Get the table ' s schema as a JSON Object */
private JSONObject getTableSchema ( String tableName , ShowTableResponse showTableResponse ) throws GPUdbException { } }

implementation: List < String > schemas = showTableResponse . getTypeSchemas ( ) ; if ( schemas == null || schemas . size ( ) != 1 ) { throw new GPUdbException ( "Error getting schema for table " + tableName ) ; } return new JSONObject ( schemas . get ( 0 ) ) ;


signature:
public class KineticaTableUtils { /** * Get the table ' s extended column properties */
private Map < String , List < String > > getColumnProperties ( String tableName , ShowTableResponse showTableResponse ) throws GPUdbException { } }

implementation: List < Map < String , List < String > > > columnPropertiesList = showTableResponse . getProperties ( ) ; if ( columnPropertiesList == null || columnPropertiesList . size ( ) != 1 ) { throw new GPUdbException ( "Error getting properties for table " + tableName ) ; } return columnPropertiesList . get ( 0 ) ;


signature:
public class KineticaTableUtils { /** * Get the Java type for a type name */
private Class < ? > getClassForType ( String typeName ) throws GPUdbException { } }

implementation: typeName = typeName . replace ( " " , "" ) ; if ( typeName . equalsIgnoreCase ( STRING_TYPE_NAME ) ) { return String . class ; } else if ( typeName . equalsIgnoreCase ( LONG_TYPE_NAME ) ) { return Long . class ; } else if ( typeName . equalsIgnoreCase ( INTEGER_TYPE_NAME ) ) { return Integer . class ; } else if ( typeName . equalsIgnoreCase ( FLOAT_TYPE_NAME ) ) { return Float . class ; } else if ( typeName . equalsIgnoreCase ( DOUBLE_TYPE_NAME ) ) { return Double . class ; } else if ( typeName . equalsIgnoreCase ( BYTES_TYPE_NAME ) ) { return ByteBuffer . class ; } else { throw new GPUdbException ( "Error: unknown type '" + typeName + "' in table schema" ) ; }


signature:
public class AerospikeBeanConfig { /** * initialize and validate configuration options
 * @ param context
 * @ param issues */
public void init ( Target . Context context , List < Target . ConfigIssue > issues ) { } }

implementation: List < Host > hosts = getAerospikeHosts ( issues , connectionString , Groups . AEROSPIKE . getLabel ( ) , "aerospikeBeanConfig.connectionString" , context ) ; ClientPolicy cp = new ClientPolicy ( ) ; try { client = new AerospikeClient ( cp , hosts . toArray ( new Host [ hosts . size ( ) ] ) ) ; int retries = 0 ; while ( ! client . isConnected ( ) && retries <= maxRetries ) { if ( retries > maxRetries ) { issues . add ( context . createConfigIssue ( Groups . AEROSPIKE . getLabel ( ) , "aerospikeBeanConfig.connectionString" , AerospikeErrors . AEROSPIKE_03 , connectionString ) ) ; return ; } retries ++ ; try { Thread . sleep ( 100 ) ; } catch ( InterruptedException ignored ) { } } } catch ( AerospikeException ex ) { issues . add ( context . createConfigIssue ( Groups . AEROSPIKE . getLabel ( ) , "aerospikeBeanConfig.connectionString" , AerospikeErrors . AEROSPIKE_03 , connectionString ) ) ; }


signature:
public class MqttClientTarget { /** * Returns the topic given the record .
 * Returns the configured topic or statically evaluated topic in case runtime resolution is not required .
 * If runtime resolution is required then the following is done :
 * 1 . Resolve the topic name by evaluating the topic expression
 * 2 . If the white list does not contain topic name and white list is not configured with " * " throw StageException
 * and the record will be handled based on the OnError configuration for the stage */
String getTopic ( Record record ) throws StageException { } }

implementation: String result = publisherConf . topic ; if ( publisherConf . runtimeTopicResolution ) { RecordEL . setRecordInContext ( topicVars , record ) ; try { result = topicEval . eval ( topicVars , publisherConf . topicExpression , String . class ) ; if ( isEmpty ( result ) ) { throw new StageException ( Errors . MQTT_08 , publisherConf . topicExpression , record . getHeader ( ) . getSourceId ( ) ) ; } if ( ! allowedTopics . contains ( result ) && ! allowAllTopics ) { throw new StageException ( Errors . MQTT_09 , result , record . getHeader ( ) . getSourceId ( ) ) ; } } catch ( ELEvalException e ) { throw new StageException ( Errors . MQTT_10 , publisherConf . topicExpression , record . getHeader ( ) . getSourceId ( ) , e . toString ( ) ) ; } } return result ;


signature:
public class Strftime { /** * Search the provided pattern and get the C standard
 * Date / Time formatting rules and convert them to the
 * Java equivalent .
 * @ param pattern The pattern to search
 * @ return The modified pattern */
public String convertDateFormat ( String pattern ) { } }

implementation:
boolean inside = false ; boolean mark = false ; boolean modifiedCommand = false ; StringBuilder buf = new StringBuilder ( ) ; for ( int i = 0 ; i < pattern . length ( ) ; i ++ ) { char c = pattern . charAt ( i ) ; if ( c == '%' && ! mark ) { mark = true ; } else { if ( mark ) { if ( modifiedCommand ) { // don ' t do anything - - we just wanted to skip a char
modifiedCommand = false ; mark = false ; } else { inside = translateCommand ( buf , pattern , i , inside ) ; // It ' s a modifier code
if ( c == 'O' || c == 'E' ) { modifiedCommand = true ; } else { mark = false ; } } } else { if ( ! inside && c != ' ' ) { // We start a literal , which we need to quote
buf . append ( "'" ) ; inside = true ; } buf . append ( c ) ; } } } if ( buf . length ( ) > 0 ) { char lastChar = buf . charAt ( buf . length ( ) - 1 ) ; if ( lastChar != '\'' && inside ) { buf . append ( '\'' ) ; } } return buf . toString ( ) ;


signature:
public class Strftime { /** * Try to get the Java Date / Time formatting associated with
 * the C standard provided .
 * @ param buf The buffer
 * @ param pattern The date / time pattern
 * @ param index The char index
 * @ param oldInside Flag value
 * @ return True if new is inside buffer */
protected boolean translateCommand ( StringBuilder buf , String pattern , int index , boolean oldInside ) { } }

implementation:
char firstChar = pattern . charAt ( index ) ; boolean newInside = oldInside ; // O and E are modifiers , they mean to present an alternative representation of the next char
// we just handle the next char as if the O or E wasn ' t there
if ( firstChar == 'O' || firstChar == 'E' ) { if ( index + 1 < pattern . length ( ) ) { newInside = translateCommand ( buf , pattern , index + 1 , oldInside ) ; } else { buf . append ( quote ( "%" + firstChar , oldInside ) ) ; } } else { String command = translate . getProperty ( String . valueOf ( firstChar ) ) ; // If we don ' t find a format , treat it as a literal - - That ' s what apache does
if ( command == null ) { buf . append ( quote ( "%" + firstChar , oldInside ) ) ; } else { // If we were inside quotes , close the quotes
if ( oldInside ) { buf . append ( '\'' ) ; } buf . append ( command ) ; newInside = false ; } } return newInside ;


signature:
public class S3ConnectionBaseConfig { /** * / * Max Error retries > = 0 are set in ClientConfig for S3Client , < 0 will use default ( 3) */
public void init ( Stage . Context context , String configPrefix , ProxyConfig proxyConfig , List < Stage . ConfigIssue > issues , int maxErrorRetries ) { } }

implementation: this . maxErrorRetries = maxErrorRetries ; commonPrefix = AWSUtil . normalizePrefix ( commonPrefix , delimiter ) ; try { createConnection ( context , configPrefix , proxyConfig , issues , maxErrorRetries ) ; } catch ( StageException ex ) { LOG . debug ( Errors . S3_SPOOLDIR_20 . getMessage ( ) , ex . toString ( ) , ex ) ; issues . add ( context . createConfigIssue ( Groups . S3 . name ( ) , configPrefix + S3ConnectionBaseConfig . AWS_CONFIG_PREFIX + "awsAccessKeyId" , Errors . S3_SPOOLDIR_20 , ex . toString ( ) ) ) ; }


signature:
public class JmsSourceUpgrader { /** * Transition to services */
private static void upgradeV5ToV6 ( List < Config > configs , Context context ) { } }

implementation:
List < Config > dataFormatConfigs = configs . stream ( ) . filter ( c -> c . getName ( ) . startsWith ( "dataFormat" ) ) . collect ( Collectors . toList ( ) ) ; // Remove those configs
configs . removeAll ( dataFormatConfigs ) ; // There is an interesting history with compression - at some point ( version 2 ) , we explicitly added it , then
// we have hidden it . So this config might or might not exists , depending on the version in which the pipeline
// was created . However the service is expecting it and thus we need to ensure that it ' s there .
if ( dataFormatConfigs . stream ( ) . noneMatch ( c -> "dataFormatConfig.compression" . equals ( c . getName ( ) ) ) ) { dataFormatConfigs . add ( new Config ( "dataFormatConfig.compression" , "NONE" ) ) ; } // And finally register new service
context . registerService ( DataFormatParserService . class , dataFormatConfigs ) ;


signature:
public class FieldPathExpressionUtil { /** * Evaluates a given field path expression against a given record and returns all field paths that satisfy the
 * expression upon evaluation . Analagous in function to { @ link FieldRegexUtil # getMatchingFieldPaths ( String , Set ) } ,
 * and even identical in behavior to it when the given { @ code fieldExpression } does not contain any EL expressions .
 * @ param fieldExpression the field path expression to evaluate
 * @ param elEval the { @ link ELEval } instance in which to evaluate the expression
 * @ param elVars the { @ link ELVars } instance to use when evaluating the expression
 * @ param record the record against which to evaluate the expression ; all returned values will be valid field paths
 * within it
 * @ param recordEscapedFieldPaths the record ' s complete set of escaped field paths ; should be identical to calling
 * record . getEscapedFieldPaths ( ) on the record param
 * @ return a { @ link List } of field paths satisfying the given expression within the given { @ code record }
 * @ throws ELEvalException */
public static List < String > evaluateMatchingFieldPaths ( String fieldExpression , ELEval elEval , ELVars elVars , Record record , Iterable < String > recordEscapedFieldPaths ) throws ELEvalException { } }

implementation:
if ( isFieldPathExpressionFast ( fieldExpression ) ) { // this field path expression actually does contain an EL expression , so need to evaluate against all fields
return evaluateMatchingFieldPathsImpl ( fieldExpression , elEval , elVars , record ) ; } else { // else it does NOT contain one , so the field regex util ( which is faster ) can be used
return FieldRegexUtil . getMatchingFieldPaths ( fieldExpression , recordEscapedFieldPaths ) ; }


signature:
public class FieldPathExpressionUtil { /** * Evaluates a field expression against a specific field in a record to see if it matches
 * @ param record the record which contains the specified field
 * @ param fieldPath the path to the specified field
 * @ param fieldExpression the field expression to evaluate against the specified record / field
 * @ param elEval the { @ link ELEval } instance in which to evaluate the expression
 * @ param elVars the { @ link ELVars } instance to use when evaluating the expression
 * @ return true if the expression matches the field , false otherwise
 * @ throws ELEvalException if an error occurs during EL evaluation */
private static boolean pathMatches ( Record record , String fieldPath , String fieldExpression , ELEval elEval , ELVars elVars ) throws ELEvalException { } }

implementation:
List < PathElement > actualPathElements = PathElement . parse ( fieldPath , true ) ; List < PathElement > matcherPathElements = PathElement . parse ( fieldExpression , false , true ) ; Iterator < PathElement > currentPathIter = actualPathElements . iterator ( ) ; Iterator < PathElement > matcherPathIter = matcherPathElements . iterator ( ) ; PathElement currentMatcher = null ; PathElement currentPath = null ; Field currentField = null ; StringBuilder currentFieldPath = new StringBuilder ( ) ; while ( matcherPathIter . hasNext ( ) ) { currentMatcher = matcherPathIter . next ( ) ; switch ( currentMatcher . getType ( ) ) { case MAP : if ( ! currentPathIter . hasNext ( ) ) { // we are expecting to match a MAP , but there are no more elements in the path
return false ; } currentPath = currentPathIter . next ( ) ; // see if the name matches the pattern
String childName = currentPath . getName ( ) ; String patternName = currentMatcher . getName ( ) ; if ( FieldRegexUtil . hasWildCards ( patternName ) ) { patternName = FieldRegexUtil . transformFieldPathRegex ( patternName ) ; } if ( childName . matches ( patternName ) ) { currentField = currentField . getValueAsMap ( ) . get ( childName ) ; currentFieldPath . append ( "/" ) ; currentFieldPath . append ( childName ) ; } else { return false ; } break ; case LIST : // see if the index matches the pattern
if ( ! currentPathIter . hasNext ( ) ) { // we are expecting to match a LIST , but there are no more elements in the path
return false ; } currentPath = currentPathIter . next ( ) ; int childIndex = currentPath . getIndex ( ) ; final int matchInd = currentMatcher . getIndex ( ) ; if ( matchInd == PathElement . WILDCARD_INDEX_ANY_LENGTH || ( matchInd == PathElement . WILDCARD_INDEX_SINGLE_CHAR && childIndex < 10 ) || matchInd == childIndex ) { currentField = currentField . getValueAsList ( ) . get ( childIndex ) ; currentFieldPath . append ( "[" ) ; currentFieldPath . append ( childIndex ) ; currentFieldPath . append ( "]" ) ; } else { return false ; } break ; case ROOT : if ( ! currentPathIter . hasNext ( ) ) { // we are expecting to match the ROOT , but there are no more elements in the path
return false ; } currentPath = currentPathIter . next ( ) ; if ( currentPath . getType ( ) != PathElement . Type . ROOT ) { // we are expecting to match the ROOT , the root element wasn ' t for some reason
return false ; } currentField = record . get ( ) ; break ; case FIELD_EXPRESSION : // see if the current field matches the given expression
FieldEL . setFieldInContext ( elVars , currentFieldPath . toString ( ) , currentPath . getName ( ) , currentField ) ; String expression = currentMatcher . getName ( ) ; final boolean result = elEval . eval ( elVars , expression , Boolean . class ) ; if ( LOG . isDebugEnabled ( ) ) { LOG . debug ( "Result of evaluating expression {} on field {} with path {} was {}" , expression , currentField , currentFieldPath , result ) ; } if ( ! result ) { return false ; } break ; } } return ! currentPathIter . hasNext ( ) ;


signature:
public class DataStore { /** * This method must be used after completing the write to output stream which was obtained by calling the
 * { @ link # getOutputStream ( ) } method .
 * If the write operation was successful , then this method must be called after calling the
 * { @ link # commit ( java . io . OutputStream ) } method . Otherwise it must be called after calling the { @ link # close ( ) } on the
 * output stream .
 * Example usage :
 * DataStore dataStore = new DataStore ( . . . ) ;
 * try ( OutputStream os = dataStore . getOutputStream ( ) ) {
 * os . write ( . . ) ;
 * dataStore . commit ( os ) ;
 * } catch ( IOException e ) {
 * } finally {
 * dataStore . release ( ) ; */
public void release ( ) { } }

implementation: CounterLock lock ; synchronized ( DataStore . class ) { lock = FILE_LOCKS . get ( file ) ; if ( lock == null ) { LOG . error ( "Trying to release unlocked file {}" , file ) ; return ; } lock . dec ( ) ; if ( lock . counter == 0 ) { FILE_LOCKS . remove ( file ) ; } } LOG . trace ( "Releasing the lock {} for '{}'" , lock , file ) ; lock . unlock ( ) ; LOG . trace ( "Released the lock {} for '{}'" , lock , file ) ;


signature:
public class DataStore { /** * Returns an input stream for the requested file .
 * After completing the read the stream must be closed . It is not necessary to call the { @ link # release ( ) }
 * method after reading from the input stream .
 * Example usage :
 * DataStore dataStore = new DataStore ( . . . ) ;
 * try ( InputStream is = dataStore . getInputStream ( ) ) {
 * / / read from is
 * } catch ( IOException e ) {
 * @ return
 * @ throws IOException */
public InputStream getInputStream ( ) throws IOException { } }

implementation: acquireLock ( ) ; try { isClosed = false ; forWrite = false ; LOG . trace ( "Starts read '{}'" , file ) ; verifyAndRecover ( ) ; InputStream is = new ProxyInputStream ( new FileInputStream ( file . toFile ( ) ) ) { @ Override public void close ( ) throws IOException { if ( isClosed ) { return ; } try { super . close ( ) ; } finally { release ( ) ; isClosed = true ; stream = null ; } LOG . trace ( "Finishes read '{}'" , file ) ; } } ; stream = is ; return is ; } catch ( Exception ex ) { release ( ) ; throw ex ; }


signature:
public class DataStore { /** * Returns an output stream for the requested file .
 * After completing the write the contents must be committed using the { @ link # commit ( java . io . OutputStream ) }
 * method and the stream must be released using the { @ link # release ( ) } method .
 * Example usage :
 * DataStore dataStore = new DataStore ( . . . ) ;
 * try ( OutputStream os = dataStore . getOutputStream ( ) ) {
 * os . write ( . . ) ;
 * dataStore . commit ( os ) ;
 * } catch ( IOException e ) {
 * } finally {
 * dataStore . release ( ) ;
 * @ return
 * @ throws IOException */
public OutputStream getOutputStream ( ) throws IOException { } }

implementation: acquireLock ( ) ; try { isClosed = false ; forWrite = true ; LOG . trace ( "Starts write '{}'" , file ) ; verifyAndRecover ( ) ; if ( Files . exists ( file ) ) { Files . move ( file , fileOld ) ; LOG . trace ( "Starting write, move '{}' to '{}'" , file , fileOld ) ; } OutputStream os = new ProxyOutputStream ( new FileOutputStream ( fileTmp . toFile ( ) ) ) { @ Override public void close ( ) throws IOException { if ( isClosed ) { return ; } try { super . close ( ) ; } finally { isClosed = true ; stream = null ; } LOG . trace ( "Finishes write '{}'" , file ) ; } } ; stream = os ; return os ; } catch ( Exception ex ) { release ( ) ; throw ex ; }


signature:
public class DataStore { /** * This method must be used to commit contents written to the output stream which is obtained by calling
 * the { @ link # getOutputStream ( ) } method . This method closes the argument output stream .
 * Example usage :
 * DataStore dataStore = new DataStore ( . . . ) ;
 * try ( OutputStream os = dataStore . getOutputStream ( ) ) {
 * os . write ( . . ) ;
 * dataStore . commit ( os ) ;
 * } catch ( IOException e ) {
 * } finally {
 * dataStore . release ( ) ;
 * @ throws IOException */
public void commit ( OutputStream out ) throws IOException { } }

implementation:
// close the stream in order to flush the contents into the disk
Utils . checkNotNull ( out , "Argument output stream cannot be null" ) ; Utils . checkState ( stream == out , "The argument output stream must be the same as the output stream obtained " + "from this data store instance" ) ; out . close ( ) ; Files . move ( fileTmp , fileNew ) ; LOG . trace ( "Committing write, move '{}' to '{}'" , fileTmp , fileNew ) ; Files . move ( fileNew , file ) ; LOG . trace ( "Committing write, move '{}' to '{}'" , fileNew , file ) ; if ( Files . exists ( fileOld ) ) { Files . delete ( fileOld ) ; LOG . trace ( "Committing write, deleting '{}'" , fileOld ) ; } LOG . trace ( "Committed" ) ;


signature:
public class DataStore { /** * Check if the DataStore exists and contains data .
 * This method will check for the presence of the set of files that can be used to read data from the store . */
public boolean exists ( ) throws IOException { } }

implementation: acquireLock ( ) ; try { verifyAndRecover ( ) ; return Files . exists ( file ) && Files . size ( file ) > 0 ; } finally { release ( ) ; }


signature:
public class JdbcMicrosoftRecordReader { /** * 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 " _ _ $ operation " in record .
if ( StringUtils . isBlank ( op ) ) { if ( record . has ( MSOperationCode . getOpField ( ) ) ) { int intOp = record . get ( MSOperationCode . getOpField ( ) ) . getValueAsInteger ( ) ; // Convert the MS specific operation code to SDC standard operation code
opCode = MSOperationCode . convertToJDBCCode ( intOp ) ; } } else { opCode = JDBCOperationType . convertToIntCode ( op ) ; } if ( opCode == - 1 ) { // Both MS code and sdc code are not set . Use default .
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 HdfsUtils { /** * Parse String representation of permissions into HDFS FsPermission class .
 * This method accepts the following formats :
 * * Octal like ' 777 ' or ' 770'
 * * HDFS style changes like ' a - rwx '
 * * Unix style write up with 9 characters like ' rwxrwx - - - '
 * @ param permissions String representing the permissions
 * @ return Parsed FsPermission object */
public static FsPermission parseFsPermission ( String permissions ) throws IllegalArgumentException { } }

implementation:
try { // Octal or symbolic representation
return new FsPermission ( permissions ) ; } catch ( IllegalArgumentException e ) { // FsPermission . valueOf will work with unix style permissions which is 10 characters
// where the first character says the type of file
if ( permissions . length ( ) == 9 ) { // This means it is a posix standard without the first character for file type
// We will simply set it to ' - ' suggesting regular file
permissions = "-" + permissions ; } // Try to parse unix style format .
return FsPermission . valueOf ( permissions ) ; }


signature:
public class ClassLoaderStageLibraryTask { /** * Validate that all required libraries are available and loaded . */
@ VisibleForTesting void validateRequiredStageLibraries ( ) { } }

implementation:
// Name of all installed libraries
Set < String > installedLibraries = stageLibraries . stream ( ) . map ( StageLibraryDefinition :: getName ) . collect ( Collectors . toSet ( ) ) ; // Required libraries
Set < String > requiredLibraries = new HashSet < > ( ) ; String config = configuration . get ( SdcConfiguration . REQUIRED_STAGELIBS , DEFAULT_REQUIRED_STAGELIBS ) ; for ( String stageLib : config . split ( "," ) ) { if ( ! stageLib . isEmpty ( ) ) { requiredLibraries . add ( stageLib ) ; } } Set < String > missingLibraries = Sets . difference ( requiredLibraries , installedLibraries ) ; if ( ! missingLibraries . isEmpty ( ) ) { throw new RuntimeException ( Utils . format ( "Some required stage libraries are missing: {}" , StringUtils . join ( missingLibraries , ", " ) ) ) ; }


signature:
public class ClassLoaderStageLibraryTask { /** * Validate service dependencies .
 * Any error is considered fatal and RuntimeException ( ) will be thrown that will terminate the SDC start up procedure . */
private void validateAllServicesAvailable ( ) { } }

implementation:
// Firstly validate that all stages have satisfied service dependencies
List < String > missingServices = new LinkedList < > ( ) ; for ( StageDefinition stage : stageList ) { for ( ServiceDependencyDefinition service : stage . getServices ( ) ) { if ( ! serviceMap . containsKey ( service . getService ( ) ) ) { missingServices . add ( Utils . format ( "Stage {} is missing service {}" , stage . getName ( ) , service . getService ( ) . getName ( ) ) ) ; } } } if ( ! missingServices . isEmpty ( ) ) { throw new RuntimeException ( "Missing services: " + StringUtils . join ( missingServices , ", " ) ) ; } // Secondly ensure that all loaded services are compatible with what is supported by our runtime engine
List < String > unsupportedServices = new LinkedList < > ( ) ; for ( ServiceDefinition serviceDefinition : serviceList ) { if ( ! ServiceRuntime . supports ( serviceDefinition . getProvides ( ) ) ) { unsupportedServices . add ( serviceDefinition . getProvides ( ) . toString ( ) ) ; } } if ( ! unsupportedServices . isEmpty ( ) ) { throw new RuntimeException ( "Unsupported services: " + StringUtils . join ( unsupportedServices , ", " ) ) ; }


signature:
public class Aggregators { /** * Creates a simple Aggregator .
 * @ param name name of the aggregator .
 * @ param klass Aggregator class
 * @ return a simple aggregator instance . */
@ SuppressWarnings ( "unchecked" ) public < A extends SimpleAggregator > A createSimple ( String name , Class < ? extends Aggregator > klass ) { } }

implementation: Utils . checkState ( ! started , "Already started" ) ; try { A aggregator = ( A ) CONSTRUCTORS . get ( klass ) . newInstance ( name ) ; dataProvider . addAggregator ( aggregator ) ; aggregator . setDataProvider ( dataProvider ) ; return aggregator ; } catch ( Exception ex ) { throw new RuntimeException ( ex ) ; }


signature:
public class Aggregators { /** * Returns the unit type of an aggregator value . Typically Long or Double .
 * @ param klass the aggregator class .
 * @ return the unit type of an aggregator value . */
< A extends SimpleAggregator , T > Class < ? extends Number > getAggregatorUnit ( Class < A > klass ) { } }

implementation: try { A aggregator = ( A ) CONSTRUCTORS . get ( klass ) . newInstance ( "forAggregatorTypeDiscoveryOnly" ) ; return aggregator . getValueType ( ) ; } catch ( Exception ex ) { throw new RuntimeException ( ex ) ; }


signature:
public class Aggregators { /** * Creates an AggregatorData .
 * @ param klass aggregator type the AggregatorData is for .
 * @ param name name of the AggregatorData instance .
 * @ param timeWindowMillis timewindow of the AggregatorData instance .
 * @ return the new AggregatorData instance . */
< A extends SimpleAggregator , T > AggregatorData < A , T > createAggregatorData ( Class < A > klass , String name , long timeWindowMillis ) { } }

implementation: try { A aggregator = ( A ) CONSTRUCTORS . get ( klass ) . newInstance ( name ) ; return aggregator . createAggregatorData ( timeWindowMillis ) ; } catch ( Exception ex ) { throw new RuntimeException ( ex ) ; }


signature:
public class Aggregators { /** * Creates a group - by Agregator .
 * @ param name name of the group - by Aggregator .
 * @ param aKlass simple aggregator class backing each group - by element .
 * @ return a group - by aggregator . */
@ SuppressWarnings ( "unchecked" ) public < A extends SimpleAggregator , N extends Number > GroupByAggregator < A , N > createGroupBy ( String name , Class < ? extends Aggregator > aKlass ) { } }

implementation: Utils . checkState ( ! started , "Already started" ) ; GroupByAggregator < A , N > aggregator = new GroupByAggregator ( name , aKlass , this ) ; dataProvider . addAggregator ( aggregator ) ; aggregator . setDataProvider ( dataProvider ) ; return aggregator ;


signature:
public class Aggregators { /** * Starts the Aggregators instance .
 * @ param newDataWindowEndTimeMillis ending time of the new DataWindow . */
public void start ( long newDataWindowEndTimeMillis ) { } }

implementation: Utils . checkState ( ! started , "Already started" ) ; Utils . checkState ( ! stopped , "Already stopped" ) ; dataProvider . start ( newDataWindowEndTimeMillis ) ; started = true ;


signature:
public class Aggregators { /** * Stops the Aggregators instance . */
public Map < Aggregator , AggregatorData > stop ( ) { } }

implementation: Utils . checkState ( started , "Already started" ) ; Utils . checkState ( ! stopped , "Already stopped" ) ; Map < Aggregator , AggregatorData > aggregatorDataMap = dataProvider . stop ( ) ; stopped = true ; return aggregatorDataMap ;


signature:
public class Aggregators { /** * Atomically rolls the DataWindow of all aggregators associated with the Aggregators instance .
 * @ param newDataWindowEndTimeMillis ending time of the new DataWindow .
 * @ return a Map with all the aggregators data of the DataWindow that closed . */
public Map < Aggregator , AggregatorData > roll ( long newDataWindowEndTimeMillis ) { } }

implementation: Utils . checkState ( started , "Not started" ) ; Utils . checkState ( ! stopped , "Already stopped" ) ; return dataProvider . roll ( newDataWindowEndTimeMillis ) ;


signature:
public class JdbcMetastoreUtil { /** * Evaluate precision or scale in context of record and given field path . */
private static int resolveScaleOrPrecisionExpression ( String type , Field field , String attributeName , String fieldPath ) throws JdbcStageCheckedException { } }

implementation: String stringValue = field . getAttribute ( attributeName ) ; try { return Integer . parseInt ( stringValue ) ; } catch ( NumberFormatException e ) { throw new JdbcStageCheckedException ( JdbcErrors . JDBC_304 , type , fieldPath , attributeName , stringValue , e ) ; }


signature:
public class LoginManager { /** * Returns an instance of ` LoginManager ` and increases its reference count .
 * ` release ( ) ` should be invoked when the ` LoginManager ` is no longer needed . This method will try to reuse an
 * existing ` LoginManager ` for the provided ` mode ` if available . However , it expects ` configs ` to be the same for
 * every invocation and it will ignore them in the case where it ' s returning a cached instance of ` LoginManager ` .
 * This is a bit ugly and it would be nicer if we could pass the ` LoginManager ` to ` ChannelBuilders . create ` and
 * shut it down when the broker or clients are closed . It ' s straightforward to do the former , but it ' s more
 * complicated to do the latter without making the consumer API more complex .
 * @ param loginType the type of the login context , it should be SERVER for the broker and CLIENT for the clients
 * ( i . e . consumer and producer )
 * @ param configs configuration as key / value pairs */
public static final LoginManager acquireLoginManager ( LoginType loginType , Map < String , ? > configs ) throws IOException , LoginException { } }

implementation: synchronized ( LoginManager . class ) { LoginManager loginManager = CACHED_INSTANCES . get ( loginType ) ; if ( loginManager == null ) { loginManager = new LoginManager ( loginType , configs ) ; CACHED_INSTANCES . put ( loginType , loginManager ) ; } return loginManager . acquire ( ) ; }


signature:
public class LoginManager { /** * Decrease the reference count for this instance and release resources if it reaches 0. */
public void release ( ) { } }

implementation: synchronized ( LoginManager . class ) { if ( refCount == 0 ) throw new IllegalStateException ( "release called on LoginManager with refCount == 0" ) ; else if ( refCount == 1 ) { CACHED_INSTANCES . remove ( loginType ) ; login . shutdown ( ) ; } -- refCount ; }


signature:
public class LoginManager { /** * / * Should only be used in tests . */
public static void closeAll ( ) { } }

implementation: synchronized ( LoginManager . class ) { for ( LoginType loginType : new ArrayList < > ( CACHED_INSTANCES . keySet ( ) ) ) { LoginManager loginManager = CACHED_INSTANCES . remove ( loginType ) ; loginManager . login . shutdown ( ) ; } }


signature:
public class FileFilter { /** * Convert a limited file glob into a simple regex .
 * @ param glob file specification glob
 * @ return regex . */
private static String globToRegex ( String glob ) { } }

implementation:
if ( glob . charAt ( 0 ) == '.' || glob . contains ( "/" ) || glob . contains ( "~" ) ) { throw new IllegalArgumentException ( "Invalid character in file glob" ) ; } // treat dot as a literal .
glob = glob . replace ( "." , "\\." ) ; glob = glob . replace ( "*" , ".+" ) ; glob = glob . replace ( "?" , ".{1}+" ) ; return glob ;


signature:
public class NetflowCommonDecoder { /** * Decodes one or more { @ link NetflowV5Message } from a packet . For this method , the data ( ByteBuf ) is assumed
 * to be complete ( i . e . it will contain all the data ) , which is the case for UDP .
 * @ param buf the byte buffer from the packet
 * @ param resultMessages a list of messages to populate from the parsing operation
 * @ param sender the packet sender address
 * @ param recipient the packet recipient address
 * @ throws OnRecordErrorException */
public void decodeStandaloneBuffer ( ByteBuf buf , List < BaseNetflowMessage > resultMessages , InetSocketAddress sender , InetSocketAddress recipient ) throws OnRecordErrorException { } }

implementation: final List < Object > results = new LinkedList < > ( ) ; try { decode ( null , buf , results , sender , recipient , true ) ; for ( Object result : results ) { if ( result == null ) { LOG . warn ( "null result found from decoding standalone Netflow buffer; skipping" ) ; continue ; } if ( result instanceof BaseNetflowMessage ) { resultMessages . add ( ( BaseNetflowMessage ) result ) ; } else { throw new IllegalStateException ( String . format ( "Found unexpected object type in results: %s" , result . getClass ( ) . getName ( ) ) ) ; } } } finally { resetStateVariables ( ) ; }


signature:
public class AmazonS3Util { /** * Lists objects from AmazonS3 in lexicographical order
 * @ param s3Client
 * @ param s3ConfigBean
 * @ param pathMatcher glob patterns to match file name against
 * @ param s3Offset current offset which provides the key name of the previous object
 * @ param fetchSize number of objects to fetch in one go
 * @ return
 * @ throws AmazonClientException */
static List < S3ObjectSummary > listObjectsLexicographically ( AmazonS3 s3Client , S3ConfigBean s3ConfigBean , AntPathMatcher pathMatcher , S3Offset s3Offset , int fetchSize ) { } }

implementation:
// Incrementally scan objects after the marker ( s3Offset ) .
List < S3ObjectSummary > list = new ArrayList < > ( fetchSize ) ; ListObjectsRequest listObjectsRequest = new ListObjectsRequest ( ) ; listObjectsRequest . setBucketName ( s3ConfigBean . s3Config . bucket ) ; listObjectsRequest . setPrefix ( s3ConfigBean . s3Config . commonPrefix ) ; listObjectsRequest . setMaxKeys ( BATCH_SIZE ) ; if ( s3Offset . getKey ( ) != null ) { listObjectsRequest . setMarker ( s3Offset . getKey ( ) ) ; } ObjectListing objectListing = s3Client . listObjects ( listObjectsRequest ) ; while ( true ) { for ( S3ObjectSummary s : objectListing . getObjectSummaries ( ) ) { String fullPrefix = s . getKey ( ) ; String remainingPrefix = fullPrefix . substring ( s3ConfigBean . s3Config . commonPrefix . length ( ) , fullPrefix . length ( ) ) ; if ( ! remainingPrefix . isEmpty ( ) ) { if ( pathMatcher . match ( s3ConfigBean . s3FileConfig . prefixPattern , remainingPrefix ) ) { list . add ( s ) ; } // We ' ve got enough objects .
if ( list . size ( ) == fetchSize ) { return list ; } } } // Listing is complete . No more objects to be listed .
if ( ! objectListing . isTruncated ( ) ) { break ; } objectListing = s3Client . listNextBatchOfObjects ( objectListing ) ; } return list ;


signature:
public class AmazonS3Util { /** * Lists objects from AmazonS3 in chronological order [ lexicographical order if 2 files have same timestamp ] which are
 * later than or equal to the timestamp of the previous offset object
 * @ param s3Client
 * @ param s3ConfigBean
 * @ param pathMatcher glob patterns to match file name against
 * @ param s3Offset current offset which provides the timestamp of the previous object
 * @ param fetchSize number of objects to fetch in one go
 * @ return
 * @ throws AmazonClientException */
static List < S3ObjectSummary > listObjectsChronologically ( AmazonS3 s3Client , S3ConfigBean s3ConfigBean , AntPathMatcher pathMatcher , S3Offset s3Offset , int fetchSize ) { } }

implementation:
// Algorithm :
// - Full scan all objects that match the file name pattern and which are later than the file in the offset
// - Select the oldest " fetchSize " number of files and return them .
TreeSet < S3ObjectSummary > treeSet = new TreeSet < > ( ( o1 , o2 ) -> { int result = o1 . getLastModified ( ) . compareTo ( o2 . getLastModified ( ) ) ; if ( result != 0 ) { // same modified time . Use name to sort
return result ; } return o1 . getKey ( ) . compareTo ( o2 . getKey ( ) ) ; } ) ; S3Objects s3ObjectSummaries = S3Objects . withPrefix ( s3Client , s3ConfigBean . s3Config . bucket , s3ConfigBean . s3Config . commonPrefix ) ; // SDC - 9413 : since the s3ObjectSummaries is in lexical order , we should get all list of files in one api call
for ( S3ObjectSummary s : s3ObjectSummaries ) { String fullPrefix = s . getKey ( ) ; String remainingPrefix = fullPrefix . substring ( s3ConfigBean . s3Config . commonPrefix . length ( ) , fullPrefix . length ( ) ) ; if ( ! remainingPrefix . isEmpty ( ) ) { // remainingPrefix can be empty .
// If the user manually creates a prefix " myFolder / mySubFolder " in bucket " myBucket " and uploads " myObject " ,
// then the first objects returned here are :
// myFolder / mySubFolder
// myFolder / mySubFolder / myObject
// All is good when pipeline is run but preview returns with no data . So we should ignore the empty file as it
// has no data
if ( pathMatcher . match ( s3ConfigBean . s3FileConfig . prefixPattern , remainingPrefix ) && isEligible ( s , s3Offset ) ) { treeSet . add ( s ) ; } if ( treeSet . size ( ) > fetchSize ) { treeSet . pollLast ( ) ; } } } return new ArrayList < > ( treeSet ) ;


signature: public class SdcHashLoginService { @ Override protected String [ ] loadRoleInfo ( UserPrincipal user ) { } }

implementation: UserIdentity id = _userStore . getUserIdentity ( user . getName ( ) ) ; if ( id == null ) return null ; Set < RolePrincipal > roles = id . getSubject ( ) . getPrincipals ( RolePrincipal . class ) ; if ( roles == null ) return null ; List < String > list = roles . stream ( ) . map ( RolePrincipal :: getName ) . collect ( Collectors . toList ( ) ) ; return list . toArray ( new String [ roles . size ( ) ] ) ;


signature: public class SdcHashLoginService { @ Override protected UserPrincipal loadUserInfo ( String userName ) { } }

implementation: UserIdentity id = _userStore . getUserIdentity ( userName ) ; if ( id != null ) { return ( UserPrincipal ) id . getUserPrincipal ( ) ; } return null ;


signature:
public class ProductionPipelineRunner { /** * Since stages are allowed to produce events during destroy ( ) phase , we handle the destroy event as simplified
 * runBatch . We go over all the StagePipes and if it ' s on data path we destroy it immediately , if it ' s on event path , we
 * run it one more time . Since the stages are sorted we know that destroyed stage will never be needed again . Non
 * stage pipes are always processed to generate required structures in PipeBatch . */
@ Override public void destroy ( SourcePipe originPipe , List < PipeRunner > pipeRunners , BadRecordsHandler badRecordsHandler , StatsAggregationHandler statsAggregationHandler ) throws StageException , PipelineRuntimeException { } }

implementation:
// We ' re no longer running
running = false ; // There are two ways a runner can be in use - used by real runner ( e . g . when origin produced data ) or when it ' s
// processing " empty " batch when the runner was idle for too long . The first case is guarded by the framework - this
// method won ' t be called until the execution successfully finished . However the second way with idle drivers is run
// from a separate thread and hence we have to ensure here that it ' s " done " before moving on with the destroy phase .
try { destroyLock . lock ( ) ; // Firstly destroy the runner , to make sure that any potential run away thread from origin will be denied
// further processing .
if ( runnerPool != null ) { runnerPool . destroy ( ) ; } int batchSize = configuration . get ( Constants . MAX_BATCH_SIZE_KEY , Constants . MAX_BATCH_SIZE_DEFAULT ) ; long lastBatchTime = offsetTracker . getLastBatchTime ( ) ; long start = System . currentTimeMillis ( ) ; FullPipeBatch pipeBatch ; // Destroy origin pipe
pipeBatch = new FullPipeBatch ( null , null , batchSize , false ) ; try { LOG . trace ( "Destroying origin pipe" ) ; pipeBatch . skipStage ( originPipe ) ; originPipe . destroy ( pipeBatch ) ; } catch ( RuntimeException e ) { LOG . warn ( "Exception throw while destroying pipe" , e ) ; } // Now destroy the pipe runners
// We ' re destroying them in reverser order to make sure that the last runner to destroy is the one with id ' 0'
// that holds reference to all class loaders . Runners with id > 0 do not own their class loaders and hence needs to
// be destroyed before the runner with id ' 0 ' .
for ( PipeRunner pipeRunner : Lists . reverse ( pipeRunners ) ) { final FullPipeBatch finalPipeBatch = pipeBatch ; pipeRunner . executeBatch ( null , null , start , pipe -> { // Set the last batch time in the stage context of each pipe
( ( StageContext ) pipe . getStage ( ) . getContext ( ) ) . setLastBatchTime ( lastBatchTime ) ; String instanceName = pipe . getStage ( ) . getConfiguration ( ) . getInstanceName ( ) ; if ( pipe instanceof StagePipe ) { // Stage pipes are processed only if they are in event path
if ( pipe . getStage ( ) . getConfiguration ( ) . isInEventPath ( ) ) { LOG . trace ( "Stage pipe {} is in event path, running last process" , instanceName ) ; pipe . process ( finalPipeBatch ) ; } else { LOG . trace ( "Stage pipe {} is in data path, skipping it's processing." , instanceName ) ; finalPipeBatch . skipStage ( pipe ) ; } } else { // Non stage pipes are executed always
LOG . trace ( "Non stage pipe {}, running last process" , instanceName ) ; pipe . process ( finalPipeBatch ) ; } // And finally destroy the pipe
try { LOG . trace ( "Running destroy for {}" , instanceName ) ; pipe . destroy ( finalPipeBatch ) ; } catch ( RuntimeException e ) { LOG . warn ( "Exception throw while destroying pipe" , e ) ; } } ) ; badRecordsHandler . handle ( null , null , pipeBatch . getErrorSink ( ) , pipeBatch . getSourceResponseSink ( ) ) ; // Next iteration should have new and empty PipeBatch
pipeBatch = new FullPipeBatch ( null , null , batchSize , false ) ; pipeBatch . skipStage ( originPipe ) ; } if ( isStatsAggregationEnabled ( ) ) { List < Record > stats = new ArrayList < > ( ) ; statsAggregatorRequests . drainTo ( stats ) ; Object timeSeriesString = pipelineConfigBean . constants . get ( MetricsEventRunnable . TIME_SERIES_ANALYSIS ) ; boolean timeSeriesAnalysis = ( timeSeriesString != null ) ? ( Boolean ) timeSeriesString : true ; String metricRegistryStr ; try { metricRegistryStr = ObjectMapperFactory . get ( ) . writer ( ) . writeValueAsString ( metrics ) ; } catch ( Exception e ) { throw new RuntimeException ( Utils . format ( "Error converting metric json to string: {}" , e ) , e ) ; } LOG . info ( "Queueing last batch of record to be sent to stats aggregator" ) ; stats . add ( AggregatorUtil . createMetricJsonRecord ( runtimeInfo . getId ( ) , runtimeInfo . getMasterSDCId ( ) , pipelineConfiguration . getMetadata ( ) , false , timeSeriesAnalysis , true , metricRegistryStr ) ) ; statsAggregationHandler . handle ( null , null , stats ) ; } } finally { destroyLock . unlock ( ) ; }


signature:
public class ProductionPipelineRunner { /** * Stops execution of the pipeline after the current batch completes */
public void stop ( ) throws PipelineException { } }

implementation: this . stop = true ; if ( batchesToCapture > 0 ) { cancelSnapshot ( this . snapshotName ) ; snapshotStore . deleteSnapshot ( pipelineName , revision , snapshotName ) ; }


signature:
public class ProductionPipelineRunner { /** * This method should be called periodically from a scheduler if the pipeline should not allow runners to be " idle "
 * for more then idleTime .
 * @ param idleTime Number of milliseconds after which a runner is considered " idle "
 * @ return Number of idle batches generated in this iteration */
public int produceEmptyBatchesForIdleRunners ( long idleTime ) throws PipelineException , StageException { } }

implementation:
LOG . debug ( "Checking if any active runner is idle" ) ; // The empty batch is suppose to be fast - almost as a zero time . It could however happened that from some reason it
// will take a long time ( possibly more then idleTime ) . To avoid infinite loops , this method will only processes up
// to total number of runners before returning .
int counter = 0 ; try { destroyLock . lock ( ) ; while ( running && counter < pipes . size ( ) ) { counter ++ ; PipeRunner runner = null ; try { runner = runnerPool . getIdleRunner ( idleTime ) ; // No more idle runners , simply stop the idle execution now
if ( runner == null ) { return counter ; } LOG . debug ( "Generating empty batch for runner: {}" , runner . getRunnerId ( ) ) ; pipeContext . getRuntimeStats ( ) . incIdleBatchCount ( ) ; // Pipe batch to keep the batch info
FullPipeBatch pipeBatch = new FullPipeBatch ( null , null , 0 , false ) ; pipeBatch . setIdleBatch ( true ) ; // We ' re explicitly skipping origin because this is framework generated , empty batch
pipeBatch . skipStage ( originPipe ) ; executeRunner ( runner , System . currentTimeMillis ( ) , pipeBatch , null , null , new HashMap < > ( ) , new HashMap < > ( ) ) ; } finally { if ( runner != null ) { runnerPool . returnRunner ( runner ) ; } } } } finally { destroyLock . unlock ( ) ; } return counter ;


signature:
public class ProductionPipelineRunner { /** * Create special batch by salvaging memory structures when pipelines gets into un - recoverable error .
 * @ param pipeBatch Current batch */
private void createFailureBatch ( FullPipeBatch pipeBatch ) { } }

implementation:
if ( ! pipelineConfigBean . shouldCreateFailureSnapshot ) { return ; } try { for ( SnapshotInfo info : snapshotStore . getSummaryForPipeline ( pipelineName , revision ) ) { // Allow only one failure snapshot to be present on a pipeline
if ( info . isFailureSnapshot ( ) ) { LOG . trace ( "Skipping creation of failure snapshot as {} already exists." , info . getId ( ) ) ; return ; } } String snapshotName = "Failure_" + UUID . randomUUID ( ) . toString ( ) ; String snapshotLabel = "Failure at " + LocalDateTime . now ( ) . toString ( ) ; snapshotStore . create ( "" , pipelineName , revision , snapshotName , snapshotLabel , true ) ; snapshotStore . save ( pipelineName , revision , snapshotName , - 1 , ImmutableList . of ( pipeBatch . createFailureSnapshot ( ) ) ) ; } catch ( PipelineException ex ) { LOG . error ( "Can't serialize failure snapshot" , ex ) ; }


signature:
public class CassandraTarget { /** * Convert a Record into a fully - bound statement . */
@ SuppressWarnings ( "unchecked" ) private BoundStatement recordToBoundStatement ( Record record ) throws StageException { } }

implementation:
ImmutableList . Builder < Object > values = new ImmutableList . Builder < > ( ) ; SortedSet < String > columnsPresent = Sets . newTreeSet ( columnMappings . keySet ( ) ) ; for ( Map . Entry < String , String > mapping : columnMappings . entrySet ( ) ) { String columnName = mapping . getKey ( ) ; String fieldPath = mapping . getValue ( ) ; // If we ' re missing fields , skip them .
// If a field is present , but null , also remove it from columnsPresent since we can ' t write nulls .
if ( ! record . has ( fieldPath ) || record . get ( fieldPath ) . getValue ( ) == null ) { columnsPresent . remove ( columnName ) ; continue ; } final Object value = record . get ( fieldPath ) . getValue ( ) ; // Special cases for handling SDC Lists and Maps ,
// basically unpacking them into raw types .
if ( value instanceof List ) { List < Object > unpackedList = new ArrayList < > ( ) ; for ( Field item : ( List < Field > ) value ) { unpackedList . add ( item . getValue ( ) ) ; } values . add ( unpackedList ) ; } else if ( value instanceof Map ) { Map < Object , Object > unpackedMap = new HashMap < > ( ) ; for ( Map . Entry < String , Field > entry : ( ( Map < String , Field > ) value ) . entrySet ( ) ) { unpackedMap . put ( entry . getKey ( ) , entry . getValue ( ) . getValue ( ) ) ; } values . add ( unpackedMap ) ; } else { values . add ( value ) ; } } PreparedStatement stmt = statementCache . getUnchecked ( columnsPresent ) ; // . toArray required to pass in a list to a varargs method .
Object [ ] valuesArray = values . build ( ) . toArray ( ) ; BoundStatement boundStmt = null ; try { boundStmt = stmt . bind ( valuesArray ) ; } catch ( CodecNotFoundException | InvalidTypeException | NullPointerException e ) { // NPE can occur if one of the values is a collection type with a null value inside it . Thus , it ' s a record
// error . Note that this runs the risk of mistakenly treating a bug as a record error .
// CodecNotFound is caused when there is no type conversion definition available from the provided type
// to the target type .
errorRecordHandler . onError ( new OnRecordErrorException ( record , Errors . CASSANDRA_06 , record . getHeader ( ) . getSourceId ( ) , e . toString ( ) , e ) ) ; } return boundStmt ;


signature:
public class ForceTarget { /** * { @ inheritDoc } */
@ Override protected List < ConfigIssue > init ( ) { } }

implementation:
// Validate configuration values and open any required resources .
List < ConfigIssue > issues = super . init ( ) ; Target . Context context = getContext ( ) ; Optional . ofNullable ( conf . init ( context , CONF_PREFIX ) ) . ifPresent ( issues :: addAll ) ; errorRecordHandler = new DefaultErrorRecordHandler ( context ) ; sObjectNameVars = getContext ( ) . createELVars ( ) ; sObjectNameEval = context . createELEval ( SOBJECT_NAME ) ; ELUtils . validateExpression ( conf . sObjectNameTemplate , context , Groups . FORCE . getLabel ( ) , SOBJECT_NAME , Errors . FORCE_12 , issues ) ; externalIdFieldVars = getContext ( ) . createELVars ( ) ; externalIdFieldEval = context . createELEval ( EXTERNAL_ID_NAME ) ; ELUtils . validateExpression ( conf . externalIdField , context , Groups . FORCE . getLabel ( ) , EXTERNAL_ID_NAME , Errors . FORCE_24 , issues ) ; if ( issues . isEmpty ( ) ) { fieldMappings = new TreeMap < > ( ) ; for ( ForceFieldMapping mapping : conf . fieldMapping ) { // SDC - 7446 Allow colon as well as period as field separator
String salesforceField = conf . useBulkAPI ? mapping . salesforceField . replace ( ':' , '.' ) : mapping . salesforceField ; fieldMappings . put ( salesforceField , mapping . sdcField ) ; } try { ConnectorConfig partnerConfig = ForceUtils . getPartnerConfig ( conf , new ForceSessionRenewer ( ) ) ; partnerConnection = Connector . newConnection ( partnerConfig ) ; if ( conf . mutualAuth . useMutualAuth ) { ForceUtils . setupMutualAuth ( partnerConfig , conf . mutualAuth ) ; } bulkConnection = ForceUtils . getBulkConnection ( partnerConfig , conf ) ; LOG . info ( "Successfully authenticated as {}" , conf . username ) ; } catch ( ConnectionException | AsyncApiException | StageException | URISyntaxException ce ) { LOG . error ( "Can't connect to SalesForce" , ce ) ; issues . add ( getContext ( ) . createConfigIssue ( Groups . FORCE . name ( ) , "connectorConfig" , Errors . FORCE_00 , ForceUtils . getExceptionCode ( ce ) + ", " + ForceUtils . getExceptionMessage ( ce ) ) ) ; } if ( conf . useBulkAPI ) { writer = new ForceBulkWriter ( fieldMappings , bulkConnection , getContext ( ) ) ; } else { writer = new ForceSoapWriter ( fieldMappings , partnerConnection ) ; } } // If issues is not empty , the UI will inform the user of each configuration issue in the list .
return issues ;


signature:
public class ForceTarget { /** * { @ inheritDoc } */
@ Override public void write ( Batch batch ) throws StageException { } }

implementation: Multimap < String , Record > partitions = ELUtils . partitionBatchByExpression ( sObjectNameEval , sObjectNameVars , conf . sObjectNameTemplate , batch ) ; Set < String > sObjectNames = partitions . keySet ( ) ; for ( String sObjectName : sObjectNames ) { List < OnRecordErrorException > errors = writer . writeBatch ( sObjectName , partitions . get ( sObjectName ) , this ) ; for ( OnRecordErrorException error : errors ) { errorRecordHandler . onError ( error ) ; } }


signature:
public class Configuration { /** * Set multiple configs at once .
 * If a value of given config is ' null ' , then this config key will be un - set . */
public void set ( Map < String , String > newConfiguration ) { } }

implementation: for ( Map . Entry < String , String > entry : newConfiguration . entrySet ( ) ) { if ( entry . getValue ( ) == null ) { this . unset ( entry . getKey ( ) ) ; } else { this . set ( entry . getKey ( ) , entry . getValue ( ) ) ; } }


signature:
public class KuduLookupProcessor { /** * { @ inheritDoc } */
@ Override protected void process ( Record record , SingleLaneProcessor . SingleLaneBatchMaker batchMaker ) throws StageException { } }

implementation:
RecordEL . setRecordInContext ( tableNameVars , record ) ; String tableName = tableNameEval . eval ( tableNameVars , conf . kuduTableTemplate , String . class ) ; if ( ! conf . caseSensitive ) { tableName = tableName . toLowerCase ( ) ; } LOG . trace ( "Processing record:{}  TableName={}" , record . toString ( ) , tableName ) ; try { try { KuduLookupKey key = generateLookupKey ( record , tableName ) ; List < Map < String , Field > > values = cache . get ( key ) ; if ( values . isEmpty ( ) ) { // No record found
if ( conf . missingLookupBehavior == MissingValuesBehavior . SEND_TO_ERROR ) { errorRecordHandler . onError ( new OnRecordErrorException ( record , Errors . KUDU_31 ) ) ; } else { // Configured to ' Send to next stage ' and ' pass as it is '
batchMaker . addRecord ( record ) ; } } else { switch ( conf . multipleValuesBehavior ) { case FIRST_ONLY : setFieldsInRecord ( record , values . get ( 0 ) ) ; batchMaker . addRecord ( record ) ; break ; case SPLIT_INTO_MULTIPLE_RECORDS : for ( Map < String , Field > lookupItem : values ) { Record newRecord = getContext ( ) . cloneRecord ( record ) ; setFieldsInRecord ( newRecord , lookupItem ) ; batchMaker . addRecord ( newRecord ) ; } break ; default : throw new IllegalStateException ( "Unknown multiple value behavior: " + conf . multipleValuesBehavior ) ; } } } catch ( ExecutionException e ) { Throwables . propagateIfPossible ( e . getCause ( ) , StageException . class ) ; Throwables . propagateIfPossible ( e . getCause ( ) , OnRecordErrorException . class ) ; throw new IllegalStateException ( e ) ; // The cache loader shouldn ' t throw anything that isn ' t a StageException .
} } catch ( OnRecordErrorException error ) { // NOSONAR
errorRecordHandler . onError ( new OnRecordErrorException ( record , error . getErrorCode ( ) , error . getParams ( ) ) ) ; }


signature:
public class KuduLookupProcessor { /** * Create a map of keyColumn - value to lookup in cache .
 * @ param record
 * @ return Map of keyColumn - value
 * @ throws OnRecordErrorException */
private KuduLookupKey generateLookupKey ( final Record record , final String tableName ) throws OnRecordErrorException { } }

implementation: Map < String , Field > keyList = new HashMap < > ( ) ; for ( Map . Entry < String , String > key : columnToField . entrySet ( ) ) { String fieldName = key . getValue ( ) ; if ( ! record . has ( fieldName ) ) { throw new OnRecordErrorException ( record , Errors . KUDU_32 , fieldName ) ; } keyList . put ( key . getKey ( ) , record . get ( fieldName ) ) ; } return new KuduLookupKey ( tableName , keyList ) ;


signature:
public class MultiKafkaSource { /** * no trespassing . . . */
private Properties getKafkaProperties ( Stage . Context context ) { } }

implementation: Properties props = new Properties ( ) ; props . putAll ( conf . kafkaOptions ) ; props . setProperty ( "bootstrap.servers" , conf . brokerURI ) ; props . setProperty ( "group.id" , conf . consumerGroup ) ; props . setProperty ( "max.poll.records" , String . valueOf ( batchSize ) ) ; props . setProperty ( KafkaConstants . KEY_DESERIALIZER_CLASS_CONFIG , conf . keyDeserializer . getKeyClass ( ) ) ; props . setProperty ( KafkaConstants . VALUE_DESERIALIZER_CLASS_CONFIG , conf . valueDeserializer . getValueClass ( ) ) ; props . setProperty ( KafkaConstants . CONFLUENT_SCHEMA_REGISTRY_URL_CONFIG , StringUtils . join ( conf . dataFormatConfig . schemaRegistryUrls , "," ) ) ; props . setProperty ( KafkaConstants . AUTO_COMMIT_OFFEST , "false" ) ; if ( context . isPreview ( ) ) { props . setProperty ( KafkaConstants . AUTO_OFFSET_RESET_CONFIG , KafkaConstants . AUTO_OFFSET_RESET_PREVIEW_VALUE ) ; } return props ;


signature:
public class MultithreadedTableProvider { /** * Checks whether any tables have had partitioning turned off or not , and updates the partition map appropriately
 * @ param reconstructedPartitions the reconstructed partitions ( may be modified ) */
private void handlePartitioningTurnedOffOrOn ( SortedSetMultimap < TableContext , TableRuntimeContext > reconstructedPartitions ) { } }

implementation: for ( TableContext tableContext : reconstructedPartitions . keySet ( ) ) { final SortedSet < TableRuntimeContext > partitions = reconstructedPartitions . get ( tableContext ) ; final TableRuntimeContext lastPartition = partitions . last ( ) ; final TableContext sourceTableContext = lastPartition . getSourceTableContext ( ) ; Utils . checkState ( sourceTableContext . equals ( tableContext ) , String . format ( "Source table context for %s should match TableContext map key of %s" , lastPartition . getDescription ( ) , tableContext . getQualifiedName ( ) ) ) ; final boolean partitioningTurnedOff = lastPartition . isPartitioned ( ) && sourceTableContext . getPartitioningMode ( ) == PartitioningMode . DISABLED ; final boolean partitioningTurnedOn = ! lastPartition . isPartitioned ( ) && sourceTableContext . isPartitionable ( ) && sourceTableContext . getPartitioningMode ( ) != PartitioningMode . DISABLED ; if ( ! partitioningTurnedOff && ! partitioningTurnedOn ) { continue ; } final Map < String , String > nextStartingOffsets = new HashMap < > ( ) ; final Map < String , String > nextMaxOffsets = new HashMap < > ( ) ; final int newPartitionSequence = lastPartition . getPartitionSequence ( ) > 0 ? lastPartition . getPartitionSequence ( ) + 1 : 1 ; if ( partitioningTurnedOff ) { LOG . info ( "Table {} has switched from partitioned to non-partitioned; partition sequence {} will be the last (with" + " no max offsets)" , sourceTableContext . getQualifiedName ( ) , newPartitionSequence ) ; lastPartition . getStartingPartitionOffsets ( ) . forEach ( ( col , off ) -> { String basedOnStartOffset = lastPartition . generateNextPartitionOffset ( col , off ) ; nextStartingOffsets . put ( col , basedOnStartOffset ) ; } ) ; } else if ( partitioningTurnedOn ) { lastPartition . getStartingPartitionOffsets ( ) . forEach ( ( col , off ) -> { String basedOnStoredOffset = lastPartition . getInitialStoredOffsets ( ) . get ( col ) ; nextStartingOffsets . put ( col , basedOnStoredOffset ) ; } ) ; nextStartingOffsets . forEach ( ( col , off ) -> nextMaxOffsets . put ( col , lastPartition . generateNextPartitionOffset ( col , off ) ) ) ; if ( ! reconstructedPartitions . remove ( sourceTableContext , lastPartition ) ) { throw new IllegalStateException ( String . format ( "Failed to remove partition %s for table %s in switching partitioning from off to on" , lastPartition . getDescription ( ) , sourceTableContext . getQualifiedName ( ) ) ) ; } LOG . info ( "Table {} has switched from non-partitioned to partitioned; using last stored offsets as the starting" + " offsets for the new partition {}" , sourceTableContext . getQualifiedName ( ) , newPartitionSequence ) ; } final TableRuntimeContext nextPartition = new TableRuntimeContext ( sourceTableContext , lastPartition . isUsingNonIncrementalLoad ( ) , ( lastPartition . isPartitioned ( ) && ! partitioningTurnedOff ) || partitioningTurnedOn , newPartitionSequence , nextStartingOffsets , nextMaxOffsets ) ; reconstructedPartitions . put ( sourceTableContext , nextPartition ) ; }


signature:
public class MultithreadedTableProvider { /** * Basically acquires more tables for the current thread to work on .
 * The maximum a thread can hold is upper bounded to the
 * value the thread number was allocated in { @ link # threadNumToMaxTableSlots }
 * If there are no tables currently owned make a blocking call to { @ link # sharedAvailableTablesQueue }
 * else simply poll { @ link # sharedAvailableTablesQueue } and it to the { @ link # ownedTablesQueue } */
@ VisibleForTesting void acquireTableAsNeeded ( int threadNumber ) throws InterruptedException { } }

implementation:
if ( ! getOwnedTablesQueue ( ) . isEmpty ( ) && batchTableStrategy == BatchTableStrategy . SWITCH_TABLES ) { final TableRuntimeContext lastOwnedPartition = getOwnedTablesQueue ( ) . pollLast ( ) ; if ( getTableContextMap ( ) . containsValue ( lastOwnedPartition . getSourceTableContext ( ) ) ) { sharedAvailableTablesQueue . offer ( lastOwnedPartition ) ; } TableContext lastOwnedTable = lastOwnedPartition . getSourceTableContext ( ) ; // need to cycle off all partitions from the same table to the end of the queue
TableRuntimeContext first = sharedAvailableTablesQueue . peek ( ) ; while ( first != null && first . getSourceTableContext ( ) . equals ( lastOwnedTable ) && ! first . equals ( lastOwnedPartition ) ) { if ( LOG . isDebugEnabled ( ) ) { LOG . debug ( "Moving partition {} to end of shared queue to comply with BatchTableStrategy of {}" , first . getDescription ( ) , batchTableStrategy . getLabel ( ) ) ; } // poll ( ) should never return null since it is actually returning ' first ' as we want to move it from the head
// of the queue , that ' s why we are calling offer , to basically remove it from the head but keep it in the queue
TableRuntimeContext toMove = sharedAvailableTablesQueue . poll ( ) ; sharedAvailableTablesQueue . offer ( toMove ) ; // Get the new head of the queue
first = sharedAvailableTablesQueue . peek ( ) ; } } if ( getOwnedTablesQueue ( ) . isEmpty ( ) ) { TableRuntimeContext head = sharedAvailableTablesQueue . poll ( ) ; if ( head != null ) { offerToOwnedTablesQueue ( head , threadNumber ) ; } } partitionFirstSharedQueueItemIfNeeded ( ) ;


signature:
public class MultithreadedTableProvider { /** * < p > Examines the first item ( " head " ) im the shared partition queue , and adds a new partition if appropriate < / p >
 * < p > A new partition will be created if the number of partitions for the head item ' s table is still less
 * than the maximum , and that table itself is partitionable < / p > */
@ VisibleForTesting void partitionFirstSharedQueueItemIfNeeded ( ) { } }

implementation: final TableRuntimeContext headPartition = getOwnedTablesQueue ( ) . peek ( ) ; if ( headPartition != null ) { synchronized ( partitionStateLock ) { keepPartitioningIfNeeded ( headPartition ) ; } } else if ( LOG . isTraceEnabled ( ) ) { LOG . trace ( "No item at head of shared partition queue" ) ; }


signature:
public class MultithreadedTableProvider { /** * Return the next table to work on for the current thread ( Will not return null )
 * Deque the current element from head of the queue and put it back at the tail to queue . */
public TableRuntimeContext nextTable ( int threadNumber ) throws InterruptedException { } }

implementation: synchronized ( partitionStateLock ) { acquireTableAsNeeded ( threadNumber ) ; final TableRuntimeContext partition = getOwnedTablesQueue ( ) . pollFirst ( ) ; if ( partition != null ) { offerToOwnedTablesQueue ( partition , threadNumber ) ; } return partition ; }


signature:
public class MultithreadedTableProvider { /** * Each { @ link TableJdbcRunnable } worker thread can call this api to update
 * if there is data / no more data on the current table */
public void reportDataOrNoMoreData ( TableRuntimeContext tableRuntimeContext , int recordCount , int batchSize , boolean resultSetEndReached , AtomicBoolean tableFinished , AtomicBoolean schemaFinished , List < String > schemaFinishedTables ) { } }

implementation:
final TableContext sourceContext = tableRuntimeContext . getSourceTableContext ( ) ; // When we see a table with data , we mark isNoMoreDataEventGeneratedAlready to false
// so we can generate event again if we don ' t see data from all tables .
if ( recordCount > 0 ) { isNoMoreDataEventGeneratedAlready = false ; tablesWithNoMoreData . remove ( tableRuntimeContext . getSourceTableContext ( ) ) ; remainingSchemasToTableContexts . put ( sourceContext . getSchema ( ) , sourceContext ) ; completedSchemasToTableContexts . remove ( sourceContext . getSchema ( ) , sourceContext ) ; } // we need to account for the activeRuntimeContexts here
// if there are still other active contexts in process , then this should do " nothing "
// if there are not other contexts , we need to figure out what the highest offset completed by the last batch was
final boolean noMoreData = recordCount == 0 || resultSetEndReached ; if ( noMoreData ) { tableRuntimeContext . setMarkedNoMoreData ( true ) ; } if ( recordCount > 0 ) { maxPartitionWithDataPerTable . put ( sourceContext , tableRuntimeContext . getPartitionSequence ( ) ) ; } boolean tableExhausted = removePartitionIfNeeded ( tableRuntimeContext ) ; if ( noMoreData ) { if ( tableExhausted ) { synchronized ( this ) { if ( LOG . isDebugEnabled ( ) ) { LOG . debug ( "Table {} exhausted" , sourceContext . getQualifiedName ( ) ) ; } final boolean newlyFinished = tablesWithNoMoreData . add ( sourceContext ) ; if ( newlyFinished && tableFinished != null ) { tableFinished . set ( true ) ; } final boolean remainingSchemaChanged = remainingSchemasToTableContexts . remove ( sourceContext . getSchema ( ) , sourceContext ) ; completedSchemasToTableContexts . put ( sourceContext . getSchema ( ) , sourceContext ) ; if ( remainingSchemaChanged && remainingSchemasToTableContexts . get ( sourceContext . getSchema ( ) ) . isEmpty ( ) && schemaFinished != null ) { schemaFinished . set ( true ) ; if ( schemaFinishedTables != null ) { completedSchemasToTableContexts . get ( sourceContext . getSchema ( ) ) . forEach ( t -> schemaFinishedTables . add ( t . getTableName ( ) ) ) ; } } } } } if ( LOG . isTraceEnabled ( ) ) { LOG . trace ( "Just released table {}; Number of Tables With No More Data {}" , tableRuntimeContext . getDescription ( ) , tablesWithNoMoreData . size ( ) ) ; }


signature:
public class MultithreadedTableProvider { /** * Used by the main thread { @ link TableJdbcSource } to check whether all
 * tables have marked no more data
 * Generate event only if we haven ' t generate no more data event already
 * or we have seen a table with records after we generated an event before */
public synchronized boolean shouldGenerateNoMoreDataEvent ( ) { } }

implementation: boolean noMoreData = ( ! isNoMoreDataEventGeneratedAlready && tablesWithNoMoreData . size ( ) == tableContextMap . size ( ) ) ; if ( noMoreData ) { isNoMoreDataEventGeneratedAlready = true ; } return noMoreData ;


signature:
public class LogCharDataParser { /** * Captures valid lines in " currentLine " and corresponding fields in " fieldsFromLogLine " .
 * Captures stack traces if any in the argument " stackTrace " */
private int readAhead ( Map < String , Field > fieldsFromLogLine , StringBuilder stackTrace ) throws DataParserException , IOException { } }

implementation:
StringBuilder multilineLog = new StringBuilder ( ) ; int read = readLine ( multilineLog ) ; int numberOfLinesRead = 0 ; while ( read > - 1 ) { try { Map < String , Field > stringFieldMap = parseLogLine ( multilineLog ) ; fieldsFromLogLine . putAll ( stringFieldMap ) ; currentLine . append ( multilineLog ) ; // If the line can be parsed successfully , do not read further
// This line will be used in the current record if this is the first line being read
// or stored for the next round if there is a line from the previous round .
break ; } catch ( DataParserException e ) { // is this the first line being read ? Yes - > throw exception
if ( previousLine . length ( ) == 0 || maxStackTraceLines == - 1 ) { throw e ; } // otherwise read until we get a line that matches pattern
if ( numberOfLinesRead < maxStackTraceLines ) { if ( numberOfLinesRead != 0 ) { stackTrace . append ( "\n" ) ; } stackTrace . append ( multilineLog . toString ( ) ) ; } numberOfLinesRead ++ ; multilineLog . setLength ( 0 ) ; read = readLine ( multilineLog ) ; } } return read ;


signature:
public class LogCharDataParser { /** * returns the reader line length , the StringBuilder has up to maxObjectLen chars */
int readLine ( StringBuilder sb ) throws IOException { } }

implementation: int c = reader . read ( ) ; int count = ( c == - 1 ) ? - 1 : 0 ; while ( c > - 1 && ! isOverMaxObjectLen ( count ) && ! checkEolAndAdjust ( c ) ) { count ++ ; sb . append ( ( char ) c ) ; c = reader . read ( ) ; } if ( isOverMaxObjectLen ( count ) ) { sb . setLength ( sb . length ( ) - 1 ) ; while ( c > - 1 && c != '\n' && c != '\r' ) { count ++ ; c = reader . read ( ) ; } checkEolAndAdjust ( c ) ; } return count ;


signature:
public class UsageTimer { /** * returns fresh UsageTimer just reset to zero accumulated time */
public UsageTimer roll ( ) { } }

implementation:
int multiplier ; synchronized ( this ) { multiplier = getMultiplier ( ) ; changeMultiplier ( - multiplier ) ; // stopAll ;
} return new UsageTimer ( ) . setName ( getName ( ) ) . changeMultiplier ( multiplier ) ;


signature:
public class SchAdmin { /** * Enable Control Hub on this Data Collector . */
public static void enableDPM ( DPMInfoJson dpmInfo , Context context ) throws IOException { } }

implementation:
Utils . checkNotNull ( dpmInfo , "DPMInfo" ) ; String dpmBaseURL = normalizeDpmBaseURL ( dpmInfo . getBaseURL ( ) ) ; // Since we support enabling / Disabling DPM , first check if token already exists for the given DPM URL .
// If token exists skip first 3 steps
String currentDPMBaseURL = context . configuration . get ( RemoteSSOService . DPM_BASE_URL_CONFIG , "" ) ; String currentAppAuthToken = context . configuration . get ( RemoteSSOService . SECURITY_SERVICE_APP_AUTH_TOKEN_CONFIG , "" ) . trim ( ) ; if ( ! currentDPMBaseURL . equals ( dpmBaseURL ) || currentAppAuthToken . length ( ) == 0 ) { // 1 . Login to DPM to get user auth token
String userAuthToken = retrieveUserToken ( dpmBaseURL , dpmInfo . getUserID ( ) , dpmInfo . getUserPassword ( ) ) ; String appAuthToken = null ; // 2 . Create Data Collector application token
Response response = null ; try { Map < String , Object > newComponentJson = new HashMap < > ( ) ; newComponentJson . put ( "organization" , dpmInfo . getOrganization ( ) ) ; newComponentJson . put ( "componentType" , "dc" ) ; newComponentJson . put ( "numberOfComponents" , 1 ) ; newComponentJson . put ( "active" , true ) ; response = ClientBuilder . newClient ( ) . target ( dpmBaseURL + "/security/rest/v1/organization/" + dpmInfo . getOrganization ( ) + "/components" ) . register ( new CsrfProtectionFilter ( "CSRF" ) ) . request ( ) . header ( SSOConstants . X_USER_AUTH_TOKEN , userAuthToken ) . put ( Entity . json ( newComponentJson ) ) ; if ( response . getStatus ( ) != Response . Status . CREATED . getStatusCode ( ) ) { throw new RuntimeException ( Utils . format ( "DPM Create Application Token failed, status code '{}': {}" , response . getStatus ( ) , response . readEntity ( String . class ) ) ) ; } List < Map < String , Object > > newComponent = response . readEntity ( new GenericType < List < Map < String , Object > > > ( ) { } ) ; if ( newComponent . size ( ) > 0 ) { appAuthToken = ( String ) newComponent . get ( 0 ) . get ( "fullAuthToken" ) ; } else { throw new RuntimeException ( "DPM Create Application Token failed: No token data from DPM Server." ) ; } } finally { if ( response != null ) { response . close ( ) ; } // Logout from DPM
logout ( dpmBaseURL , userAuthToken ) ; } // 3 . Update App Token file
updateTokenFile ( context , appAuthToken ) ; } // 4 . Update dpm . properties file
updateDpmProperties ( context , dpmBaseURL , dpmInfo . getLabels ( ) , true ) ;


signature:
public class SchAdmin { /** * Disable Control Hub on this Data Collector - with explicit login . */
public static void disableDPM ( String username , String password , String organizationId , Context context ) throws IOException { } }

implementation: String dpmBaseURL = normalizeDpmBaseURL ( context . configuration . get ( RemoteSSOService . DPM_BASE_URL_CONFIG , "" ) ) ; String userToken = retrieveUserToken ( dpmBaseURL , username , password ) ; try { disableDPM ( userToken , organizationId , context ) ; } finally { logout ( dpmBaseURL , userToken ) ; }


signature:
public class SchAdmin { /** * Disable Control Hub on this Data Collector - using existing auth token . */
public static void disableDPM ( String userAuthToken , String organizationId , Context context ) throws IOException { } }

implementation:
// check if DPM enabled
if ( ! context . runtimeInfo . isDPMEnabled ( ) ) { throw new RuntimeException ( "disableDPM is supported only when DPM is enabled" ) ; } String dpmBaseURL = normalizeDpmBaseURL ( context . configuration . get ( RemoteSSOService . DPM_BASE_URL_CONFIG , "" ) ) ; String componentId = context . runtimeInfo . getId ( ) ; // 2 . Deactivate Data Collector System Component
Response response = null ; try { response = ClientBuilder . newClient ( ) . target ( dpmBaseURL + "/security/rest/v1/organization/" + organizationId + "/components/deactivate" ) . register ( new CsrfProtectionFilter ( "CSRF" ) ) . request ( ) . header ( SSOConstants . X_USER_AUTH_TOKEN , userAuthToken ) . header ( SSOConstants . X_REST_CALL , true ) . post ( Entity . json ( ImmutableList . of ( componentId ) ) ) ; if ( response . getStatus ( ) != Response . Status . OK . getStatusCode ( ) ) { throw new RuntimeException ( Utils . format ( " Deactivate Data Collector System Component from DPM failed, status code '{}': {}" , response . getStatus ( ) , response . readEntity ( String . class ) ) ) ; } } finally { if ( response != null ) { response . close ( ) ; } } // 3 . Delete Data Collector System Component
try { response = ClientBuilder . newClient ( ) . target ( dpmBaseURL + "/security/rest/v1/organization/" + organizationId + "/components/delete" ) . register ( new CsrfProtectionFilter ( "CSRF" ) ) . request ( ) . header ( SSOConstants . X_USER_AUTH_TOKEN , userAuthToken ) . header ( SSOConstants . X_REST_CALL , true ) . post ( Entity . json ( ImmutableList . of ( componentId ) ) ) ; if ( response . getStatus ( ) != Response . Status . OK . getStatusCode ( ) ) { throw new RuntimeException ( Utils . format ( " Deactivate Data Collector System Component from DPM failed, status code '{}': {}" , response . getStatus ( ) , response . readEntity ( String . class ) ) ) ; } } finally { if ( response != null ) { response . close ( ) ; } } // 4 . Delete from Job Runner SDC list
try { response = ClientBuilder . newClient ( ) . target ( dpmBaseURL + "/jobrunner/rest/v1/sdc/" + componentId ) . register ( new CsrfProtectionFilter ( "CSRF" ) ) . request ( ) . header ( SSOConstants . X_USER_AUTH_TOKEN , userAuthToken ) . header ( SSOConstants . X_REST_CALL , true ) . delete ( ) ; if ( response . getStatus ( ) != Response . Status . OK . getStatusCode ( ) ) { throw new RuntimeException ( Utils . format ( "Delete from DPM Job Runner SDC list failed, status code '{}': {}" , response . getStatus ( ) , response . readEntity ( String . class ) ) ) ; } } finally { if ( response != null ) { response . close ( ) ; } } // 5 . Update App Token file
updateTokenFile ( context , "" ) ; // 4 . Update dpm . properties file
updateDpmProperties ( context , dpmBaseURL , null , false ) ;


signature:
public class SchAdmin { /** * Normalize Control Hub URL - primarily drop training slash . */
private static String normalizeDpmBaseURL ( String url ) { } }

implementation: if ( url . endsWith ( "/" ) ) { url = url . substring ( 0 , url . length ( ) - 1 ) ; } return url ;


signature:
public class SchAdmin { /** * Login user and retrieve authentication token . */
private static String retrieveUserToken ( String url , String username , String password ) { } }

implementation: Response response = null ; try { Map < String , String > loginJson = new HashMap < > ( ) ; loginJson . put ( "userName" , username ) ; loginJson . put ( "password" , password ) ; response = ClientBuilder . newClient ( ) . target ( url + "/security/public-rest/v1/authentication/login" ) . register ( new CsrfProtectionFilter ( "CSRF" ) ) . request ( ) . post ( Entity . json ( loginJson ) ) ; if ( response . getStatus ( ) != Response . Status . OK . getStatusCode ( ) ) { throw new RuntimeException ( Utils . format ( "DPM Login failed, status code '{}': {}" , response . getStatus ( ) , response . readEntity ( String . class ) ) ) ; } } finally { if ( response != null ) { response . close ( ) ; } } return response . getHeaderString ( SSOConstants . X_USER_AUTH_TOKEN ) ;


signature:
public class SchAdmin { /** * Logout given token . */
private static void logout ( String dpmBaseURL , String userAuthToken ) { } }

implementation: Response response = null ; try { response = ClientBuilder . newClient ( ) . target ( dpmBaseURL + "/security/_logout" ) . register ( new CsrfProtectionFilter ( "CSRF" ) ) . request ( ) . header ( SSOConstants . X_USER_AUTH_TOKEN , userAuthToken ) . cookie ( SSOConstants . AUTHENTICATION_COOKIE_PREFIX + "LOGIN" , userAuthToken ) . get ( ) ; } finally { if ( response != null ) { response . close ( ) ; } }


signature:
public class SchAdmin { /** * Update token file with the SDC access token . */
private static void updateTokenFile ( Context context , String appAuthToken ) throws IOException { } }

implementation: File tokenFile = context . tokenFilePath == null ? new File ( context . runtimeInfo . getConfigDir ( ) , APP_TOKEN_FILE ) : new File ( context . tokenFilePath ) ; DataStore dataStore = new DataStore ( tokenFile ) ; try ( OutputStream os = dataStore . getOutputStream ( ) ) { IOUtils . write ( appAuthToken , os ) ; dataStore . commit ( os ) ; } finally { dataStore . release ( ) ; }


signature:
public class SchAdmin { /** * Update dpm . properties file with new configuration . */
private static void updateDpmProperties ( Context context , String dpmBaseURL , List < String > labels , boolean enableSch ) { } }

implementation: if ( context . skipUpdatingDpmProperties ) { return ; } try { FileBasedConfigurationBuilder < PropertiesConfiguration > builder = new FileBasedConfigurationBuilder < > ( PropertiesConfiguration . class ) . configure ( new Parameters ( ) . properties ( ) . setFileName ( context . runtimeInfo . getConfigDir ( ) + "/dpm.properties" ) . setThrowExceptionOnMissing ( true ) . setListDelimiterHandler ( new DefaultListDelimiterHandler ( ';' ) ) . setIncludesAllowed ( false ) ) ; PropertiesConfiguration config = null ; config = builder . getConfiguration ( ) ; config . setProperty ( RemoteSSOService . DPM_ENABLED , Boolean . toString ( enableSch ) ) ; config . setProperty ( RemoteSSOService . DPM_BASE_URL_CONFIG , dpmBaseURL ) ; config . setProperty ( RemoteSSOService . SECURITY_SERVICE_APP_AUTH_TOKEN_CONFIG , APP_TOKEN_FILE_PROP_VAL ) ; if ( labels != null && labels . size ( ) > 0 ) { config . setProperty ( RemoteEventHandlerTask . REMOTE_JOB_LABELS , StringUtils . join ( labels , ',' ) ) ; } else { config . setProperty ( RemoteEventHandlerTask . REMOTE_JOB_LABELS , "" ) ; } builder . save ( ) ; } catch ( ConfigurationException e ) { throw new RuntimeException ( Utils . format ( "Updating dpm.properties file failed: {}" , e . getMessage ( ) ) , e ) ; }


signature:
public class CTJdbcRunnable { /** * Create record and add it to { @ link com . streamsets . pipeline . api . BatchMaker } */
@ Override public void createAndAddRecord ( ResultSet rs , TableRuntimeContext tableRuntimeContext , BatchContext batchContext ) throws SQLException , StageException { } }

implementation:
ResultSetMetaData md = rs . getMetaData ( ) ; LinkedHashMap < String , Field > fields = jdbcUtil . resultSetToFields ( rs , commonSourceConfigBean , errorRecordHandler , tableJdbcConfigBean . unknownTypeAction , recordHeader , DatabaseVendor . SQL_SERVER ) ; Map < String , String > columnOffsets = new HashMap < > ( ) ; // Generate Offset includes primary keys , sys _ change _ version , and sys _ change _ operation
for ( String key : tableRuntimeContext . getSourceTableContext ( ) . getOffsetColumns ( ) ) { String value = rs . getString ( key ) ; if ( Strings . isNullOrEmpty ( value ) ) { value = fields . get ( key ) != null ? fields . get ( key ) . getValueAsString ( ) : "" ; } columnOffsets . put ( key , value ) ; } columnOffsets . put ( SYS_CHANGE_OPERATION , rs . getString ( SYS_CHANGE_OPERATION ) ) ; String offsetFormat = OffsetQueryUtil . getOffsetFormat ( columnOffsets ) ; Record record = context . createRecord ( tableRuntimeContext . getQualifiedName ( ) + "::" + offsetFormat ) ; record . set ( Field . createListMap ( fields ) ) ; // Set Column Headers
jdbcUtil . setColumnSpecificHeaders ( record , Collections . singleton ( tableRuntimeContext . getSourceTableContext ( ) . getTableName ( ) ) , md , JDBC_NAMESPACE_HEADER ) ; // Set Operation Headers
int op = MSOperationCode . convertToJDBCCode ( rs . getString ( SYS_CHANGE_OPERATION ) ) ; record . getHeader ( ) . setAttribute ( OperationType . SDC_OPERATION_TYPE , String . valueOf ( op ) ) ; for ( String fieldName : recordHeader ) { record . getHeader ( ) . setAttribute ( JDBC_NAMESPACE_HEADER + fieldName , rs . getString ( fieldName ) != null ? rs . getString ( fieldName ) : "NULL" ) ; } int columns = rs . getMetaData ( ) . getColumnCount ( ) ; if ( fields . size ( ) != columns ) { errorRecordHandler . onError ( JdbcErrors . JDBC_35 , fields . size ( ) , columns ) ; return ; // Don ' t output this record .
} else { batchContext . getBatchMaker ( ) . addRecord ( record ) ; } offsets . put ( tableRuntimeContext . getOffsetKey ( ) , offsetFormat ) ;


signature:
public class BulkRecordCreator { /** * When pullMap is called , the caller should have consumed the opening tag for the record */
private Field pullMap ( XMLEventReader reader ) throws StageException , XMLStreamException { } }

implementation:
LinkedHashMap < String , Field > map = new LinkedHashMap < > ( ) ; String type = null ; String fieldValue = null ; while ( reader . hasNext ( ) ) { XMLEvent event = reader . nextEvent ( ) ; if ( event . isStartElement ( ) ) { if ( event . asStartElement ( ) . getName ( ) . getLocalPart ( ) . equals ( TYPE ) ) { // Move to content
event = reader . nextEvent ( ) ; type = event . asCharacters ( ) . getData ( ) . toLowerCase ( ) ; // Consume closing tag
reader . nextEvent ( ) ; } else { String fieldName = event . asStartElement ( ) . getName ( ) . getLocalPart ( ) ; Attribute attr = event . asStartElement ( ) . getAttributeByName ( XSI_TYPE ) ; if ( attr != null && attr . getValue ( ) . equals ( S_OBJECT ) ) { // Element is a nested record
map . put ( fieldName , pullMap ( reader ) ) ; } else { event = reader . nextEvent ( ) ; fieldValue = null ; switch ( event . getEventType ( ) ) { case XMLEvent . START_ELEMENT : // Element is a nested list of records
// Advance over < done > , < queryLocator > to record list
while ( ! ( event . isStartElement ( ) && event . asStartElement ( ) . getName ( ) . getLocalPart ( ) . equals ( RECORDS ) ) ) { event = reader . nextEvent ( ) ; } // Read record list
List < Field > recordList = new ArrayList < > ( ) ; while ( event . isStartElement ( ) && event . asStartElement ( ) . getName ( ) . getLocalPart ( ) . equals ( RECORDS ) ) { recordList . add ( pullMap ( reader ) ) ; event = reader . nextEvent ( ) ; } map . put ( fieldName , Field . create ( recordList ) ) ; break ; case XMLEvent . CHARACTERS : // Element is a field value
fieldValue = event . asCharacters ( ) . getData ( ) ; // Consume closing tag
reader . nextEvent ( ) ; // Intentional fall through to next case !
case XMLEvent . END_ELEMENT : // Create the SDC field
if ( type == null ) { throw new StageException ( Errors . FORCE_38 ) ; } // Is this a relationship to another object ?
com . sforce . soap . partner . Field sfdcField = metadataCache . get ( type ) . getFieldFromRelationship ( fieldName ) ; if ( sfdcField != null ) { // See if we already added fields from the related record
if ( map . get ( fieldName ) != null ) { // We already created this node - don ' t overwrite it !
sfdcField = null ; } } else { sfdcField = getFieldMetadata ( type , fieldName ) ; } if ( sfdcField != null ) { Field field = createField ( fieldValue , sfdcField ) ; if ( conf . createSalesforceNsHeaders ) { setHeadersOnField ( field , getFieldMetadata ( type , fieldName ) ) ; } map . put ( fieldName , field ) ; } break ; default : throw new StageException ( Errors . FORCE_41 , event . getEventType ( ) ) ; } } } } else if ( event . isEndElement ( ) ) { // Done with record
return Field . createListMap ( map ) ; } } throw new StageException ( Errors . FORCE_39 ) ;


signature:
public class AbstractSSOAuthenticator { /** * Terminates the request with an HTTP Unauthorized response */
protected Authentication returnUnauthorized ( HttpServletRequest httpReq , HttpServletResponse httpRes , String principalId , String logMessageTemplate ) throws ServerAuthException { } }

implementation: return returnUnauthorized ( httpReq , httpRes , UNAUTHORIZED_JSON , principalId , logMessageTemplate ) ;


signature:
public class AvroSchemaGenerator { /** * Generate schema for given field and optionally wrap it in union with null if configured . */
private Schema . Field schemaFieldForType ( String fieldPath , Record record , String fieldName , Field field ) throws OnRecordErrorException { } }

implementation:
Schema simpleSchema = simpleSchemaForType ( fieldPath , record , field ) ; Schema finalSchema = simpleSchema ; // If Nullable check box was selected , wrap the whole schema in union with null
if ( getConfig ( ) . avroNullableFields ) { finalSchema = Schema . createUnion ( ImmutableList . of ( Schema . create ( Schema . Type . NULL ) , simpleSchema ) ) ; } return new Schema . Field ( fieldName , finalSchema , null , getDefaultValue ( simpleSchema ) ) ;


signature:
public class AvroSchemaGenerator { /** * Generates complex schema for given field that will include optional union with null and potentially default value
 * as well . Particularly useful to generate nested structures . */
private Schema complexSchemaForType ( String fieldPath , Record record , Field field ) throws OnRecordErrorException { } }

implementation: Schema simpleSchema = simpleSchemaForType ( fieldPath , record , field ) ; Schema finalSchema = simpleSchema ; if ( getConfig ( ) . avroNullableFields ) { finalSchema = Schema . createUnion ( ImmutableList . of ( Schema . create ( Schema . Type . NULL ) , simpleSchema ) ) ; } JsonNode defaultValue = getDefaultValue ( simpleSchema ) ; if ( defaultValue != null ) { finalSchema . addProp ( "defaultValue" , defaultValue ) ; } return finalSchema ;


signature:
public class AvroSchemaGenerator { /** * Generate simple schema for given field - it will never contain union nor a default value . Particularly useful
 * for generating Schema . Field as this will not convert simple " string " to " { type : string , defaultValue : something } "
 * which is hard to undo . */
private Schema simpleSchemaForType ( String fieldPath , Record record , Field field ) throws OnRecordErrorException { } }

implementation:
switch ( field . getType ( ) ) { // Primitive types
case BOOLEAN : return Schema . create ( Schema . Type . BOOLEAN ) ; case INTEGER : return Schema . create ( Schema . Type . INT ) ; case LONG : return Schema . create ( Schema . Type . LONG ) ; case FLOAT : return Schema . create ( Schema . Type . FLOAT ) ; case DOUBLE : return Schema . create ( Schema . Type . DOUBLE ) ; case STRING : return Schema . create ( Schema . Type . STRING ) ; case BYTE_ARRAY : return Schema . create ( Schema . Type . BYTES ) ; // Logical types
case DECIMAL : int precision = getDecimalScaleOrPrecision ( record , field , getConfig ( ) . precisionAttribute , getConfig ( ) . defaultPrecision , 1 ) ; int scale = getDecimalScaleOrPrecision ( record , field , getConfig ( ) . scaleAttribute , getConfig ( ) . defaultScale , 0 ) ; Schema decimalSchema = Schema . create ( Schema . Type . BYTES ) ; decimalSchema . addProp ( AvroTypeUtil . LOGICAL_TYPE , AvroTypeUtil . LOGICAL_TYPE_DECIMAL ) ; decimalSchema . addProp ( AvroTypeUtil . LOGICAL_TYPE_ATTR_PRECISION , new IntNode ( precision ) ) ; decimalSchema . addProp ( AvroTypeUtil . LOGICAL_TYPE_ATTR_SCALE , new IntNode ( scale ) ) ; return decimalSchema ; case DATE : Schema dateSchema = Schema . create ( Schema . Type . INT ) ; dateSchema . addProp ( AvroTypeUtil . LOGICAL_TYPE , AvroTypeUtil . LOGICAL_TYPE_DATE ) ; return dateSchema ; case TIME : Schema timeSchema = Schema . create ( Schema . Type . INT ) ; timeSchema . addProp ( AvroTypeUtil . LOGICAL_TYPE , AvroTypeUtil . LOGICAL_TYPE_TIME_MILLIS ) ; return timeSchema ; case DATETIME : Schema dateTimeSchema = Schema . create ( Schema . Type . LONG ) ; dateTimeSchema . addProp ( AvroTypeUtil . LOGICAL_TYPE , AvroTypeUtil . LOGICAL_TYPE_TIMESTAMP_MILLIS ) ; return dateTimeSchema ; // Complex types
case LIST : // In avro list must be of the same type - which is not true with our records
// We can ' t generate the list type from empty list
if ( field . getValueAsList ( ) . isEmpty ( ) ) { throw new OnRecordErrorException ( record , Errors . SCHEMA_GEN_0006 , fieldPath ) ; } // And all items in the list must have the same schema
Schema itemSchema = null ; int index = 0 ; for ( Field listItem : field . getValueAsList ( ) ) { Schema currentListItemSchema = complexSchemaForType ( fieldPath + "[" + index + "]" , record , listItem ) ; if ( itemSchema == null ) { itemSchema = currentListItemSchema ; } else if ( ! itemSchema . equals ( currentListItemSchema ) ) { throw new OnRecordErrorException ( record , Errors . SCHEMA_GEN_0005 , fieldPath , itemSchema , currentListItemSchema ) ; } index ++ ; } return Schema . createArray ( itemSchema ) ; case MAP : case LIST_MAP : // In avro maps must have key of string ( same as us ) and value must be the same for all items - which
// is different then our records .
// We can ' t generate the map value type from null or empty map
if ( field . getValueAsMap ( ) == null || field . getValueAsMap ( ) . isEmpty ( ) ) { throw new OnRecordErrorException ( record , Errors . SCHEMA_GEN_0008 , fieldPath ) ; } // And all values in the map must be the same
Schema mapSchema = null ; for ( Map . Entry < String , Field > item : field . getValueAsMap ( ) . entrySet ( ) ) { Schema currentListItemSchema = complexSchemaForType ( fieldPath + "/" + item . getKey ( ) , record , item . getValue ( ) ) ; if ( mapSchema == null ) { mapSchema = currentListItemSchema ; } else if ( ! mapSchema . equals ( currentListItemSchema ) ) { throw new OnRecordErrorException ( record , Errors . SCHEMA_GEN_0007 , fieldPath , mapSchema , currentListItemSchema ) ; } } return Schema . createMap ( mapSchema ) ; // Types that does not have direct equivalent in Avro
case SHORT : if ( getConfig ( ) . avroExpandTypes ) { return Schema . create ( Schema . Type . INT ) ; } // fall through
case CHAR : if ( getConfig ( ) . avroExpandTypes ) { return Schema . create ( Schema . Type . STRING ) ; } // fall through
// Not supported data types :
case BYTE : case FILE_REF : default : throw new OnRecordErrorException ( record , Errors . SCHEMA_GEN_0002 , field . getType ( ) ) ; }


signature:
public class AvroSchemaGenerator { /** * Resolve parameters of decimal type . */
private int getDecimalScaleOrPrecision ( Record record , Field field , String attributeName , int defaultValue , int minAllowed ) throws OnRecordErrorException { } }

implementation:
int finalValue = - 1 ; // Invalid value
// Firstly try the field attribute
String stringValue = field . getAttribute ( attributeName ) ; if ( ! StringUtils . isEmpty ( stringValue ) ) { finalValue = Integer . valueOf ( stringValue ) ; } // If it ' s invalid , then use the default value
if ( finalValue < minAllowed ) { finalValue = defaultValue ; } // If even the default value is invalid , then send the record to error
if ( finalValue < minAllowed ) { throw new OnRecordErrorException ( record , Errors . SCHEMA_GEN_0004 , finalValue , field ) ; } return finalValue ;


signature:
public class AvroSchemaGenerator { /** * Returns default value for given field or null if no default value should be used . */
private JsonNode getDefaultValue ( Schema schema ) { } }

implementation: if ( getConfig ( ) . avroNullableFields && getConfig ( ) . avroDefaultNullable ) { return NullNode . getInstance ( ) ; } if ( ! getConfig ( ) . avroNullableFields && defaultValuesForTypes . containsKey ( schema . getType ( ) ) ) { return defaultValuesForTypes . get ( schema . getType ( ) ) ; } return null ;


signature:
public class FileRefUtil { /** * Creates a gauge if it is already not . This is done only once for the stage
 * @ param context the { @ link com . streamsets . pipeline . api . Stage . Context } of this stage */
@ SuppressWarnings ( "unchecked" ) public static synchronized void initMetricsIfNeeded ( ProtoConfigurableEntity . Context context ) { } }

implementation:
Gauge < Map < String , Object > > gauge = context . getGauge ( fileStatisticGaugeName ( context ) ) ; if ( gauge == null ) { gauge = context . createGauge ( fileStatisticGaugeName ( context ) , Comparator . comparing ( GAUGE_MAP_ORDERING :: get ) ) ; Map < String , Object > gaugeStatistics = gauge . getValue ( ) ; // File name is populated at the MetricEnabledWrapperStream .
gaugeStatistics . put ( FileRefUtil . FILE , "" ) ; gaugeStatistics . put ( FileRefUtil . TRANSFER_THROUGHPUT , 0L ) ; gaugeStatistics . put ( FileRefUtil . SENT_BYTES , String . format ( FileRefUtil . BRACKETED_TEMPLATE , 0 , 0 ) ) ; gaugeStatistics . put ( FileRefUtil . REMAINING_BYTES , 0L ) ; gaugeStatistics . put ( FileRefUtil . COMPLETED_FILE_COUNT , 0L ) ; } Meter dataTransferMeter = context . getMeter ( FileRefUtil . TRANSFER_THROUGHPUT_METER ) ; if ( dataTransferMeter == null ) { context . createMeter ( FileRefUtil . TRANSFER_THROUGHPUT_METER ) ; }


signature:
public class LocalFileSystem { /** * This method is a simple wrapper that lets us find the NoSuchFileException if that was the cause . */
public int compare ( WrappedFile path1 , WrappedFile path2 , boolean useLastModified ) { } }

implementation:
// why not just check if the file exists ? Well , there is a possibility file gets moved / archived / deleted right after
// that check . In that case we will still fail . So fail , and recover .
try { if ( useLastModified && ! exists ( path2 ) ) { return 1 ; } return getComparator ( useLastModified ) . compare ( path1 , path2 ) ; } catch ( RuntimeException ex ) { Throwable cause = ex . getCause ( ) ; // Happens only in timestamp ordering .
// Very unlikely this will happen , new file has to be added to the queue at the exact time when
// the currentFile was consumed and archived while a new file has not yet been picked up for processing .
// Ignore - we just add the new file , since this means this file is indeed newer
// ( else this would have been consumed and archived first )
if ( cause != null && cause instanceof NoSuchFileException ) { LOG . debug ( "Starting file may have already been archived." , cause ) ; return 1 ; } LOG . warn ( "Error while comparing files" , ex ) ; throw ex ; }


signature:
public class Pattern { /** * Gets the group index of a named capture group at the
 * specified index . If only one instance of the named
 * group exists , use index 0.
 * @ param groupName name of capture group
 * @ param index the instance index of the named capture group within
 * the pattern ; e . g . , index is 2 for the third instance
 * @ return group index or - 1 if not found
 * @ throws IndexOutOfBoundsException if instance index is out of bounds */
public int indexOf ( String groupName , int index ) { } }

implementation: int idx = - 1 ; if ( groupInfo . containsKey ( groupName ) ) { List < GroupInfo > list = groupInfo . get ( groupName ) ; idx = list . get ( index ) . groupIndex ( ) ; } return idx ;