signature:
public class ShellExecutor { /** * Attempts to retrieve PID from internal JVM classes . This method is not guaranteed to work as JVM is free
 * to change their implementation at will . Hence the return value should be only used for troubleshooting or
 * debug and not for main functionality . */
private static int retrievePidIfFeasible ( Process process ) { } }

implementation: if ( unixProcessClass == null ) { return UNDETERMINED_PID ; } if ( ! unixProcessClass . isInstance ( process ) ) { LOG . debug ( "Do not support retrieving PID from {}" , process . getClass ( ) . getName ( ) ) ; return UNDETERMINED_PID ; } try { return ( int ) pidField . get ( process ) ; } catch ( IllegalAccessException e ) { LOG . debug ( "Can't retrieve PID value from the field" , e ) ; return UNDETERMINED_PID ; }


signature:
public class FieldHasherProcessor { /** * Basically throw out map , list map , list and null values fields . */
private Set < String > validateAndExtractFieldsToHash ( Record record , Set < String > fieldsDontExist , Set < String > fieldsWithListOrMapType , Set < String > fieldsWithNull , Collection < String > matchingFieldsPath ) { } }

implementation: Set < String > validFieldsToHashForThisConfig = new HashSet < String > ( ) ; for ( String matchingFieldPath : matchingFieldsPath ) { if ( record . has ( matchingFieldPath ) ) { Field field = record . get ( matchingFieldPath ) ; if ( UNSUPPORTED_FIELD_TYPES . contains ( field . getType ( ) ) ) { fieldsWithListOrMapType . add ( matchingFieldPath ) ; } else if ( field . getValue ( ) == null ) { fieldsWithNull . add ( matchingFieldPath ) ; } else { validFieldsToHashForThisConfig . add ( matchingFieldPath ) ; } } else { fieldsDontExist . add ( matchingFieldPath ) ; } } return validFieldsToHashForThisConfig ;


signature:
public class HadoopSecurityUtil { /** * Return UGI object that should be used for any remote operation .
 * This object will be impersonate according to the configuration . This method is meant to be called once during
 * initialization and it ' s expected that caller will cache the result for a lifetime of the stage execution . */
public static UserGroupInformation getProxyUser ( String user , // Hadoop user ( HDFS User , HBase user , generally the to - be - impersonated user in component ' s configuration )
Stage . Context context , // Stage context object
UserGroupInformation loginUser , // Login UGI ( sdc user )
List < Stage . ConfigIssue > issues , // Reports errors
String configGroup , // Group where " HDFS User " is present
String configName // Config name of " HDFS User "
) { } }

implementation:
// Should we always impersonate current user ?
boolean alwaysImpersonate = context . getConfiguration ( ) . get ( HadoopConfigConstants . IMPERSONATION_ALWAYS_CURRENT_USER , false ) ; // If so , propagate current user to " user " ( the one to be impersonated )
if ( alwaysImpersonate ) { if ( ! StringUtils . isEmpty ( user ) ) { issues . add ( context . createConfigIssue ( configGroup , configName , Errors . HADOOP_00001 ) ) ; } user = context . getUserContext ( ) . getAliasName ( ) ; } // If impersonated user is empty , simply return login UGI ( no impersonation performed )
if ( StringUtils . isEmpty ( user ) ) { return loginUser ; } // Optionally lower case the user name
boolean lowerCase = context . getConfiguration ( ) . get ( HadoopConfigConstants . LOWERCASE_USER , false ) ; if ( lowerCase ) { user = user . toLowerCase ( ) ; } return UserGroupInformation . createProxyUser ( user , loginUser ) ;


signature:
public class ActiveStats { /** * returns fresh bean with same UsageTimers just reset to zero accumulated time to be used as the new live stats */
public ActiveStats roll ( ) { } }

implementation: long now = System . currentTimeMillis ( ) ; setEndTime ( now ) ; ActiveStats statsBean = new ActiveStats ( ) . setStartTime ( now ) . setDataCollectorVersion ( getDataCollectorVersion ( ) ) . setDpmEnabled ( isDpmEnabled ( ) ) . setUpTime ( getUpTime ( ) . roll ( ) ) ; statsBean . setPipelines ( getPipelines ( ) . stream ( ) . map ( UsageTimer :: roll ) . collect ( Collectors . toList ( ) ) ) ; statsBean . setStages ( getStages ( ) . stream ( ) . filter ( timer -> timer . getMultiplier ( ) > 0 ) . map ( UsageTimer :: roll ) . collect ( Collectors . toList ( ) ) ) ; return statsBean ;


signature:
public class ActiveStats { /** * returns a snapshot for persistency */
public ActiveStats snapshot ( ) { } }

implementation: ActiveStats snapshot = new ActiveStats ( ) . setStartTime ( getStartTime ( ) ) . setDataCollectorVersion ( getDataCollectorVersion ( ) ) . setDpmEnabled ( isDpmEnabled ( ) ) . setUpTime ( getUpTime ( ) . snapshot ( ) ) . setRecordCount ( getRecordCount ( ) ) ; snapshot . setPipelines ( getPipelines ( ) . stream ( ) . map ( UsageTimer :: snapshot ) . collect ( Collectors . toList ( ) ) ) ; snapshot . setStages ( getStages ( ) . stream ( ) . map ( UsageTimer :: snapshot ) . collect ( Collectors . toList ( ) ) ) ; return snapshot ;


signature:
public class HdfsMetadataExecutor { /** * Ensure that given directory exists .
 * Creates the directory if it doesn ' t exists . No - op if it does . */
private void ensureDirectoryExists ( FileSystem fs , Path path ) throws IOException { } }

implementation: if ( ! fs . exists ( path ) ) { LOG . debug ( "Creating directory: {}" , path ) ; if ( ! fs . mkdirs ( path ) ) { throw new IOException ( "Can't create directory: " + path ) ; } }


signature:
public class VaultCredentialStore { /** * Name format is Vault ' s [ path ] @ [ key ] */
@ Override public CredentialValue get ( String group , String name , String credentialStoreOptions ) throws StageException { } }

implementation: Utils . checkNotNull ( group , "group cannot be NULL" ) ; Utils . checkNotNull ( name , "name cannot be NULL" ) ; try { Map < String , String > optionsMap = Splitter . on ( "," ) . omitEmptyStrings ( ) . trimResults ( ) . withKeyValueSeparator ( "=" ) . split ( credentialStoreOptions ) ; String separator = optionsMap . get ( SEPARATOR_OPTION ) ; if ( separator == null ) { separator = pathKeySeparator ; } String [ ] splits = name . split ( separator , 2 ) ; if ( splits . length != 2 ) { throw new IllegalArgumentException ( Utils . format ( "Vault CredentialStore name '{}' should be <path>{}<key>" , name , separator ) ) ; } String delayStr = optionsMap . get ( DELAY_OPTION ) ; long delay = ( delayStr == null ) ? 0 : Long . parseLong ( delayStr ) ; CredentialValue credential = new VaultCredentialValue ( splits [ 0 ] , splits [ 1 ] , delay ) ; credential . get ( ) ; return credential ; } catch ( Exception ex ) { throw new StageException ( Errors . VAULT_001 , name , ex ) ; }


signature:
public class JdbcUtil { /** * Formats the error message of a { @ link java . sql . SQLException } for human consumption .
 * @ param ex SQLException
 * @ return Formatted string with database - specific error code , error message , and SQLState */
public String formatSqlException ( SQLException ex ) { } }

implementation: StringBuilder sb = new StringBuilder ( ) ; Set < String > messages = new HashSet < > ( ) ; for ( Throwable e : ex ) { if ( e instanceof SQLException ) { String message = e . getMessage ( ) ; if ( ! messages . add ( message ) ) { continue ; } sb . append ( "SQLState: " + ( ( SQLException ) e ) . getSQLState ( ) + "\n" ) . append ( "Error Code: " + ( ( SQLException ) e ) . getErrorCode ( ) + "\n" ) . append ( "Message: " + message + "\n" ) ; Throwable t = ex . getCause ( ) ; while ( t != null ) { if ( messages . add ( t . getMessage ( ) ) ) { sb . append ( "Cause: " + t + "\n" ) ; } t = t . getCause ( ) ; } } } return sb . toString ( ) ;


signature:
public class JdbcUtil { /** * Wrapper for { @ link Connection # getCatalog ( ) } to solve problems with RDBMs for which catalog and schema are the same
 * thing . For these RDBMs , it returns the schema name when it is not - null and not - empty ; otherwise , it returns the
 * value of java . sql . Connection . getCatalog ( ) . For other RDBMs , it returns always the value of
 * java . sql . Connection . getCatalog ( ) .
 * @ param connection An open JDBC connection
 * @ param schema The schema name we want to use
 * @ return The current catalog or the schema name passed as argument
 * @ throws SQLException */
private String getCatalog ( Connection connection , String schema ) throws SQLException { } }

implementation: if ( Strings . isNullOrEmpty ( schema ) ) { return connection . getCatalog ( ) ; } String name = connection . getMetaData ( ) . getDatabaseProductName ( ) . toLowerCase ( ) ; for ( String d : RDBMS_WITHOUT_SCHEMAS ) { if ( name . contains ( d ) ) { return schema ; } } return connection . getCatalog ( ) ;


signature:
public class JdbcUtil { /** * Wrapper for { @ link java . sql . DatabaseMetaData # getColumns ( String , String , String , String ) } that detects
 * the format of the supplied tableName .
 * @ param connection An open JDBC connection
 * @ param tableName table name that is optionally fully qualified with a schema in the form schema . tableName
 * @ return ResultSet containing the column metadata
 * @ throws SQLException */
public ResultSet getColumnMetadata ( Connection connection , String schema , String tableName ) throws SQLException { } }

implementation:
DatabaseMetaData metadata = connection . getMetaData ( ) ; // Get all columns for this table
return metadata . getColumns ( getCatalog ( connection , schema ) , schema , tableName , null ) ;


signature:
public class JdbcUtil { /** * Wrapper for { @ link java . sql . DatabaseMetaData # getTables ( String , String , String , String [ ] ) }
 * @ param connection open JDBC connection
 * @ param schema schema name , can be null
 * @ param tableName table name or pattern , optionally fully qualified in the form schema . tableName
 * @ return ResultSet containing the table and view metadata
 * @ throws SQLException */
public ResultSet getTableAndViewMetadata ( Connection connection , String schema , String tableName ) throws SQLException { } }

implementation: return connection . getMetaData ( ) . getTables ( getCatalog ( connection , schema ) , schema , tableName , METADATA_TABLE_VIEW_TYPE ) ;


signature:
public class JdbcUtil { /** * Wrapper for { @ link java . sql . DatabaseMetaData # getTables ( String , String , String , String [ ] ) }
 * @ param connection open JDBC connection
 * @ param schema schema name , can be null
 * @ param tableName table name or pattern , optionally fully qualified in the form schema . tableName
 * @ return ResultSet containing the table metadata
 * @ throws SQLException */
public ResultSet getTableMetadata ( Connection connection , String schema , String tableName ) throws SQLException { } }

implementation: DatabaseMetaData metadata = connection . getMetaData ( ) ; return metadata . getTables ( getCatalog ( connection , schema ) , schema , tableName , METADATA_TABLE_TYPE ) ;


signature:
public class JdbcUtil { /** * Wrapper for { @ link java . sql . DatabaseMetaData # getPrimaryKeys ( String , String , String ) }
 * @ param connection An open JDBC connection
 * @ param tableName table name that is optionally fully qualified with a schema in the form schema . tableName
 * @ return List of primary key column names for a table
 * @ throws SQLException */
public List < String > getPrimaryKeys ( Connection connection , String schema , String tableName ) throws SQLException { } }

implementation: String table = tableName ; DatabaseMetaData metadata = connection . getMetaData ( ) ; List < String > keys = new ArrayList < > ( ) ; try ( ResultSet result = metadata . getPrimaryKeys ( getCatalog ( connection , schema ) , schema , table ) ) { while ( result . next ( ) ) { keys . add ( result . getString ( COLUMN_NAME ) ) ; } } return keys ;


signature:
public class JdbcUtil { /** * Wrapper for { @ link java . sql . DatabaseMetaData # getImportedKeys ( String , String , String ) }
 * @ param connection An open JDBC connection
 * @ param tableName table name that is optionally fully qualified with a schema in the form schema . tableName
 * @ return List of Table Names whose primary key are referred as foreign key by the table tableName
 * @ throws SQLException */
public Set < String > getReferredTables ( Connection connection , String schema , String tableName ) throws SQLException { } }

implementation: DatabaseMetaData metadata = connection . getMetaData ( ) ; ResultSet result = metadata . getImportedKeys ( getCatalog ( connection , schema ) , schema , tableName ) ; Set < String > referredTables = new HashSet < > ( ) ; while ( result . next ( ) ) { referredTables . add ( result . getString ( PK_TABLE_NAME ) ) ; } return referredTables ;


signature:
public class JdbcUtil { /** * Write records to potentially different schemas and tables using EL expressions , and handle errors .
 * @ param batch batch of SDC records
 * @ param schemaTableClassifier classifier to group records according to the schema and table names , resolving the
 * EL expressions involved .
 * @ param recordWriters JDBC record writer cache
 * @ param errorRecordHandler error record handler
 * @ param perRecord indicate record or batch update
 * @ throws StageException */
public void write ( Batch batch , SchemaTableClassifier schemaTableClassifier , LoadingCache < SchemaAndTable , JdbcRecordWriter > recordWriters , ErrorRecordHandler errorRecordHandler , boolean perRecord ) throws StageException { } }

implementation: Multimap < SchemaAndTable , Record > partitions = schemaTableClassifier . classify ( batch ) ; for ( SchemaAndTable key : partitions . keySet ( ) ) { Iterator < Record > recordIterator = partitions . get ( key ) . iterator ( ) ; write ( recordIterator , key , recordWriters , errorRecordHandler , perRecord ) ; }


signature:
public class JdbcUtil { /** * Write records to the evaluated tables and handle errors .
 * @ param batch batch of SDC records
 * @ param tableNameEval table name EL eval
 * @ param tableNameVars table name EL vars
 * @ param tableNameTemplate table name template
 * @ param recordWriters JDBC record writer cache
 * @ param errorRecordHandler error record handler
 * @ param perRecord indicate record or batch update
 * @ throws StageException */
public void write ( Batch batch , ELEval tableNameEval , ELVars tableNameVars , String tableNameTemplate , LoadingCache < String , JdbcRecordWriter > recordWriters , ErrorRecordHandler errorRecordHandler , boolean perRecord ) throws StageException { } }

implementation: Multimap < String , Record > partitions = ELUtils . partitionBatchByExpression ( tableNameEval , tableNameVars , tableNameTemplate , batch ) ; for ( String tableName : partitions . keySet ( ) ) { Iterator < Record > recordIterator = partitions . get ( tableName ) . iterator ( ) ; write ( recordIterator , tableName , recordWriters , errorRecordHandler , perRecord ) ; }


signature:
public class JdbcUtil { /** * Write records to a JDBC destination using the recordWriter specified by key , and handle errors
 * @ param recordIterator iterator of SDC records
 * @ param key key to select the recordWriter
 * @ param recordWriters JDBC record writer cache
 * @ param errorRecordHandler error record handler
 * @ param perRecord indicate record or batch update
 * @ throws StageException */
public < T > void write ( Iterator < Record > recordIterator , T key , LoadingCache < T , JdbcRecordWriter > recordWriters , ErrorRecordHandler errorRecordHandler , boolean perRecord ) throws StageException { } }

implementation:
final JdbcRecordWriter jdbcRecordWriter ; try { jdbcRecordWriter = recordWriters . getUnchecked ( key ) ; } catch ( UncheckedExecutionException ex ) { final Throwable throwable = ex . getCause ( ) ; final ErrorCode errorCode ; final Object [ ] messageParams ; if ( throwable instanceof StageException ) { StageException stageEx = ( StageException ) ex . getCause ( ) ; errorCode = stageEx . getErrorCode ( ) ; messageParams = stageEx . getParams ( ) ; } else { errorCode = JdbcErrors . JDBC_301 ; messageParams = new Object [ ] { ex . getMessage ( ) , ex . getCause ( ) } ; } // Failed to create RecordWriter , report all as error records .
while ( recordIterator . hasNext ( ) ) { Record record = recordIterator . next ( ) ; errorRecordHandler . onError ( new OnRecordErrorException ( record , errorCode , messageParams ) ) ; } return ; } List < OnRecordErrorException > errors = perRecord ? jdbcRecordWriter . writePerRecord ( recordIterator ) : jdbcRecordWriter . writeBatch ( recordIterator ) ; for ( OnRecordErrorException error : errors ) { errorRecordHandler . onError ( error ) ; }


signature:
public class JdbcUtil { /** * Generates the no - more - data event */
public void generateNoMoreDataEvent ( PushSource . Context context ) { } }

implementation: LOG . info ( "No More data to process, Triggered No More Data Event" ) ; BatchContext batchContext = context . startBatch ( ) ; CommonEvents . NO_MORE_DATA . create ( context , batchContext ) . createAndSend ( ) ; context . processBatch ( batchContext ) ;


signature:
public class HiveMetadataProcessor { /** * Using partition name and value that were obtained from record , compare them
 * with cached partition .
 * @ param partitionValues Partition representation
 * @ param pCache Cache that has existing partitions
 * @ return Diff of partitions if new partition is detected . Otherwise null .
 * @ throws StageException */
private Map < PartitionInfoCacheSupport . PartitionValues , String > detectNewPartition ( PartitionInfoCacheSupport . PartitionValues partitionValues , PartitionInfoCacheSupport . PartitionInfo pCache , String location ) throws StageException { } }

implementation: Map < PartitionInfoCacheSupport . PartitionValues , String > partitionInfoDiff = new HashMap < > ( ) ; partitionInfoDiff . put ( partitionValues , location ) ; partitionInfoDiff = ( pCache != null ) ? pCache . getDiff ( partitionInfoDiff ) : partitionInfoDiff ; if ( pCache == null || ! partitionInfoDiff . isEmpty ( ) ) { return partitionInfoDiff ; } return null ;


signature:
public class HiveMetadataProcessor { /** * Obtain a list of partition values from record .
 * @ return LinkedHashMap that contains pairs of partition name - values
 * @ throws StageException */
@ VisibleForTesting LinkedHashMap < String , String > getPartitionValuesFromRecord ( ELVars variables ) throws StageException { } }

implementation:
LinkedHashMap < String , String > values = new LinkedHashMap < > ( ) ; for ( PartitionConfig pName : partitionConfigList ) { String ret = HiveMetastoreUtil . resolveEL ( elEvals . partitionValueELEval , variables , pName . valueEL ) ; if ( ret == null || ret . isEmpty ( ) ) { // If no partition value is found in record , this record goes to Error Record
throw new HiveStageCheckedException ( Errors . HIVE_METADATA_02 , pName . valueEL ) ; } else if ( HiveMetastoreUtil . hasUnsupportedChar ( ret ) ) { throw new HiveStageCheckedException ( Errors . HIVE_METADATA_10 , pName . valueEL , ret ) ; } values . put ( pName . name . toLowerCase ( ) , ret ) ; } return values ;


signature:
public class HiveMetadataProcessor { /** * Generate a record for new partition . It creates a new Record and fill in metadata .
 * @ param database database name
 * @ param tableName table name
 * @ param partitionList New partition to be created
 * @ param location Directory path in Hadoop filesystem for the new partition .
 * @ param customLocation Whether the { @ code location } is a custom path or the default path defined by Hive .
 * @ return New metadata record
 * @ throws StageException */
@ VisibleForTesting Record generateNewPartitionRecord ( String database , String tableName , LinkedHashMap < String , String > partitionList , String location , boolean customLocation , Map < String , String > metadataHeaderAttributes ) throws StageException { } }

implementation:
// creating a record with uuid as postfix so multiple SDCs won ' t generate the record with same id .
Record metadataRecord = getContext ( ) . createRecord ( "Partition Metadata Record" + UUID . randomUUID ( ) . toString ( ) ) ; Field metadataField = HiveMetastoreUtil . newPartitionMetadataFieldBuilder ( database , tableName , partitionList , location , customLocation , dataFormat ) ; metadataRecord . set ( metadataField ) ; for ( Map . Entry < String , String > entry : metadataHeaderAttributes . entrySet ( ) ) { metadataRecord . getHeader ( ) . setAttribute ( entry . getKey ( ) , entry . getValue ( ) ) ; } return metadataRecord ;


signature:
public class HiveMetadataProcessor { /** * Add header information to send to HDFS */
@ VisibleForTesting static void updateRecordForHDFS ( Record record , boolean roll , String avroSchema , String location ) { } }

implementation: if ( roll ) { record . getHeader ( ) . setAttribute ( HDFS_HEADER_ROLL , "true" ) ; } record . getHeader ( ) . setAttribute ( HDFS_HEADER_AVROSCHEMA , avroSchema ) ; record . getHeader ( ) . setAttribute ( HDFS_HEADER_TARGET_DIRECTORY , location ) ; LOG . trace ( "Record {} will be stored in {} path: roll({}), avro schema: {}" , record . getHeader ( ) . getSourceId ( ) , location , roll , avroSchema ) ;


signature:
public class GroupByAggregator { /** * Processes the given value into the corresponding group - by element of the aggregator .
 * @ param group the group - by element .
 * @ param value the value to process . */
public void process ( String group , T value ) { } }

implementation: getData ( ) . process ( ImmutableMap . of ( group , value ) ) ;


signature:
public class StageValidator { /** * Returns if and only if both stage classes have defined the same version . */
public static boolean isSameVersion ( Class < ? extends Stage > a , Class < ? extends Stage > b ) { } }

implementation: StageDef aDef = a . getAnnotation ( StageDef . class ) ; StageDef bDef = b . getAnnotation ( StageDef . class ) ; return aDef . version ( ) == bDef . version ( ) ;


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

implementation: BootstrapCluster . printSystemPropsEnvVariables ( ) ; String mesosDir = System . getenv ( "MESOS_DIRECTORY" ) ; if ( mesosDir == null ) { throw new IllegalStateException ( "Expected the env. variable MESOS_DIRECTORY to be defined" ) ; } File mesosHomeDir = new File ( mesosDir ) ; String sparkDir = System . getenv ( "SPARK_HOME" ) ; if ( sparkDir == null ) { throw new IllegalStateException ( "Expected the env. variable SPARK_HOME to be defined" ) ; } File sparkHomeDir = new File ( sparkDir ) ; int processExitValue = BootstrapCluster . findAndExtractJar ( mesosHomeDir , sparkHomeDir ) ; if ( processExitValue != 0 ) { throw new IllegalStateException ( "Process extracting archives from uber jar exited abnormally; check Mesos driver stdout file" ) ; } System . setProperty ( "SDC_MESOS_BASE_DIR" , new File ( mesosHomeDir , BootstrapCluster . SDC_MESOS_BASE_DIR ) . getAbsolutePath ( ) ) ; final Class < ? > clazz = Class . forName ( "com.streamsets.pipeline.BootstrapClusterStreaming" ) ; final Method method = clazz . getMethod ( "main" , String [ ] . class ) ; method . invoke ( null , new Object [ ] { args } ) ;


signature:
public class ProductionPipelineRunnable { /** * This function is called while thread is waiting at countDownLatch . await in stop ( ) .
 * It cancels tasks , change pipeline state and call countDown ( ) , so that the waiting thread
 * can proceed to terminate . */
public void forceQuit ( ) { } }

implementation: synchronized ( relatedTasks ) { if ( runningThread != null ) { runningThread . interrupt ( ) ; runningThread = null ; cancelTask ( ) ; postStop ( ) ; } } countDownLatch . countDown ( ) ;


signature:
public class HttpClientCommon { /** * Helper to apply authentication properties to Jersey client .
 * @ param clientBuilder Jersey Client builder to configure */
private void configureAuthAndBuildClient ( ClientBuilder clientBuilder , List < Stage . ConfigIssue > issues ) { } }

implementation:
if ( jerseyClientConfig . authType == AuthenticationType . OAUTH ) { String consumerKey = jerseyClientConfig . oauth . resolveConsumerKey ( context , "CREDENTIALS" , "conf.oauth." , issues ) ; String consumerSecret = jerseyClientConfig . oauth . resolveConsumerSecret ( context , "CREDENTIALS" , "conf.oauth." , issues ) ; String token = jerseyClientConfig . oauth . resolveToken ( context , "CREDENTIALS" , "conf.oauth." , issues ) ; String tokenSecret = jerseyClientConfig . oauth . resolveTokenSecret ( context , "CREDENTIALS" , "conf.oauth." , issues ) ; if ( issues . isEmpty ( ) ) { authToken = JerseyClientUtil . configureOAuth1 ( consumerKey , consumerSecret , token , tokenSecret , clientBuilder ) ; } } else if ( jerseyClientConfig . authType . isOneOf ( AuthenticationType . DIGEST , AuthenticationType . BASIC , AuthenticationType . UNIVERSAL ) ) { String username = jerseyClientConfig . basicAuth . resolveUsername ( context , "CREDENTIALS" , "conf.basicAuth." , issues ) ; String password = jerseyClientConfig . basicAuth . resolvePassword ( context , "CREDENTIALS" , "conf.basicAuth." , issues ) ; if ( issues . isEmpty ( ) ) { JerseyClientUtil . configurePasswordAuth ( jerseyClientConfig . authType , username , password , clientBuilder ) ; } } try { buildNewAuthenticatedClient ( issues , false ) ; clientInitialized = true ; } catch ( StageException e ) { // should not happen , since we passed throwExceptions as false above
ExceptionUtils . throwUndeclared ( e ) ; }


signature:
public class HttpClientCommon { /** * Returns true if the request contains potentially sensitive information such as a vault : read EL .
 * @ return whether or not the request had sensitive information detected . */
public boolean requestContainsSensitiveInfo ( Map < String , String > headers , String requestBody ) { } }

implementation: boolean sensitive = false ; for ( Map . Entry < String , String > header : headers . entrySet ( ) ) { if ( header . getKey ( ) . contains ( VAULT_EL_PREFIX ) || header . getValue ( ) . contains ( VAULT_EL_PREFIX ) ) { sensitive = true ; break ; } } if ( requestBody != null && requestBody . contains ( VAULT_EL_PREFIX ) ) { sensitive = true ; } return sensitive ;


signature:
public class HttpClientCommon { /** * Evaluates any EL expressions in the headers section of the stage configuration .
 * @ param record current record in context for EL evaluation
 * @ return Map of headers that can be added to the Jersey Client request
 * @ throws StageException if an expression could not be evaluated */
public MultivaluedMap < String , Object > resolveHeaders ( Map < String , String > headers , Record record ) throws StageException { } }

implementation: RecordEL . setRecordInContext ( headerVars , record ) ; MultivaluedMap < String , Object > requestHeaders = new MultivaluedHashMap < > ( ) ; for ( Map . Entry < String , String > entry : headers . entrySet ( ) ) { List < Object > header = new ArrayList < > ( 1 ) ; Object resolvedValue = headerEval . eval ( headerVars , entry . getValue ( ) , String . class ) ; header . add ( resolvedValue ) ; requestHeaders . put ( entry . getKey ( ) , header ) ; } return requestHeaders ;


signature:
public class HttpClientCommon { /** * Determines the HTTP method to use for the next request . It may include an EL expression to evaluate .
 * @ param record Current record to set in context .
 * @ return the { @ link HttpMethod } to use for the request
 * @ throws ELEvalException if an expression is supplied that cannot be evaluated */
public HttpMethod getHttpMethod ( HttpMethod httpMethod , String methodExpression , Record record ) throws ELEvalException { } }

implementation: if ( httpMethod != HttpMethod . EXPRESSION ) { return httpMethod ; } RecordEL . setRecordInContext ( methodVars , record ) ; return HttpMethod . valueOf ( methodEval . eval ( methodVars , methodExpression , String . class ) ) ;


signature:
public class ScriptTypedNullObject { /** * Add each object of typed null to SimpleBindings , so that script languages
 * can use constants such as " NULL _ INTEGER " , " NULL _ LONG " without importing
 * other files .
 * @ param bindings : initialized in AbstractScriptingProcessor */
public static void fillNullTypes ( SimpleBindings bindings ) { } }

implementation: bindings . put ( "NULL_BOOLEAN" , NULL_BOOLEAN ) ; bindings . put ( "NULL_CHAR" , NULL_CHAR ) ; bindings . put ( "NULL_BYTE" , NULL_BYTE ) ; bindings . put ( "NULL_SHORT" , NULL_SHORT ) ; bindings . put ( "NULL_INTEGER" , NULL_INTEGER ) ; bindings . put ( "NULL_LONG" , NULL_LONG ) ; bindings . put ( "NULL_FLOAT" , NULL_FLOAT ) ; bindings . put ( "NULL_DOUBLE" , NULL_DOUBLE ) ; bindings . put ( "NULL_DATE" , NULL_DATE ) ; bindings . put ( "NULL_DATETIME" , NULL_DATETIME ) ; bindings . put ( "NULL_TIME" , NULL_TIME ) ; bindings . put ( "NULL_DECIMAL" , NULL_DECIMAL ) ; bindings . put ( "NULL_BYTE_ARRAY" , NULL_BYTE_ARRAY ) ; bindings . put ( "NULL_STRING" , NULL_STRING ) ; bindings . put ( "NULL_LIST" , NULL_LIST ) ; bindings . put ( "NULL_MAP" , NULL_MAP ) ;


signature:
public class ScriptTypedNullObject { /** * Receive record and fieldPath from scripting processor .
 * It resolves type of the field , and if value is null , it returns
 * one of the NULL _ XXX objects defined in this class .
 * If field value is not null , it returns the value stored in the field .
 * @ param record Record
 * @ param fieldPath field Path
 * @ return One of the NULL _ XXX objects if field has type but value is null . */
public static Object getFieldNull ( Record record , String fieldPath ) { } }

implementation: Field f = record . get ( fieldPath ) ; if ( f != null ) { return f . getValue ( ) == null ? getTypedNullFromField ( f ) : f . getValue ( ) ; } return null ;


signature:
public class ScriptTypedNullObject { /** * Receive a scriptOject and find out if the scriptObect is one of the NULL _ * *
 * object defined in this class . If so , create a new field with the type
 * and null value , then return the field .
 * If the scriptObject is not one of the typed null object , it returns a
 * new field with string converted from the value .
 * @ param scriptObject : ScriptObject , this might be one of the Typed Null object .
 * @ return */
public static Field getTypedNullFieldFromScript ( Object scriptObject ) { } }

implementation:
Field field ; if ( scriptObject == NULL_BOOLEAN ) field = Field . create ( Field . Type . BOOLEAN , null ) ; else if ( scriptObject == NULL_CHAR ) field = Field . create ( Field . Type . CHAR , null ) ; else if ( scriptObject == NULL_BYTE ) field = Field . create ( Field . Type . BYTE , null ) ; else if ( scriptObject == NULL_SHORT ) field = Field . create ( Field . Type . SHORT , null ) ; else if ( scriptObject == NULL_INTEGER ) field = Field . create ( Field . Type . INTEGER , null ) ; else if ( scriptObject == NULL_LONG ) field = Field . create ( Field . Type . LONG , null ) ; else if ( scriptObject == NULL_FLOAT ) field = Field . create ( Field . Type . FLOAT , null ) ; else if ( scriptObject == NULL_DOUBLE ) field = Field . create ( Field . Type . DOUBLE , null ) ; else if ( scriptObject == NULL_DATE ) field = Field . createDate ( null ) ; else if ( scriptObject == NULL_DATETIME ) field = Field . createDatetime ( null ) ; else if ( scriptObject == NULL_TIME ) field = Field . createTime ( null ) ; else if ( scriptObject == NULL_DECIMAL ) field = Field . create ( Field . Type . DECIMAL , null ) ; else if ( scriptObject == NULL_BYTE_ARRAY ) field = Field . create ( Field . Type . BYTE_ARRAY , null ) ; else if ( scriptObject == NULL_STRING ) field = Field . create ( Field . Type . STRING , null ) ; else if ( scriptObject == NULL_LIST ) field = Field . create ( Field . Type . LIST , null ) ; else if ( scriptObject == NULL_MAP ) field = Field . create ( Field . Type . MAP , null ) ; else // this scriptObject is not Null typed field . Return null .
field = null ; return field ;


signature:
public class DateTimeColumnHandler { /** * This method returns an { @ linkplain Field } that represents a DATE , TIME or TIMESTAMP . It is possible for user to upgrade
 * a field from DATE to TIMESTAMP , and if we read the table schema on startup after this upgrade , we would assume the field
 * should be returned as DATETIME field . But it is possible that the first change we read was made before the upgrade from
 * DATE to TIMESTAMP . So we check whether the returned SQL has TO _ TIMESTAMP - if it does we return it as DATETIME , else we
 * return it as DATE . */
public Field getDateTimeStampField ( String column , String columnValue , int columnType , String actualType ) throws StageException { } }

implementation:
Field . Type type ; if ( DATE . equalsIgnoreCase ( actualType ) ) { type = Field . Type . DATE ; } else if ( TIME . equalsIgnoreCase ( actualType ) ) { type = Field . Type . TIME ; } else if ( TIMESTAMP . equalsIgnoreCase ( actualType ) ) { type = Field . Type . DATETIME ; } else { throw new StageException ( JDBC_37 , columnType , column ) ; } if ( columnValue == null ) { return Field . create ( type , null ) ; } else { Optional < String > ts = matchDateTimeString ( toTimestampPattern . matcher ( columnValue ) ) ; if ( ts . isPresent ( ) ) { if ( timestampAsString ) { return Field . create ( Field . Type . STRING , ts . get ( ) ) ; } Timestamp timestamp = Timestamp . valueOf ( ts . get ( ) ) ; Field field = Field . create ( type , timestamp ) ; JdbcUtil . setNanosecondsinAttribute ( timestamp . getNanos ( ) , field ) ; return field ; } // We did not find TO _ TIMESTAMP , so try TO _ DATE
Optional < String > dt = matchDateTimeString ( toDatePattern . matcher ( columnValue ) ) ; return Field . create ( Field . Type . DATE , dt . map ( s -> Date . from ( getDate ( s ) . atZone ( zoneId ) . toInstant ( ) ) ) . orElse ( null ) ) ; }


signature:
public class AvroSchemaHelper { /** * Parses and returns an Avro schema loaded from the schema registry using the provided schema ID
 * if available , or the latest version of a schema for the specified subject .
 * @ param subject optional schema subject ( if schema ID is provided )
 * @ param schemaId optional schema ID ( if subject is provided )
 * @ return parsed avro schema
 * @ throws SchemaRegistryException if there is an error with the registry client */
public Schema loadFromRegistry ( String subject , int schemaId ) throws SchemaRegistryException { } }

implementation: try { if ( isEmpty ( subject ) ) { return loadFromRegistry ( schemaId ) ; } else { return loadFromRegistry ( subject ) ; } } catch ( SchemaRegistryException e ) { throw new SchemaRegistryException ( e ) ; }


signature:
public class AvroSchemaHelper { /** * Registers a parsed schema with the schema registry under the specified subject .
 * @ param schema parsed avro schema
 * @ param subject subject to register the schema under
 * @ return schemaId if registration was successful
 * @ throws SchemaRegistryException if there is an error with the registry client */
public int registerSchema ( Schema schema , String subject ) throws SchemaRegistryException { } }

implementation: try { return schemaIdCache . get ( subject + schema . hashCode ( ) , ( ) -> registryClient . register ( subject , schema ) ) ; } catch ( ExecutionException e ) { throw new SchemaRegistryException ( e ) ; }


signature:
public class AvroSchemaHelper { /** * Loads and parses a schema for the specified subject from the schema registry
 * @ param subject subject for which to fetch the latest version of a schema .
 * @ return parsed avro schema
 * @ throws SchemaRegistryException if there is an error with the registry client */
public Schema loadFromRegistry ( String subject ) throws SchemaRegistryException { } }

implementation: try { SchemaMetadata metadata = registryClient . getLatestSchemaMetadata ( subject ) ; return registryClient . getByID ( metadata . getId ( ) ) ; } catch ( IOException | RestClientException e ) { throw new SchemaRegistryException ( e ) ; }


signature:
public class AvroSchemaHelper { /** * Looks up schema id for the specified subject from the schema registry
 * @ param subject subject for which schema Id must be looked up .
 * @ return the schema id
 * @ throws SchemaRegistryException if there is an error with the registry client */
public int getSchemaIdFromSubject ( String subject ) throws SchemaRegistryException { } }

implementation: try { SchemaMetadata metadata = registryClient . getLatestSchemaMetadata ( subject ) ; return metadata . getId ( ) ; } catch ( IOException | RestClientException e ) { throw new SchemaRegistryException ( e ) ; }


signature:
public class AvroSchemaHelper { /** * Loads and parses a schema for the specified schema ID from the schema registry
 * @ param id schema ID to fetch from the registry
 * @ return parsed avro schema
 * @ throws SchemaRegistryException if there is an error with the registry client */
public Schema loadFromRegistry ( int id ) throws SchemaRegistryException { } }

implementation: try { return registryClient . getByID ( id ) ; } catch ( IOException | RestClientException e ) { throw new SchemaRegistryException ( e ) ; }


signature:
public class AvroSchemaHelper { /** * Writes the magic byte and schema ID to an output stream , replicating the functionality
 * of the Confluent Kafka Avro Serializer
 * @ param os OutputStream to write to
 * @ return schema ID that was written
 * @ throws IOException if there is an error */
public int writeSchemaId ( OutputStream os , int schemaId ) throws IOException { } }

implementation: if ( schemaId > 0 ) { os . write ( MAGIC_BYTE ) ; os . write ( ByteBuffer . allocate ( ID_SIZE ) . putInt ( schemaId ) . array ( ) ) ; } return schemaId ;


signature:
public class AvroSchemaHelper { /** * Checks for a magic byte in the data and if present extracts the schemaId
 * @ param data byte array representing a kafka message
 * @ return parsed schema ID */
public Optional < Integer > detectSchemaId ( byte [ ] data ) { } }

implementation:
if ( data . length < 5 ) { return Optional . empty ( ) ; } ByteBuffer wrapped = ByteBuffer . wrap ( data ) ; // 5 = = MAGIC _ BYTE + ID _ SIZE
if ( wrapped . get ( ) != MAGIC_BYTE ) { return Optional . empty ( ) ; } return Optional . of ( wrapped . getInt ( ) ) ;


signature:
public class AvroSchemaHelper { /** * Helper method to extract default values from a Schema . This is normally done
 * in DataGeneratorFormat validation , however we have to do it at runtime for
 * Schema Registry .
 * @ param schema schema to extract default values from
 * @ return map of default value
 * @ throws SchemaRegistryException */
public static Map < String , Object > getDefaultValues ( Schema schema ) throws SchemaRegistryException { } }

implementation: Map < String , Object > defaultValues = new HashMap < > ( ) ; try { defaultValues . putAll ( AvroTypeUtil . getDefaultValuesFromSchema ( schema , new HashSet < String > ( ) ) ) ; } catch ( IOException e ) { throw new SchemaRegistryException ( e ) ; } return defaultValues ;


signature:
public class ServicesUtil { /** * Convenience method to parse all available records in given payload . This method assumes that the stage already
 * depends on DataFormatParserService service and loads the service instance from Stage . Context .
 * @ param stageContext Stage ' s context
 * @ param toErrorContext Error context for any parser errors
 * @ param produceSingleRecordPerMessage If set , then output will be list of exactly one record
 * @ param messageId Message id base to be used with generated records
 * @ param payload Binary payload to parse
 * @ return
 * @ throws StageException */
public static List < Record > parseAll ( Stage . Context stageContext , ToErrorContext toErrorContext , boolean produceSingleRecordPerMessage , String messageId , byte [ ] payload ) throws StageException { } }

implementation:
List < Record > records = new ArrayList < > ( ) ; try ( DataParser parser = stageContext . getService ( DataFormatParserService . class ) . getParser ( messageId , payload ) ) { Record record = null ; do { try { record = parser . parse ( ) ; } catch ( RecoverableDataParserException e ) { handleException ( stageContext , toErrorContext , messageId , e , e . getUnparsedRecord ( ) ) ; // Go to next record
continue ; } if ( record != null ) { records . add ( record ) ; } } while ( record != null ) ; } catch ( IOException | DataParserException ex ) { Record record = stageContext . createRecord ( messageId ) ; record . set ( Field . create ( payload ) ) ; handleException ( stageContext , toErrorContext , messageId , ex , record ) ; return records ; } if ( produceSingleRecordPerMessage ) { List < Field > list = new ArrayList < > ( ) ; for ( Record record : records ) { list . add ( record . get ( ) ) ; } Record record = records . get ( 0 ) ; record . set ( Field . create ( list ) ) ; records . clear ( ) ; records . add ( record ) ; } return records ;


signature:
public class ClasspathValidatorResult { /** * Generate small report into log . */
public void logDetails ( ) { } }

implementation: if ( isValid ( ) ) { return ; } LOG . warn ( "Validation results for {}" , name ) ; if ( ! unparseablePaths . isEmpty ( ) ) { LOG . warn ( "Can't parse the following artifacts:" ) ; for ( String path : unparseablePaths ) { LOG . warn ( "  {}" , path ) ; } } if ( ! versionCollisions . isEmpty ( ) ) { LOG . warn ( "Detected colliding dependency versions:" ) ; for ( Map . Entry < String , Map < String , List < Dependency > > > entry : versionCollisions . entrySet ( ) ) { LOG . warn ( "  Dependency {} have versions: {}" , entry . getKey ( ) , StringUtils . join ( entry . getValue ( ) . keySet ( ) , ", " ) ) ; for ( Map . Entry < String , List < Dependency > > versionEntry : entry . getValue ( ) . entrySet ( ) ) { LOG . warn ( "    Version: {}" , versionEntry . getKey ( ) ) ; for ( Dependency dependency : versionEntry . getValue ( ) ) { LOG . warn ( "      {}" , dependency . getSourceName ( ) ) ; } } } }


signature:
public class FieldOrderProcessor { /** * Remove initial " / " , replace all other slashes with dots */
private String toFieldName ( String fieldPath ) { } }

implementation: String path = fieldPath . substring ( 1 ) . replaceAll ( "/" , "." ) ; path = "/" + path ; return EscapeUtil . getLastFieldNameFromPath ( path ) ;


signature:
public class HttpClientConfigBean { /** * Validates the parameters for this config bean .
 * @ param context Stage Context
 * @ param groupName Group name this bean is used in
 * @ param prefix Prefix to the parameter names ( e . g . parent beans )
 * @ param issues List of issues to augment */
public void init ( Source . Context context , String groupName , String prefix , List < Stage . ConfigIssue > issues ) { } }

implementation:
// Validate the ELs for string configs
List < String > elConfigs = ImmutableList . of ( "resourceUrl" , "requestBody" ) ; for ( String configName : elConfigs ) { ELVars vars = context . createELVars ( ) ; vars . addVariable ( START_AT , 0 ) ; ELEval eval = context . createELEval ( configName ) ; try { eval . eval ( vars , ( String ) getClass ( ) . getField ( configName ) . get ( this ) , String . class ) ; } catch ( ELEvalException | NoSuchFieldException | IllegalAccessException e ) { LOG . error ( Errors . HTTP_06 . getMessage ( ) , e . toString ( ) , e ) ; issues . add ( context . createConfigIssue ( groupName , prefix + configName , Errors . HTTP_06 , e . toString ( ) ) ) ; } } client . init ( context , Groups . PROXY . name ( ) , prefix + "client." , issues ) ; // Validate the EL for each header entry
ELVars headerVars = context . createELVars ( ) ; ELEval headerEval = context . createELEval ( "headers" ) ; for ( String headerValue : headers . values ( ) ) { try { headerEval . eval ( headerVars , headerValue , String . class ) ; } catch ( ELEvalException e ) { LOG . error ( Errors . HTTP_06 . getMessage ( ) , e . toString ( ) , e ) ; issues . add ( context . createConfigIssue ( groupName , prefix + "headers" , Errors . HTTP_06 , e . toString ( ) ) ) ; } }


signature:
public class HttpClientTarget { /** * Waits for the Jersey client to complete an asynchronous request , checks the response code
 * and continues to parse the response if it is deemed ok .
 * @ param record the current record to set in context for any expression evaluation
 * @ param responseFuture the async HTTP request future
 * @ param maxRequestCompletionSecs maximum time to wait for request completion ( start to finish )
 * @ throws StageException if the request fails , times out , or cannot be parsed */
private void processResponse ( Record record , Future < Response > responseFuture , long maxRequestCompletionSecs , boolean failOn403 ) throws StageException { } }

implementation: Response response ; try { response = responseFuture . get ( maxRequestCompletionSecs , TimeUnit . SECONDS ) ; String responseBody = "" ; if ( response . hasEntity ( ) ) { responseBody = response . readEntity ( String . class ) ; } response . close ( ) ; if ( conf . client . useOAuth2 && response . getStatus ( ) == 403 && ! failOn403 ) { HttpStageUtil . getNewOAuth2Token ( conf . client . oauth2 , httpClientCommon . getClient ( ) ) ; } else if ( response . getStatus ( ) < 200 || response . getStatus ( ) >= 300 ) { throw new OnRecordErrorException ( record , Errors . HTTP_40 , response . getStatus ( ) , response . getStatusInfo ( ) . getReasonPhrase ( ) + " " + responseBody ) ; } else { if ( conf . responseConf . sendResponseToOrigin ) { if ( ResponseType . SUCCESS_RECORDS . equals ( conf . responseConf . responseType ) ) { getContext ( ) . toSourceResponse ( record ) ; } else { getContext ( ) . toSourceResponse ( createResponseRecord ( responseBody ) ) ; } } } } catch ( InterruptedException | ExecutionException e ) { LOG . error ( Errors . HTTP_41 . getMessage ( ) , e . toString ( ) , e ) ; throw new OnRecordErrorException ( record , Errors . HTTP_41 , e . toString ( ) ) ; } catch ( TimeoutException e ) { LOG . error ( "HTTP request future timed out" , e . toString ( ) , e ) ; throw new OnRecordErrorException ( record , Errors . HTTP_41 , e . toString ( ) ) ; }


signature:
public class KuduRecordConverter { /** * delete record have / OldData */
private String binLogRecordFieldtoSdc ( String fieldPath , int operation ) { } }

implementation: if ( operation == KuduOperationType . DELETE . code ) { return "/OldData" + fieldPath ; } return fieldPath ;


signature:
public class BadRecordsHandler { /** * Generate list of error records from the error sink . What precise records will be returned depends on the error
 * record policy configuration .
 * @ param errorSink Error sink with the error records
 * @ return List with records that should be sent to error stream */
private List < Record > getBadRecords ( ErrorSink errorSink ) { } }

implementation: List < Record > badRecords = new ArrayList < > ( ) ; for ( Map . Entry < String , List < Record > > entry : errorSink . getErrorRecords ( ) . entrySet ( ) ) { for ( Record record : entry . getValue ( ) ) { RecordImpl errorRecord ; switch ( errorRecordPolicy ) { case ORIGINAL_RECORD : errorRecord = ( RecordImpl ) ( ( RecordImpl ) record ) . getHeader ( ) . getSourceRecord ( ) ; errorRecord . getHeader ( ) . copyErrorFrom ( record ) ; break ; case STAGE_RECORD : errorRecord = ( RecordImpl ) record ; break ; default : throw new IllegalArgumentException ( "Uknown error record policy: " + errorRecordPolicy ) ; } errorRecord . getHeader ( ) . setErrorContext ( runtimeInfo . getId ( ) , pipelineName ) ; badRecords . add ( errorRecord ) ; } } return badRecords ;


signature:
public class Version { /** * Actual comparison with the same semantics as compareTo ( ) method . */
public int compare ( Version other ) { } }

implementation: int maxParts = Math . max ( this . versions . length , other . versions . length ) ; for ( int i = 0 ; i < maxParts ; i ++ ) { int eq = this . getVersionPosition ( i ) - other . getVersionPosition ( i ) ; if ( eq != 0 ) { if ( eq > 0 ) { return 1 ; } else { return - 1 ; } } } return 0 ;


signature:
public class SchemaGenerator { /** * Initialize the schema generator
 * @ param config User configuration
 * @ param context Stage context
 * @ return Any validation issues this generator might have encounter */
public List < Stage . ConfigIssue > init ( SchemaGeneratorConfig config , Stage . Context context ) { } }

implementation: this . config = config ; return Collections . emptyList ( ) ;


signature:
public class DirectedGraph { /** * Returns the outward flowing edge vertices .
 * @ param vertex vertex
 * @ return the outward flowing edge vertices . */
public Collection < V > getOutwardEdgeVertices ( V vertex ) { } }

implementation: Collection < V > outwardEdgeVerticesForVertex = outwardEdgeVertices . get ( vertex ) ; return outwardEdgeVerticesForVertex != null ? outwardEdgeVerticesForVertex : Collections . < V > emptySet ( ) ;


signature:
public class DirectedGraph { /** * Returns the inward flowing edge vertices .
 * @ param vertex vertex
 * @ return the inward flowing edge vertices . */
public Collection < V > getInwardEdgeVertices ( V vertex ) { } }

implementation: Collection < V > inwardEdgeVerticesForVertex = inwardEdgesVertices . get ( vertex ) ; return inwardEdgeVerticesForVertex != null ? inwardEdgeVerticesForVertex : Collections . < V > emptySet ( ) ;


signature:
public class DirectedGraph { /** * Add a directed edge from vertex1 to vertex2 adding those vertices graph ( if needed )
 * @ param vertex1 vertex1
 * @ param vertex2 vertex2 */
public void addDirectedEdge ( V vertex1 , V vertex2 ) { } }

implementation: addVertex ( vertex1 ) ; addVertex ( vertex2 ) ; outwardEdgeVertices . put ( vertex1 , vertex2 ) ; inwardEdgesVertices . put ( vertex2 , vertex1 ) ;


signature:
public class OffsetUtil { /** * Serialize the Map of table to offset to a String
 * @ param offsetMap Map of table to Offset .
 * @ return Serialized offset
 * @ throws StageException When Serialization exception happens */
public static String serializeOffsetMap ( Map < String , String > offsetMap ) throws IOException { } }

implementation: return JSON_MAPPER . writeValueAsString ( offsetMap ) ;


signature:
public class OffsetUtil { /** * Deserialize String offset to Map of table to offset
 * @ param lastSourceOffset Serialized offset String
 * @ return Map of table to lastOffset
 * @ throws StageException When Deserialization exception happens */
@ SuppressWarnings ( "unchecked" ) public static Map < String , String > deserializeOffsetMap ( String lastSourceOffset ) throws IOException { } }

implementation: Map < String , String > offsetMap ; if ( lastSourceOffset == null || lastSourceOffset . isEmpty ( ) ) { offsetMap = new HashMap < > ( ) ; } else { offsetMap = JSON_MAPPER . readValue ( lastSourceOffset , Map . class ) ; } return offsetMap ;


signature:
public class OmniturePollingConsumer { /** * Queue a report using the Report . Queue method . This will post a request with the report description
 * to the Omniture API and return a report ID that can be used to retrieve the report once it ' s ready .
 * @ return report ID
 * @ throws IOException
 * @ throws InterruptedException
 * @ throws ExecutionException
 * @ throws TimeoutException */
public int queueReport ( ) throws IOException , InterruptedException , ExecutionException , TimeoutException , StageException { } }

implementation: final AsyncInvoker asyncInvoker = queueResource . request ( ) . header ( WSSE_HEADER , OmnitureAuthUtil . getHeader ( username . get ( ) , sharedSecret . get ( ) ) ) . async ( ) ; LOG . debug ( "Queueing report using URL {} with description {}" , queueResource . getUri ( ) . toURL ( ) . toString ( ) , reportDescription ) ; final Future < Response > responseFuture = asyncInvoker . post ( Entity . json ( reportDescription ) ) ; Response response = responseFuture . get ( responseTimeoutMillis , TimeUnit . MILLISECONDS ) ; if ( response == null ) { LOG . error ( "Failed to get response using URL {}" , queueResource . getUri ( ) . toURL ( ) . toString ( ) ) ; throw new StageException ( Errors . OMNITURE_01 , "HTTP response was null" ) ; } LOG . debug ( "Received response: status {}" , response . getStatus ( ) ) ; ObjectMapper mapper = new ObjectMapper ( ) ; String json = response . readEntity ( String . class ) ; LOG . trace ( "Response JSON: {}" , json ) ; JsonNode root = mapper . readTree ( json ) ; if ( root == null ) { LOG . error ( "Invalid JSON in response: {}" , json ) ; throw new StageException ( Errors . OMNITURE_01 , json ) ; } if ( root . has ( "error" ) ) { throw new StageException ( Errors . OMNITURE_01 , root . get ( "error_description" ) . asText ( ) ) ; } LOG . info ( "Omniture report queued" ) ; return root . get ( "reportID" ) . asInt ( ) ;


signature:
public class OmniturePollingConsumer { /** * Posts a request to the Omniture API to get a report back . Reports may take a while to generate ,
 * so this will loop on the Report . Get request , and ignore any errors indicating that the report
 * is not yet ready .
 * @ param reportId ID of report to get
 * @ throws InterruptedException
 * @ throws ExecutionException
 * @ throws TimeoutException
 * @ throws IOException */
public void getReport ( int reportId ) throws InterruptedException , ExecutionException , TimeoutException , IOException , StageException { } }

implementation:
int waitTime = 1000 ; Response response = null ; while ( ! stop ) { final AsyncInvoker asyncInvoker = getResource . request ( ) . header ( WSSE_HEADER , OmnitureAuthUtil . getHeader ( username . get ( ) , sharedSecret . get ( ) ) ) . async ( ) ; LOG . debug ( "Getting report using URL {} with report ID {}" , getResource . getUri ( ) . toURL ( ) . toString ( ) , reportId ) ; final Future < Response > responseFuture = asyncInvoker . post ( Entity . json ( "{ \"reportID\": " + reportId + " }" ) ) ; response = responseFuture . get ( responseTimeoutMillis , TimeUnit . MILLISECONDS ) ; String input = response . readEntity ( String . class ) ; ObjectMapper mapper = new ObjectMapper ( ) ; JsonNode root = mapper . readTree ( input ) ; // If the report has an error field , it means the report has not finished generating
if ( ! root . has ( "error" ) ) { boolean accepted = entityQueue . offer ( input , responseTimeoutMillis , TimeUnit . MILLISECONDS ) ; if ( ! accepted ) { LOG . warn ( "Response buffer full, dropped record." ) ; } break ; } else { // Exponential backoff while making subsequent Report . Get requests
if ( root . get ( "error" ) . textValue ( ) . equals ( "report_not_ready" ) ) { waitTime *= 2 ; LOG . info ( "Report not available. Sleeping for {} seconds" , waitTime / 1000 ) ; Thread . sleep ( waitTime ) ; } else { throw new StageException ( Errors . OMNITURE_02 , root . get ( "error" ) . get ( "error_description" ) . asText ( ) ) ; } } } response . close ( ) ;


signature:
public class SdcClusterOffsetHelper { /** * This tell us SDC is check pointing */
public boolean isSDCCheckPointing ( ) { } }

implementation: try { return fs . exists ( checkPointFilePath ) || fs . exists ( backupCheckPointFilePath ) ; } catch ( IOException ex ) { LOG . error ( "Error doing isSDCCheckPointing" , ex ) ; throw new RuntimeException ( Utils . format ( "Error checking exists on hdfs path: {}. Reason: {}" , checkPointFilePath . toString ( ) , ex . toString ( ) ) , ex ) ; }


signature:
public class SdcClusterOffsetHelper { /** * Or if the file is corrupted , we want to update the right offsets to the main offset file . */
private void writeOffsetsToMainOffsetFile ( Map < Integer , Long > partitionToOffsetMap ) throws IOException { } }

implementation:
LOG . info ( "Saving the following offset {} to {}" , partitionToOffsetMap , checkPointFilePath ) ; // Creating a marker file ( overwriting if it already exists ) to mark that we are going to write offsets out the offsets to the main offset file .
try ( OutputStream os = fs . create ( checkPointMarkerFilePath , true ) ) { // NOOP
} // If the both above passes and writing fails or leaves corrupted file we will have the back file
try ( OutputStream os = fs . create ( checkPointFilePath , true ) ) { OBJECT_MAPPER . writeValue ( os , new ClusterSourceOffsetJson ( serializeKafkaPartitionOffset ( partitionToOffsetMap ) , SDC_STREAMING_OFFSET_VERSION ) ) ; } // If this fails we are still good , as we will start from the backup offset file . ( Not optimal , but deterministic )
boolean deleted = fs . delete ( checkPointMarkerFilePath , false ) ; LOG . warn ( "Status {} for Deleting Marker File {}" , deleted , checkPointMarkerFilePath ) ; // If the write fails we don ' t want to touch the timestamp and will error out so not doing this in finally
lastOffsetStoredTime = System . currentTimeMillis ( ) ;


signature:
public class SdcClusterOffsetHelper { /** * Ex : Partition Offset empty , empty map in the offset , cannot deserialize */
private Map < Integer , Long > readClusterOffsetFile ( Path checkPointFilePath , int numberOfPartitions ) throws IOException { } }

implementation: if ( ! fs . exists ( checkPointFilePath ) ) { throw new IOException ( Utils . format ( "Checkpoint file path {} does not exist" , checkPointFilePath ) ) ; } ClusterSourceOffsetJson clusterSourceOffsetJson = OBJECT_MAPPER . readValue ( ( InputStream ) fs . open ( checkPointFilePath ) , ClusterSourceOffsetJson . class ) ; String lastSourceOffset = clusterSourceOffsetJson . getOffset ( ) ; if ( ! StringUtils . isEmpty ( lastSourceOffset ) ) { return deserializeKafkaPartitionOffset ( lastSourceOffset , numberOfPartitions ) ; } else { throw new IOException ( "Partition Offset Cannot be empty" ) ; }


signature:
public class TableOrderProviderFactory { /** * Create { @ link TableOrderProvider } for the { @ link TableOrderStrategy }
 * @ return { @ link TableOrderProvider } */
public TableOrderProvider create ( ) { } }

implementation:
switch ( tableOrderStrategy ) { case NONE : // Don ' t do any ordering , just add tables as per the incoming order .
return new DefaultTableOrderProvider ( new LinkedHashSet < > ( ) ) ; case ALPHABETICAL : // Alphabetical sorting based on table qualified names .
return new DefaultTableOrderProvider ( new TreeSet < > ( ) ) ; case REFERENTIAL_CONSTRAINTS : return new ReferentialTblOrderProvider ( connection ) ; default : throw new IllegalArgumentException ( Utils . format ( "Unknown table order strategy: {}" , tableOrderStrategy ) ) ; }


signature:
public class GcsObjectPostProcessingHandler { /** * Delete a blob for gcs
 * @ param blobId blobId */
private void delete ( BlobId blobId ) { } }

implementation: LOG . debug ( "Deleting object '{}'" , String . format ( BLOB_PATH_TEMPLATE , blobId . getBucket ( ) , blobId . getName ( ) ) ) ; boolean deleted = storage . delete ( blobId ) ; if ( ! deleted ) { LOG . error ( "Cannot delete object '{}'" , String . format ( BLOB_PATH_TEMPLATE , blobId . getBucket ( ) , blobId . getName ( ) ) ) ; }


signature:
public class GcsObjectPostProcessingHandler { /** * Copy a blob for gcs to a destination bucket and a path ( and delete the source blob if needed )
 * @ param sourceBlobId sourceBlobId
 * @ param destinationBucket destination bucket
 * @ param destinationPath destination path
 * @ param deleteSource delete source blob */
private void copy ( BlobId sourceBlobId , String destinationBucket , String destinationPath , boolean deleteSource ) { } }

implementation: LOG . debug ( "Copying object '{}' to Object '{}'" , String . format ( BLOB_PATH_TEMPLATE , sourceBlobId . getBucket ( ) , sourceBlobId . getName ( ) ) , String . format ( BLOB_PATH_TEMPLATE , destinationBucket , destinationPath ) ) ; Storage . CopyRequest copyRequest = new Storage . CopyRequest . Builder ( ) . setSource ( sourceBlobId ) . setTarget ( BlobId . of ( destinationBucket , destinationPath ) ) . build ( ) ; Blob destinationBlob = storage . copy ( copyRequest ) . getResult ( ) ; LOG . debug ( "Copied object '{}' to Object '{}'" , String . format ( BLOB_PATH_TEMPLATE , sourceBlobId . getBucket ( ) , sourceBlobId . getName ( ) ) , String . format ( BLOB_PATH_TEMPLATE , destinationBlob . getBlobId ( ) . getBucket ( ) , destinationBlob . getBlobId ( ) . getName ( ) ) ) ; if ( deleteSource ) { delete ( sourceBlobId ) ; }


signature:
public class GcsObjectPostProcessingHandler { /** * Handle error Blob
 * @ param blobId blob id */
void handleError ( BlobId blobId ) { } }

implementation: switch ( gcsOriginErrorConfig . errorHandlingOption ) { case NONE : break ; case ARCHIVE : handleArchive ( blobId ) ; break ; case DELETE : delete ( blobId ) ; break ; }


signature:
public class GcsObjectPostProcessingHandler { /** * Archive the blob
 * @ param blobId blobId */
private void handleArchive ( BlobId blobId ) { } }

implementation: String destinationPath = getDestinationPath ( blobId , gcsOriginErrorConfig . errorPrefix ) ; switch ( gcsOriginErrorConfig . archivingOption ) { case COPY_TO_BUCKET : copy ( blobId , gcsOriginErrorConfig . errorBucket , destinationPath , false ) ; break ; case MOVE_TO_BUCKET : copy ( blobId , gcsOriginErrorConfig . errorBucket , destinationPath , true ) ; break ; case COPY_TO_PREFIX : copy ( blobId , blobId . getBucket ( ) , destinationPath , false ) ; break ; case MOVE_TO_PREFIX : copy ( blobId , blobId . getBucket ( ) , destinationPath , true ) ; break ; }


signature:
public class FileContext { /** * prepares and gets the reader if available before a read . */
public LiveFileReader getReader ( ) throws IOException { } }

implementation:
Utils . checkState ( open , "FileContext is closed" ) ; if ( reader == null ) { currentFile = getStartingCurrentFileName ( ) ; long fileOffset = getStartingOffset ( ) ; boolean needsToScan = currentFile == null || fileOffset == Long . MAX_VALUE ; if ( needsToScan ) { if ( currentFile != null ) { // we need to refresh the file in case the name changed before scanning as the scanner does not refresh
currentFile = currentFile . refresh ( ) ; } currentFile = scanner . scan ( currentFile ) ; fileOffset = 0 ; } if ( currentFile != null ) { reader = new SingleLineLiveFileReader ( getRollMode ( ) , getMultiFileInfo ( ) . getTag ( ) , currentFile , charset , fileOffset , maxLineLength ) ; if ( ! multiFileInfo . getMultiLineMainLinePatter ( ) . isEmpty ( ) ) { reader = new MultiLineLiveFileReader ( getMultiFileInfo ( ) . getTag ( ) , reader , Pattern . compile ( multiFileInfo . getMultiLineMainLinePatter ( ) ) ) ; } if ( fileOffset == 0 ) { // file start event
eventPublisher . publish ( new FileEvent ( currentFile , FileEvent . Action . START ) ) ; } } } return reader ;


signature:
public class FileContext { /** * updates reader and offsets after a read . */
public void releaseReader ( boolean inErrorDiscardReader ) throws IOException { } }

implementation:
Utils . checkState ( open , "FileContext is closed" ) ; // update starting offsets for next invocation either cold ( no reader ) or hot ( reader )
boolean hasNext ; try { hasNext = reader != null && reader . hasNext ( ) ; } catch ( IOException ex ) { IOUtils . closeQuietly ( reader ) ; reader = null ; hasNext = false ; } boolean doneWithFile = ! hasNext || inErrorDiscardReader ; if ( doneWithFile ) { IOUtils . closeQuietly ( reader ) ; reader = null ; // Using Long . MAX _ VALUE to signal we reach the end of the file and next iteration should get the next file .
setStartingCurrentFileName ( currentFile ) ; setStartingOffset ( Long . MAX_VALUE ) ; // If we failed to open the file in first place , it will be null and hence we won ' t do anything with it .
if ( currentFile == null ) { return ; } // File end event
LiveFile file = currentFile . refresh ( ) ; if ( inErrorDiscardReader ) { LOG . warn ( "Processing file '{}' produced an error, skipping '{}' post processing on that file" , file , postProcessing ) ; eventPublisher . publish ( new FileEvent ( file , FileEvent . Action . ERROR ) ) ; } else { eventPublisher . publish ( new FileEvent ( file , FileEvent . Action . END ) ) ; switch ( postProcessing ) { case NONE : LOG . debug ( "File '{}' processing completed, post processing action 'NONE'" , file ) ; break ; case DELETE : if ( ! inPreviewMode ) { try { Files . delete ( file . getPath ( ) ) ; LOG . debug ( "File '{}' processing completed, post processing action 'DELETED'" , file ) ; } catch ( IOException ex ) { throw new IOException ( Utils . format ( "Could not delete '{}': {}" , file , ex . toString ( ) ) , ex ) ; } } break ; case ARCHIVE : if ( ! inPreviewMode ) { Path fileArchive = Paths . get ( archiveDir , file . getPath ( ) . toString ( ) ) ; if ( fileArchive == null ) { throw new IOException ( "Could not find archive file" ) ; } try { Files . createDirectories ( fileArchive . getParent ( ) ) ; Files . move ( file . getPath ( ) , fileArchive ) ; LOG . debug ( "File '{}' processing completed, post processing action 'ARCHIVED' as" , file ) ; } catch ( IOException ex ) { throw new IOException ( Utils . format ( "Could not archive '{}': {}" , file , ex . toString ( ) ) , ex ) ; } } break ; } } } else { setStartingCurrentFileName ( currentFile ) ; setStartingOffset ( getReader ( ) . getOffset ( ) ) ; }


signature:
public class ApiClient { /** * Helper method to set dpmBaseURL for the first HTTP DPM authentication . */
public ApiClient setDPMBaseURL ( String dpmBaseURL ) { } }

implementation: if ( dpmBaseURL != null && authentication != null ) { authentication . setDPMBaseURL ( dpmBaseURL ) ; } return this ;


signature:
public class ApiClient { /** * Add a default header .
 * @ param key The header ' s key
 * @ param value The header ' s value */
public ApiClient addDefaultHeader ( String key , String value ) { } }

implementation: defaultHeaderMap . put ( key , value ) ; return this ;


signature:
public class ApiClient { /** * Parse the given string into Date object . */
public Date parseDate ( String str ) { } }

implementation: try { return dateFormat . parse ( str ) ; } catch ( java . text . ParseException e ) { throw new RuntimeException ( e ) ; }


signature:
public class ApiClient { /** * Select the Accept header ' s value from the given accepts array :
 * if JSON exists in the given array , use it ;
 * otherwise use all of them ( joining into a string )
 * @ param accepts The accepts array to select from
 * @ return The Accept header to use . If the given array is empty ,
 * null will be returned ( not to set the Accept header explicitly ) . */
public String selectHeaderAccept ( String [ ] accepts ) { } }

implementation: if ( accepts . length == 0 ) return null ; if ( StringUtil . containsIgnoreCase ( accepts , "application/json" ) ) return "application/json" ; return StringUtil . join ( accepts , "," ) ;


signature:
public class ApiClient { /** * Select the Content - Type header ' s value from the given array :
 * if JSON exists in the given array , use it ;
 * otherwise use the first one of the array .
 * @ param contentTypes The Content - Type array to select from
 * @ return The Content - Type header to use . If the given array is empty ,
 * JSON will be used . */
public String selectHeaderContentType ( String [ ] contentTypes ) { } }

implementation: if ( contentTypes . length == 0 ) return "application/json" ; if ( StringUtil . containsIgnoreCase ( contentTypes , "application/json" ) ) return "application/json" ; return contentTypes [ 0 ] ;


signature:
public class ApiClient { /** * Escape the given string to be used as URL query value . */
public String escapeString ( String str ) { } }

implementation: try { return URLEncoder . encode ( str , "utf8" ) . replaceAll ( "\\+" , "%20" ) ; } catch ( UnsupportedEncodingException e ) { return str ; }


signature:
public class ApiClient { /** * Serialize the given Java object into string according the given
 * Content - Type ( only JSON is supported for now ) . */
public String serialize ( Object obj , String contentType ) throws ApiException { } }

implementation: if ( contentType . startsWith ( "application/json" ) ) { return json . serialize ( obj ) ; } else { throw new ApiException ( 400 , "can not serialize object into Content-Type: " + contentType ) ; }


signature:
public class ApiClient { /** * Deserialize response body to Java object according to the Content - Type . */
private < T > T deserialize ( Response response , TypeRef returnType ) throws ApiException { } }

implementation: String contentType = null ; List < Object > contentTypes = response . getHeaders ( ) . get ( "Content-Type" ) ; if ( contentTypes != null && ! contentTypes . isEmpty ( ) ) { contentType = ( String ) contentTypes . get ( 0 ) ; } if ( contentType == null ) { throw new ApiException ( 500 , "missing Content-Type in response" ) ; } if ( contentType . startsWith ( "application/json" ) ) { String body ; if ( response . hasEntity ( ) ) { body = response . readEntity ( String . class ) ; } else { body = "" ; } if ( body . length ( ) > 0 ) { return json . deserialize ( body , returnType ) ; } return null ; } if ( contentType . startsWith ( "image" ) ) { return ( T ) response . readEntity ( InputStream . class ) ; } else { throw new ApiException ( 500 , "can not deserialize Content-Type: " + contentType ) ; }


signature:
public class ApiClient { /** * Get an existing client or create a new client to handle HTTP request . */
private Client getClient ( ) { } }

implementation: if ( ! hostMap . containsKey ( basePath ) ) { ClientConfig config = new ClientConfig ( ) ; config . property ( ClientProperties . SUPPRESS_HTTP_COMPLIANCE_VALIDATION , true ) ; Client client = ClientBuilder . newClient ( config ) ; client . register ( new CsrfProtectionFilter ( "CSRF" ) ) ; hostMap . put ( basePath , client ) ; } return hostMap . get ( basePath ) ;


signature:
public class KineticaHelper { /** * Connect to the database */
private GPUdb initConnection ( KineticaConfigBean conf ) throws GPUdbException , StageException { } }

implementation: KineticaConnectionUtils kineticaConnectionUtils = new KineticaConnectionUtils ( ) ; return kineticaConnectionUtils . getGPUdb ( conf ) ;


signature:
public class KineticaHelper { /** * Get metadata for the table */
private void getTableMetadata ( GPUdb gpudb , String tableName ) throws GPUdbException { } }

implementation: KineticaTableUtils kineticaTableUtils = new KineticaTableUtils ( gpudb , tableName ) ; type = kineticaTableUtils . getType ( ) ;


signature:
public class KineticaHelper { /** * Create a BulkInserter */
private BulkInserter < IndexedRecord > createBulkInserter ( GPUdb gpudb , Type type , KineticaConfigBean conf ) throws GPUdbException { } }

implementation: KineticaBulkInserterUtils kineticaBulkInserterUtils = new KineticaBulkInserterUtils ( gpudb , type , conf ) ; return kineticaBulkInserterUtils . createBulkInserter ( ) ;


signature:
public class EventCreator { /** * Create new event record according for this stage context and event context . */
public EventBuilder create ( Stage . Context context , ToEventContext toEvent ) { } }

implementation: return new EventBuilder ( context , toEvent ) ;


signature:
public class AvroSchemaGenerator { /** * Build a schema with type " record " . This will be the top level schema and contains fields */
public static Schema buildSchema ( Map < String , Schema > fields , Object ... levels ) { } }

implementation:
List < Schema . Field > recordFields = new ArrayList < > ( fields . size ( ) ) ; for ( Map . Entry < String , Schema > entry : fields . entrySet ( ) ) { recordFields . add ( new Schema . Field ( entry . getKey ( ) , entry . getValue ( ) , null , // Avro ' s Schema . Field constructor requires doc .
entry . getValue ( ) . getJsonProp ( "default" ) ) ) ; } Schema recordSchema ; if ( levels . length == 0 ) { recordSchema = Schema . createRecord ( schemaName , null , null , false ) ; } else { LinkedList < String > lvl = ( LinkedList < String > ) levels [ 0 ] ; recordSchema = Schema . createRecord ( joiner . join ( lvl ) , null , null , false ) ; } recordSchema . setFields ( recordFields ) ; return recordSchema ;


signature:
public class OracleCDCOperationCode { /** * This is called when JDBC target didn ' t find sdc . operation . code in record header
 * but found oracle . cdc . operation . Since oracle . cdc . operation contains Oracle specific
 * operation code , we need to convert to SDC operation code .
 * @ param code Operation code defined by Oracle .
 * @ return Operation code defined by SDC . */
public static int convertFromOracleToSDCCode ( String code ) { } }

implementation:
try { int intCode = Integer . parseInt ( code ) ; switch ( intCode ) { case INSERT_CODE : return OperationType . INSERT_CODE ; case DELETE_CODE : return OperationType . DELETE_CODE ; case UPDATE_CODE : case SELECT_FOR_UPDATE_CODE : return OperationType . UPDATE_CODE ; default : // DDL _ CODE
throw new UnsupportedOperationException ( Utils . format ( "Operation code {} is not supported" , code ) ) ; } } catch ( NumberFormatException ex ) { throw new NumberFormatException ( "Operation code must be a numeric value. " + ex . getMessage ( ) ) ; }


signature:
public class MapreduceUtils { /** * Add jars containing the following classes to the job ' s classpath . */
public static void addJarsToJob ( Configuration conf , Class ... klasses ) { } }

implementation:
// Build set of jars that needs to be added , order doesn ' t matter for us and we will remove duplicates
Set < String > additinonalJars = new HashSet < > ( ) ; for ( Class klass : klasses ) { final String jar = jarForClass ( klass ) ; LOG . info ( "Adding jar {} for class {}" , jar , klass . getCanonicalName ( ) ) ; additinonalJars . add ( jar ) ; } appendJars ( conf , additinonalJars ) ;


signature:
public class MapreduceUtils { /** * Add jars whose names contain the given patterns to the job ' s classpath .
 * Searches the { @ link ClassLoader } - which must be an instance of { @ link URLClassLoader } or an
 * { @ link IllegalStateException } will be thrown - of the { @ link MapreduceUtils } class itself for the jar matching
 * a pattern indicated each arguments .
 * Each supplied jarPattern is treated as a case - insensitive substring for a full jar path . This method expects
 * to find at least one jar matching each substring , or an { @ link IllegalArgumentException } will be thrown . If more
 * than one jar is matched , then the result will depend on the allowMultiple parameter - if true , then all will be
 * added , while if false , an { @ link IllegalArgumentException } will be thrown .
 * Care has < b > NOT < / b > been taken to optimize the performance of this method ; the runtime is M * N where M is the
 * number of entries on the classpath and N is the number of given jarPatterns . It is assumed that this method
 * will be called infrequently and with relatively few jarPatterns .
 * @ param conf the Hadoop conf for the MapReduce job
 * @ param allowMultiple whether multiple matching jars are allowed to be added for a single pattern
 * @ param jarPatterns the patterns to search for within the classpath */
public static void addJarsToJob ( Configuration conf , boolean allowMultiple , String ... jarPatterns ) { } }

implementation: final ClassLoader loader = MapreduceUtils . class . getClassLoader ( ) ; if ( ! ( loader instanceof URLClassLoader ) ) { throw new IllegalStateException ( String . format ( "ClassLoader for %s is not an instance of URLClassLoader (it is %s), and thus this method cannot be used" , MapreduceUtils . class . getCanonicalName ( ) , loader . getClass ( ) . getCanonicalName ( ) ) ) ; } final URLClassLoader urlClassLoader = ( URLClassLoader ) loader ; addJarsToJob ( conf , allowMultiple , urlClassLoader . getURLs ( ) , jarPatterns ) ;


signature:
public class StreamSetsRecordProcessor { /** * { @ inheritDoc } */
@ Override public void initialize ( InitializationInput initializationInput ) { } }

implementation: shardId = initializationInput . getShardId ( ) ; if ( LOG . isDebugEnabled ( ) ) { LOG . debug ( "Initializing record processor at: {}" , initializationInput . getExtendedSequenceNumber ( ) . toString ( ) ) ; LOG . debug ( "Initializing record processor for shard: {}" , shardId ) ; }


signature:
public class StreamSetsRecordProcessor { /** * { @ inheritDoc } */
@ Override public void processRecords ( ProcessRecordsInput processRecordsInput ) { } }

implementation:
LOG . debug ( "RecordProcessor processRecords called" ) ; try { IRecordProcessorCheckpointer checkpointer = processRecordsInput . getCheckpointer ( ) ; startBatch ( ) ; Optional < Record > lastProcessedRecord = Optional . empty ( ) ; int recordCount = 0 ; for ( Record kRecord : processRecordsInput . getRecords ( ) ) { try { KinesisUtil . processKinesisRecord ( shardId , kRecord , parserFactory ) . forEach ( batchMaker :: addRecord ) ; lastProcessedRecord = Optional . of ( kRecord ) ; if ( ++ recordCount == maxBatchSize ) { recordCount = 0 ; finishBatch ( checkpointer , kRecord ) ; startBatch ( ) ; } } catch ( DataParserException | IOException e ) { com . streamsets . pipeline . api . Record record = context . createRecord ( kRecord . getSequenceNumber ( ) ) ; record . set ( Field . create ( kRecord . getData ( ) . array ( ) ) ) ; try { errorRecordHandler . onError ( new OnRecordErrorException ( record , Errors . KINESIS_03 , kRecord . getSequenceNumber ( ) , e . toString ( ) , e ) ) ; // move the lastProcessedRecord forward if not set to stop pipeline
lastProcessedRecord = Optional . of ( kRecord ) ; } catch ( StageException ex ) { // KCL skips over the data records that were passed prior to the exception
// that is , these records are not re - sent to this record processor
// or to any other record processor in the consumer .
lastProcessedRecord . ifPresent ( r -> finishBatch ( checkpointer , r ) ) ; try { error . put ( ex ) ; } catch ( InterruptedException ie ) { Thread . currentThread ( ) . interrupt ( ) ; } return ; } } } lastProcessedRecord . ifPresent ( r -> finishBatch ( checkpointer , r ) ) ; } catch ( Exception e ) { LOG . error ( "Unknown error while processing records: {}" , e . toString ( ) , e ) ; context . reportError ( Errors . KINESIS_17 , e . toString ( ) ) ; }


signature:
public class StreamSetsRecordProcessor { /** * We don ' t checkpoint on SHUTDOWN _ REQUESTED because we currently always
 * checkpoint each batch in { @ link # processRecords } .
 * @ param shutdownInput { @ inheritDoc } */
@ Override public void shutdown ( ShutdownInput shutdownInput ) { } }

implementation:
LOG . info ( "Shutting down record processor for shard: {}" , shardId ) ; if ( ShutdownReason . TERMINATE . equals ( shutdownInput . getShutdownReason ( ) ) ) { // Shard is closed / finished processing . Checkpoint all processing up to here .
try { shutdownInput . getCheckpointer ( ) . checkpoint ( ) ; LOG . debug ( "Checkpointed due to record processor shutdown request." ) ; } catch ( InvalidStateException | ShutdownException e ) { LOG . error ( "Error checkpointing batch: {}" , e . toString ( ) , e ) ; } }


signature:
public class AwsRequestSigningApacheInterceptor { /** * { @ inheritDoc } */
@ Override public void process ( final HttpRequest request , final HttpContext context ) throws HttpException , IOException { } }

implementation:
URIBuilder uriBuilder ; try { uriBuilder = new URIBuilder ( request . getRequestLine ( ) . getUri ( ) ) ; } catch ( URISyntaxException e ) { throw new IOException ( "Invalid URI" , e ) ; } // Copy Apache HttpRequest to AWS DefaultRequest
DefaultRequest < ? > signableRequest = new DefaultRequest < > ( service ) ; HttpHost host = ( HttpHost ) context . getAttribute ( HttpCoreContext . HTTP_TARGET_HOST ) ; if ( host != null ) { signableRequest . setEndpoint ( URI . create ( host . toURI ( ) ) ) ; } final HttpMethodName httpMethod = HttpMethodName . fromValue ( request . getRequestLine ( ) . getMethod ( ) ) ; signableRequest . setHttpMethod ( httpMethod ) ; try { signableRequest . setResourcePath ( uriBuilder . build ( ) . getRawPath ( ) ) ; } catch ( URISyntaxException e ) { throw new IOException ( "Invalid URI" , e ) ; } if ( request instanceof HttpEntityEnclosingRequest ) { HttpEntityEnclosingRequest httpEntityEnclosingRequest = ( HttpEntityEnclosingRequest ) request ; if ( httpEntityEnclosingRequest . getEntity ( ) != null ) { signableRequest . setContent ( httpEntityEnclosingRequest . getEntity ( ) . getContent ( ) ) ; } } signableRequest . setParameters ( nvpToMapParams ( uriBuilder . getQueryParams ( ) ) ) ; signableRequest . setHeaders ( headerArrayToMap ( request . getAllHeaders ( ) ) ) ; // Sign it
signer . sign ( signableRequest , awsCredentialsProvider . getCredentials ( ) ) ; // Now copy everything back
request . setHeaders ( mapToHeaderArray ( signableRequest . getHeaders ( ) ) ) ; if ( request instanceof HttpEntityEnclosingRequest ) { HttpEntityEnclosingRequest httpEntityEnclosingRequest = ( HttpEntityEnclosingRequest ) request ; if ( httpEntityEnclosingRequest . getEntity ( ) != null ) { BasicHttpEntity basicHttpEntity = new BasicHttpEntity ( ) ; basicHttpEntity . setContent ( signableRequest . getContent ( ) ) ; httpEntityEnclosingRequest . setEntity ( basicHttpEntity ) ; } }


signature:
public class ForceSource { /** * Returns true if the first ORDER BY field matches fieldName */
private boolean checkFieldOrderByList ( SOQLParser . FieldOrderByListContext fieldOrderByList , String fieldName ) { } }

implementation: return fieldOrderByList . fieldOrderByElement ( 0 ) . fieldElement ( ) . getText ( ) . equalsIgnoreCase ( fieldName ) ;


signature:
public class ForceSource { /** * Returns true if any of the nested conditions contains fieldName */
private boolean checkConditionExpressions ( SOQLParser . ConditionExpressionsContext conditionExpressions , String fieldName ) { } }

implementation: for ( SOQLParser . ConditionExpressionContext ce : conditionExpressions . conditionExpression ( ) ) { if ( ( ce . conditionExpressions ( ) != null && checkConditionExpressions ( ce . conditionExpressions ( ) , fieldName ) ) || ( ce . fieldExpression ( ) != null && ce . fieldExpression ( ) . fieldElement ( ) . getText ( ) . equalsIgnoreCase ( fieldName ) ) ) { return true ; } } return false ;


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

implementation:
// SDC - 6258 - destroy ( ) is called from a different thread , so we
// need to signal produce ( ) to terminate early
destroyed . set ( true ) ; if ( job != null ) { try { try { bulkConnection . abortJob ( job . getId ( ) ) ; } catch ( AsyncApiException e ) { ForceUtils . renewSession ( bulkConnection , e ) ; bulkConnection . abortJob ( job . getId ( ) ) ; } job = null ; } catch ( AsyncApiException e ) { LOG . error ( "Exception while aborting job" , e ) ; } } job = null ; if ( forceConsumer != null ) { try { forceConsumer . stop ( ) ; forceConsumer = null ; } catch ( Exception e ) { LOG . error ( "Exception while stopping ForceStreamConsumer." , e ) ; } if ( ! messageQueue . isEmpty ( ) ) { LOG . error ( "Queue still had {} entities at shutdown." , messageQueue . size ( ) ) ; } else { LOG . info ( "Queue was empty at shutdown. No data lost." ) ; } } if ( recordCreator != null ) { recordCreator . destroy ( ) ; } // Clean up any open resources .
super . destroy ( ) ;


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

implementation:
String nextSourceOffset = null ; LOG . debug ( "lastSourceOffset: {}" , lastSourceOffset ) ; // send event only once for each time we run out of data .
if ( shouldSendNoMoreDataEvent ) { CommonEvents . NO_MORE_DATA . create ( getContext ( ) ) . createAndSend ( ) ; shouldSendNoMoreDataEvent = false ; return lastSourceOffset ; } int batchSize = Math . min ( conf . basicConfig . maxBatchSize , maxBatchSize ) ; if ( ! conf . queryExistingData || ( null != lastSourceOffset && lastSourceOffset . startsWith ( EVENT_ID_OFFSET_PREFIX ) ) ) { if ( conf . subscribeToStreaming ) { nextSourceOffset = streamingProduce ( lastSourceOffset , batchSize , batchMaker ) ; } else { // We ' re done reading existing data , but we don ' t want to subscribe to Streaming API
return null ; } } else if ( conf . queryExistingData ) { if ( ! queryInProgress ( ) ) { long now = System . currentTimeMillis ( ) ; long delay = Math . max ( 0 , ( lastQueryCompletedTime + ( 1000 * conf . queryInterval ) ) - now ) ; if ( delay > 0 ) { // Sleep in one second increments so we don ' t tie up the app .
LOG . info ( "{}ms remaining until next fetch." , delay ) ; ThreadUtil . sleep ( Math . min ( delay , 1000 ) ) ; return lastSourceOffset ; } } if ( conf . useBulkAPI ) { nextSourceOffset = bulkProduce ( lastSourceOffset , batchSize , batchMaker ) ; } else { nextSourceOffset = soapProduce ( lastSourceOffset , batchSize , batchMaker ) ; } } else if ( conf . subscribeToStreaming ) { // No offset , and we ' re not querying existing data , so switch to streaming
nextSourceOffset = READ_EVENTS_FROM_NOW ; } LOG . debug ( "nextSourceOffset: {}" , nextSourceOffset ) ; return nextSourceOffset ;


signature:
public class FileContextProviderUtil { /** * If passed a valid fileOffsetString , it will return what is the offset lag in the file .
 * @ param fileOffsetString
 * @ return offset lag for the live file .
 * @ throws IOException */
public static long getOffsetLagForFile ( String fileOffsetString ) throws IOException { } }

implementation:
long offset = FileContextProviderUtil . getLongOffsetFromFileOffset ( fileOffsetString ) ; // We are refreshing the live file here because we are going to get the size by using path .
LiveFile file = FileContextProviderUtil . getRefreshedLiveFileFromFileOffset ( fileOffsetString ) ; long fileSizeInBytes = Files . size ( file . getPath ( ) . toAbsolutePath ( ) ) ; return ( fileSizeInBytes - offset ) ;


signature:
public class BootstrapMain { /** * Visible due to JVM requirements only */
public static void premain ( String args , Instrumentation instrumentation ) { } }

implementation: if ( BootstrapMain . instrumentation == null ) { BootstrapMain . instrumentation = instrumentation ; } else { throw new IllegalStateException ( "Premain method cannot be called twice (" + BootstrapMain . instrumentation + ")" ) ; }


signature:
public class BootstrapMain { /** * if whitelist is ' * ' set is NULL , else whitelist has the whitelisted values */
public static Set < String > getWhiteList ( String configDir , String property ) { } }

implementation: Set < String > set = null ; File whiteListFile = new File ( configDir , WHITE_LIST_FILE ) . getAbsoluteFile ( ) ; if ( whiteListFile . exists ( ) ) { try ( InputStream is = new FileInputStream ( whiteListFile ) ) { Properties props = new Properties ( ) ; props . load ( is ) ; String whiteList = props . getProperty ( property ) ; if ( whiteList == null ) { throw new IllegalArgumentException ( String . format ( WHITE_LIST_PROPERTY_MISSING_MSG , property , whiteListFile ) ) ; } whiteList = whiteList . trim ( ) ; if ( ! whiteList . equals ( ALL_VALUES ) ) { set = new HashSet < > ( ) ; for ( String name : whiteList . split ( "," ) ) { name = name . trim ( ) ; if ( ! name . isEmpty ( ) ) { set . add ( "+" + name . trim ( ) ) ; } } } } catch ( IOException ex ) { throw new IllegalArgumentException ( String . format ( WHITE_LIST_COULD_NOT_LOAD_FILE_MSG , whiteListFile , ex . toString ( ) ) , ex ) ; } } return set ;


signature:
public class BootstrapMain { /** * Visible for testing */
public static Map < String , List < URL > > getStageLibrariesClasspaths ( String stageLibrariesDir , String librariesExtraDir , final Set < String > stageLibs , String libsCommonLibDir ) throws Exception { } }

implementation:
Map < String , List < URL > > map = new LinkedHashMap < String , List < URL > > ( ) ; File baseDir = new File ( stageLibrariesDir ) . getAbsoluteFile ( ) ; if ( baseDir . exists ( ) ) { File [ ] libDirs = baseDir . listFiles ( createStageLibFilter ( stageLibs ) ) ; StringBuilder commonLibJars = new StringBuilder ( ) ; if ( libsCommonLibDir != null ) { commonLibJars . append ( new File ( libsCommonLibDir ) . getAbsolutePath ( ) ) . append ( FILE_SEPARATOR ) . append ( JARS_WILDCARD ) . append ( CLASSPATH_SEPARATOR ) ; } for ( File libDir : libDirs ) { File jarsDir = new File ( libDir , STAGE_LIB_JARS_DIR ) ; File etc = new File ( libDir , STAGE_LIB_CONF_DIR ) ; if ( ! jarsDir . exists ( ) ) { throw new IllegalArgumentException ( String . format ( MISSING_STAGE_LIB_JARS_DIR_MSG , libDir ) ) ; } StringBuilder sb = new StringBuilder ( ) ; if ( etc . exists ( ) ) { sb . append ( etc . getAbsolutePath ( ) ) . append ( FILE_SEPARATOR ) . append ( CLASSPATH_SEPARATOR ) ; } sb . append ( commonLibJars ) ; sb . append ( jarsDir . getAbsolutePath ( ) ) . append ( FILE_SEPARATOR ) . append ( JARS_WILDCARD ) ; // add extralibs if avail
if ( librariesExtraDir != null ) { File libExtraDir = new File ( librariesExtraDir , libDir . getName ( ) ) ; if ( libExtraDir . exists ( ) ) { File extraJarsDir = new File ( libExtraDir , STAGE_LIB_JARS_DIR ) ; if ( extraJarsDir . exists ( ) ) { sb . append ( CLASSPATH_SEPARATOR ) . append ( extraJarsDir . getAbsolutePath ( ) ) . append ( FILE_SEPARATOR ) . append ( JARS_WILDCARD ) ; } File extraEtc = new File ( libExtraDir , STAGE_LIB_CONF_DIR ) ; if ( extraEtc . exists ( ) ) { sb . append ( CLASSPATH_SEPARATOR ) . append ( extraEtc . getAbsolutePath ( ) ) ; } } } map . put ( libDir . getParentFile ( ) . getName ( ) + FILE_SEPARATOR + libDir . getName ( ) , getClasspathUrls ( sb . toString ( ) ) ) ; } } else { throw new IllegalArgumentException ( String . format ( MISSING_STAGE_LIBRARIES_DIR_MSG , baseDir ) ) ; } return map ;


signature:
public class BootstrapMain { /** * Visible for testing */
public static List < URL > getClasspathUrls ( String classPath ) throws Exception { } }

implementation: List < URL > urls = new ArrayList < URL > ( ) ; for ( String path : classPath . split ( CLASSPATH_SEPARATOR ) ) { if ( ! path . isEmpty ( ) ) { if ( path . toLowerCase ( ) . endsWith ( JARS_WILDCARD ) ) { path = path . substring ( 0 , path . length ( ) - JARS_WILDCARD . length ( ) ) ; File f = new File ( path ) . getAbsoluteFile ( ) ; if ( f . exists ( ) ) { File [ ] jars = f . listFiles ( new FileFilter ( ) { @ Override public boolean accept ( File pathname ) { return pathname . getName ( ) . toLowerCase ( ) . endsWith ( JARS_WILDCARD . substring ( 1 ) ) ; } } ) ; for ( File jar : jars ) { urls . add ( jar . toURI ( ) . toURL ( ) ) ; } } else { throw new IllegalArgumentException ( String . format ( CLASSPATH_DIR_DOES_NOT_EXIST_MSG , f ) ) ; } } else { if ( ! path . endsWith ( FILE_SEPARATOR ) ) { path = path + FILE_SEPARATOR ; } File f = new File ( path ) . getAbsoluteFile ( ) ; if ( f . exists ( ) ) { if ( f . isDirectory ( ) ) { urls . add ( f . toURI ( ) . toURL ( ) ) ; } else { throw new IllegalArgumentException ( String . format ( CLASSPATH_PATH_S_IS_NOT_A_DIR_MSG , f ) ) ; } } else { throw new IllegalArgumentException ( String . format ( CLASSPATH_DIR_DOES_NOT_EXIST_MSG , f ) ) ; } } } } return urls ;


signature:
public class BaseKafkaSource { /** * This API is being used by ClusterKafkaSource */
public int getParallelism ( ) throws StageException { } }

implementation:
if ( originParallelism == 0 ) { // origin parallelism is not yet calculated
originParallelism = kafkaValidationUtil . getPartitionCount ( conf . metadataBrokerList , conf . topic , new HashMap < String , Object > ( conf . kafkaConsumerConfigs ) , 3 , 1000 ) ; if ( originParallelism < 1 ) { throw new StageException ( KafkaErrors . KAFKA_42 , conf . topic ) ; } } return originParallelism ;


signature:
public class SDCClassLoader { /** * Arranges the urls in the following order :
 * < ul >
 * < li > stage lib jars < / li >
 * < li > protolib jars < / li >
 * < li > non protolib jars < / li >
 * < / ul >
 * @ param stageLibName
 * @ param urls
 * @ return */
static List < URL > bringStageAndProtoLibsToFront ( String stageLibName , List < URL > urls ) { } }

implementation:
List < URL > otherJars = new ArrayList < > ( ) ; List < URL > protolibJars = new ArrayList < > ( ) ; List < URL > stageLibjars = new ArrayList < > ( ) ; for ( URL url : urls ) { String str = url . toExternalForm ( ) ; if ( str . endsWith ( ".jar" ) ) { int nameIdx = str . lastIndexOf ( "/" ) ; if ( nameIdx > - 1 ) { String jarName = str . substring ( nameIdx + 1 ) ; if ( jarName . contains ( "-protolib-" ) ) { // adding only protolib jars
protolibJars . add ( url ) ; } else if ( jarName . contains ( stageLibName ) ) { stageLibjars . add ( url ) ; } else { otherJars . add ( url ) ; } } else { otherJars . add ( url ) ; } } else { otherJars . add ( url ) ; } } List < URL > allJars = new ArrayList < > ( ) ; if ( stageLibjars . size ( ) != 1 ) { throw new ExceptionInInitializerError ( "Expected exactly 1 stage lib jar but found " + stageLibjars . size ( ) + " with name " + stageLibName ) ; } allJars . addAll ( stageLibjars ) ; allJars . addAll ( protolibJars ) ; allJars . addAll ( otherJars ) ; return allJars ;


signature:
public class BigQueryDelegate { /** * Executes a query request and returns the results . A timeout is required to avoid
 * waiting for an indeterminate amount of time . If the query fails to complete within the
 * timeout it is aborted .
 * @ param queryConfig request describing the query to execute
 * @ param timeout maximum amount of time to wait for job completion before attempting to cancel
 * @ return result of the query .
 * @ throws StageException if the query does not complete within the requested timeout or
 * there is an error executing the query . */
public TableResult runQuery ( QueryJobConfiguration queryConfig , long timeout , long pageSize ) throws StageException { } }

implementation:
checkArgument ( timeout >= 1000 , "Timeout must be at least one second." ) ; Instant maxTime = Instant . now ( ) . plusMillis ( timeout ) ; // Create a job ID so that we can safely retry .
JobId jobId = JobId . of ( UUID . randomUUID ( ) . toString ( ) ) ; JobInfo jobInfo = JobInfo . newBuilder ( queryConfig ) . setJobId ( jobId ) . build ( ) ; Job queryJob = bigquery . create ( jobInfo ) ; // Check for errors
if ( queryJob == null ) { LOG . error ( "Job no longer exists: {}" , jobInfo ) ; throw new RuntimeException ( "Job no longer exists: " + jobInfo ) ; } else if ( queryJob . getStatus ( ) . getError ( ) != null ) { BigQueryError error = queryJob . getStatus ( ) . getError ( ) ; LOG . error ( "Query Job execution error: {}" , error ) ; throw new StageException ( Errors . BIGQUERY_02 , error ) ; } // Should consider using . waitFor ( RetryOption . totalTimeout ( ) )
while ( ! queryJob . isDone ( ) ) { if ( Instant . now ( clock ) . isAfter ( maxTime ) || ! ThreadUtil . sleep ( 100 ) ) { if ( bigquery . cancel ( queryJob . getJobId ( ) ) ) { LOG . info ( "Job {} cancelled successfully." , queryJob . getJobId ( ) ) ; } else { LOG . warn ( "Job {} not found" , queryJob . getJobId ( ) ) ; } throw new StageException ( Errors . BIGQUERY_00 ) ; } } if ( queryJob . getStatus ( ) . getError ( ) != null ) { String errorMsg = queryJob . getStatus ( ) . getError ( ) . toString ( ) ; throw new StageException ( Errors . BIGQUERY_02 , errorMsg ) ; } // Get the results .
TableResult result = null ; try { result = queryJob . getQueryResults ( QueryResultsOption . pageSize ( pageSize ) ) ; } catch ( InterruptedException e ) { String errorMsg = e . getMessage ( ) ; throw new StageException ( Errors . BIGQUERY_02 , errorMsg ) ; } return result ;


signature:
public class BigQueryDelegate { /** * Returns the SDC record { @ link Field } type mapped from a
 * BigQuery { @ link com . google . cloud . bigquery . Field } type .
 * @ param field A BigQuery { @ link com . google . cloud . bigquery . Field }
 * @ return SDC record { @ link Field . Type } */
public Field . Type asRecordFieldType ( com . google . cloud . bigquery . Field field ) { } }

implementation: Field . Type type = asRecordFieldTypeFunction . apply ( field ) ; return checkNotNull ( type , Utils . format ( "Unsupported type '{}'" , field . getType ( ) ) ) ;


signature:
public class BigQueryDelegate { /** * Converts a list of BigQuery fields to SDC Record fields .
 * The provided parameters must have matching lengths .
 * If not , an unchecked exception will be thrown . This method is called when the resulting
 * container type should be a { @ link Field . Type # LIST _ MAP }
 * @ param schema List of { @ link com . google . cloud . bigquery . Field } representing the schema
 * at the current level of nesting . For example , if processing a
 * { @ link LegacySQLTypeName # RECORD } or { @ link StandardSQLTypeName # STRUCT }
 * this would only include the fields for that particular data structure and not the entire
 * result set .
 * @ param values List of { @ link FieldValue } representing the values to set in the generated
 * fields .
 * @ return Specifically , a LinkedHashMap as the return value of this method is expected to be
 * used to create a
 * { @ link Field . Type # LIST _ MAP } field . */
public LinkedHashMap < String , Field > fieldsToMap ( // NOSONAR
List < com . google . cloud . bigquery . Field > schema , List < FieldValue > values ) { } }

implementation: checkState ( schema . size ( ) == values . size ( ) , "Schema '{}' and Values '{}' sizes do not match." , schema . size ( ) , values . size ( ) ) ; LinkedHashMap < String , Field > root = new LinkedHashMap < > ( ) ; for ( int i = 0 ; i < values . size ( ) ; i ++ ) { FieldValue value = values . get ( i ) ; com . google . cloud . bigquery . Field field = schema . get ( i ) ; if ( value . getAttribute ( ) . equals ( FieldValue . Attribute . PRIMITIVE ) ) { root . put ( field . getName ( ) , fromPrimitiveField ( field , value ) ) ; } else if ( value . getAttribute ( ) . equals ( FieldValue . Attribute . RECORD ) ) { root . put ( field . getName ( ) , Field . create ( fieldsToMap ( field . getSubFields ( ) , value . getRecordValue ( ) ) ) ) ; } else if ( value . getAttribute ( ) . equals ( FieldValue . Attribute . REPEATED ) ) { root . put ( field . getName ( ) , Field . create ( fromRepeatedField ( field , value . getRepeatedValue ( ) ) ) ) ; } } return root ;


signature:
public class BigQueryDelegate { /** * Repeated fields are simply fields that may appear more than once . In SDC we will
 * represent them as a list field . For example a repeated field of type RECORD would be a
 * { @ link Field . Type # LIST } of
 * { @ link Field . Type # LIST _ MAP } and a repeated field of type STRING would be
 * a { @ link Field . Type # LIST } of { @ link Field . Type # STRING } .
 * @ param schema The field metadata for the repeated field
 * @ param repeatedValue a list of individual field values that represent the repeated field
 * @ return a list of SDC record fields .
 * Intended to be used for creating a { @ link Field . Type # LIST } { @ link Field } */
public List < Field > fromRepeatedField ( com . google . cloud . bigquery . Field schema , List < FieldValue > repeatedValue ) { } }

implementation: if ( repeatedValue . isEmpty ( ) ) { return Collections . emptyList ( ) ; } FieldValue . Attribute repeatedFieldType = repeatedValue . get ( 0 ) . getAttribute ( ) ; BiFunction < com . google . cloud . bigquery . Field , FieldValue , Field > transform = transforms . get ( repeatedFieldType ) ; return repeatedValue . stream ( ) . map ( v -> transform . apply ( schema , v ) ) . collect ( Collectors . toList ( ) ) ;