signature:
public class AvroTypeUtil { /** * Returns a { @ link BigDecimal } from the given bytes and scale . The bytes must adhere to the format that Avro
 * stores decimals in .
 * < br >
 * Avro stores decimal values as two ' s complement , big - endian for the integral portion , then the decimal place
 * separately ( via the scale ) .
 * @ see < a href = " https : / / avro . apache . org / docs / 1.8.1 / spec . html # Decimal " > Avro documentation < / a >
 * @ param decimalBytes the bytes of the decimal value ' s integral value , assumed to be two ' s complement , big - endian
 * @ param scale the scale of the decimal value
 * @ return a { @ link BigDecimal } constructed from the decimalBytes and scale */
public static BigDecimal bigDecimalFromBytes ( byte [ ] decimalBytes , int scale ) { } }

implementation: final BigInteger bigInt = new BigInteger ( decimalBytes ) ; return new BigDecimal ( bigInt , scale ) ;


signature:
public class JmsTargetUpgrader { /** * Migrating to service for data format library . */
private void upgradeV1ToV2 ( List < Config > configs , Context context ) { } }

implementation:
List < Config > dataFormatConfigs = configs . stream ( ) . filter ( c -> c . getName ( ) . startsWith ( "dataFormat" ) ) . collect ( Collectors . toList ( ) ) ; // Remove those configs
configs . removeAll ( dataFormatConfigs ) ; // Provide proper prefix
dataFormatConfigs = dataFormatConfigs . stream ( ) . map ( c -> new Config ( c . getName ( ) . replace ( "dataFormatConfig." , "dataGeneratorFormatConfig." ) , c . getValue ( ) ) ) . collect ( Collectors . toList ( ) ) ; // And finally register new service
context . registerService ( DataFormatGeneratorService . class , dataFormatConfigs ) ;


signature:
public class RecordWriter { /** * buffer size . */
public long getLength ( ) throws IOException { } }

implementation: long length = - 1 ; if ( generator != null ) { length = textOutputStream . getByteCount ( ) ; } else if ( seqWriter != null ) { length = seqWriter . getLength ( ) ; } return length ;


signature:
public class BaseClusterProvider { /** * Copy BlobStore resources to data directory */
private void copyBlobstore ( List < String > blobStoreResources , File rootDataDir , File pipelineDir ) throws IOException { } }

implementation:
if ( blobStoreResources == null ) { return ; } File blobstoreDir = new File ( runtimeInfo . getDataDir ( ) , BLOBSTORE_BASE_DIR ) ; File stagingBlobstoreDir = new File ( rootDataDir , BLOBSTORE_BASE_DIR ) ; if ( ! stagingBlobstoreDir . exists ( ) ) { if ( ! stagingBlobstoreDir . mkdirs ( ) ) { throw new RuntimeException ( "Failed to create blobstore directory: " + pipelineDir . getPath ( ) ) ; } } for ( String blobstoreFile : blobStoreResources ) { File srcFile = new File ( blobstoreDir , blobstoreFile ) ; if ( srcFile . exists ( ) ) { final File dstFile = new File ( stagingBlobstoreDir , srcFile . getName ( ) ) ; if ( srcFile . canRead ( ) ) { // ignore files which cannot be read
try ( InputStream in = new FileInputStream ( ( srcFile ) ) ) { try ( OutputStream out = new FileOutputStream ( ( dstFile ) ) ) { IOUtils . copy ( in , out ) ; } } } } }


signature:
public class SQLListener { /** * Reset the listener to use with the next statement . All column information is cleared . */
public void reset ( ) { } }

implementation: columns . clear ( ) ; this . columnsExpected = null ; columnNames = null ; table = null ; schema = null ; insideStatement = false ;


signature:
public class ThreadHealthReporter { /** * Updates gauge for the registered thread with the given details .
 * Note that the value of the threadName argument must match the one used to register .
 * @ param threadName
 * @ param scheduledDelay
 * @ param timestamp
 * @ return */
public boolean reportHealth ( String threadName , int scheduledDelay , long timestamp ) { } }

implementation: ThreadHealthReport threadHealthReport = new ThreadHealthReport ( threadName , scheduledDelay , timestamp ) ; if ( threadToGaugeMap . containsKey ( threadName ) ) { threadToGaugeMap . get ( threadName ) . setThreadHealthReport ( threadHealthReport ) ; return true ; } return false ;


signature:
public class ThreadHealthReporter { /** * Creates and registers a Gauge with the given thread name .
 * The same name must be used to report health .
 * @ param threadName
 * @ return */
public boolean register ( String threadName ) { } }

implementation: if ( threadToGaugeMap . containsKey ( threadName ) ) { return false ; } ThreadHealthReportGauge threadHealthReportGauge = new ThreadHealthReportGauge ( ) ; MetricsConfigurator . createGauge ( metrics , getHealthGaugeName ( threadName ) , threadHealthReportGauge , name , rev ) ; threadToGaugeMap . put ( threadName , threadHealthReportGauge ) ; return true ;


signature:
public class RuntimeInfo { /** * Store configuration from control hub in persistent manner inside data directory . This configuration will be
 * loaded on data collector start and will override any configuration from sdc . properties .
 * This method call is able to remove existing properties if the value is " null " . Please note that the removal will
 * only happen from the ' override ' file . This method does not have the capability to remove configuration directly
 * from sdc . properties .
 * @ param runtimeInfo RuntimeInfo instance
 * @ param newConfigs New set of config properties
 * @ throws IOException */
public static void storeControlHubConfigs ( RuntimeInfo runtimeInfo , Map < String , String > newConfigs ) throws IOException { } }

implementation:
File configFile = new File ( runtimeInfo . getDataDir ( ) , SCH_CONF_OVERRIDE ) ; Properties properties = new Properties ( ) ; // Load existing properties from disk if they exists
if ( configFile . exists ( ) ) { try ( FileReader reader = new FileReader ( configFile ) ) { properties . load ( reader ) ; } } // Propagate updated configuration
for ( Map . Entry < String , String > entry : newConfigs . entrySet ( ) ) { if ( entry . getValue ( ) == null ) { properties . remove ( entry . getKey ( ) ) ; } else { properties . setProperty ( entry . getKey ( ) , entry . getValue ( ) ) ; } } // Store the new updated configuration back to disk
try ( FileWriter writer = new FileWriter ( configFile ) ) { properties . store ( writer , null ) ; }


signature:
public class LdapLoginModule { /** * get the available information about the user
 * for this LoginModule , the credential can be null which will result in a
 * binding ldap authentication scenario
 * roles are also an optional concept if required
 * @ param username
 * @ return the userinfo for the username
 * @ throws Exception */
@ Override public UserInfo getUserInfo ( String username ) throws Exception { } }

implementation: LdapEntry entry = getEntryWithCredential ( username ) ; if ( entry == null ) { return null ; } String pwdCredential = getUserCredential ( entry ) ; pwdCredential = convertCredentialLdapToJetty ( pwdCredential ) ; Credential credential = Credential . getCredential ( pwdCredential ) ; List < String > roles = getUserRoles ( username , entry . getDn ( ) ) ; return new UserInfo ( username , credential , roles ) ;


signature:
public class LdapLoginModule { /** * attempts to get the users credentials from the users context
 * NOTE : this is not an user authenticated operation
 * @ param username
 * @ return
 * @ throws LoginException */
private LdapEntry getEntryWithCredential ( String username ) throws LdapException { } }

implementation:
if ( StringUtils . isBlank ( _userObjectClass ) || StringUtils . isBlank ( _userIdAttribute ) || StringUtils . isBlank ( _userBaseDn ) || StringUtils . isBlank ( _userPasswordAttribute ) ) { LOG . error ( "Failed to get user because at least one of the following is null : " + "[_userObjectClass, _userIdAttribute, _userBaseDn, _userPasswordAttribute ]" ) ; return null ; } // Create the format of & ( objectClass = _ userObjectClass ) ( _ userIdAttribute = { user } ) )
String userFilter = buildFilter ( _userFilter , _userObjectClass , _userIdAttribute ) ; if ( userFilter . contains ( "{user}" ) ) { userFilter = userFilter . replace ( "{user}" , username ) ; } LOG . debug ( "Searching user using the filter {} on user baseDn {}" , userFilter , _userBaseDn ) ; // Get the group names from each group , which is obtained from roleNameAttribute attribute .
SearchRequest request = new SearchRequest ( _userBaseDn , userFilter , _userPasswordAttribute ) ; request . setSearchScope ( SearchScope . SUBTREE ) ; request . setSizeLimit ( 1 ) ; try { SearchOperation search = new SearchOperation ( conn ) ; org . ldaptive . SearchResult result = search . execute ( request ) . getResult ( ) ; LdapEntry entry = result . getEntry ( ) ; LOG . info ( "Found user?: {}" , entry != null ) ; return entry ; } catch ( LdapException ex ) { LOG . error ( "{}" , ex . toString ( ) , ex ) ; return null ; }


signature:
public class LdapLoginModule { /** * attempts to get the users roles
 * NOTE : this is not an user authenticated operation
 * @ param username
 * @ return
 * @ throws LoginException */
private List < String > getUserRoles ( String username , String userDn ) { } }

implementation:
List < String > roleList = new ArrayList < > ( ) ; if ( StringUtils . isBlank ( _roleBaseDn ) || StringUtils . isBlank ( _roleObjectClass ) || StringUtils . isBlank ( _roleNameAttribute ) || StringUtils . isBlank ( _roleMemberAttribute ) ) { LOG . debug ( "Failed to get roles because at least one of the following is null : " + "[_roleBaseDn, _roleObjectClass, _roleNameAttribute, _roleMemberAttribute ]" ) ; return roleList ; } String roleFilter = buildFilter ( _roleFilter , _roleObjectClass , _roleMemberAttribute ) ; if ( _roleFilter . contains ( DN ) ) { userDn = userDn . replace ( "\\" , "\\\\\\" ) ; roleFilter = roleFilter . replace ( DN , userDn ) ; } else if ( _roleFilter . contains ( USER ) ) { roleFilter = roleFilter . replace ( USER , username ) ; } else { LOG . error ( "roleFilter contains invalid filter {}. Check the roleFilter option" ) ; return roleList ; } LOG . debug ( "Searching roles using the filter {} on role baseDn {}" , roleFilter , _roleBaseDn ) ; // Get the group names from each group , which is obtained from roleNameAttribute attribute .
SearchRequest request = new SearchRequest ( _roleBaseDn , roleFilter , _roleNameAttribute ) ; request . setSearchScope ( SearchScope . SUBTREE ) ; try { SearchOperation search = new SearchOperation ( conn ) ; org . ldaptive . SearchResult result = search . execute ( request ) . getResult ( ) ; Collection < LdapEntry > entries = result . getEntries ( ) ; LOG . info ( "Found roles?: {}" , ! ( entries == null || entries . isEmpty ( ) ) ) ; if ( entries != null ) { for ( LdapEntry entry : entries ) { roleList . add ( entry . getAttribute ( ) . getStringValue ( ) ) ; } } } catch ( LdapException ex ) { LOG . error ( ex . getMessage ( ) , ex ) ; } LOG . info ( "Found roles: {}" , roleList ) ; return roleList ;


signature:
public class LdapLoginModule { /** * Given a filter ( user / role filter ) , replace attributes using given information from config .
 * This will create complete filter , which will look like & ( objectClass = inetOrgPerson ) ( uid = { user } ) )
 * @ param attrFilter We obtain this part from config
 * @ param objClass objectClass
 * @ param attrName attribute name
 * @ return Complete filter */
@ VisibleForTesting static String buildFilter ( String attrFilter , String objClass , String attrName ) { } }

implementation:
// check if the filter has surrounding " ( ) "
if ( ! attrFilter . startsWith ( "(" ) ) { attrFilter = "(" + attrFilter ; } if ( ! attrFilter . endsWith ( ")" ) ) { attrFilter = attrFilter + ")" ; } return String . format ( filterFormat , objClass , String . format ( attrFilter , attrName ) ) ;


signature:
public class LdapLoginModule { /** * since ldap uses a context bind for valid authentication checking , we override login ( )
 * if credentials are not available from the users context or if we are forcing the binding check
 * then we try a binding authentication check , otherwise if we have the users encoded password then
 * we can try authentication via that mechanic
 * @ return true if authenticated , false otherwise
 * @ throws LoginException */
@ Override public boolean login ( ) throws LoginException { } }

implementation:
try { if ( getCallbackHandler ( ) == null ) { throw new LoginException ( "No callback handler" ) ; } if ( conn == null ) { return false ; } Callback [ ] callbacks = configureCallbacks ( ) ; getCallbackHandler ( ) . handle ( callbacks ) ; String webUserName = ( ( NameCallback ) callbacks [ 0 ] ) . getName ( ) ; Object webCredential = ( ( ObjectCallback ) callbacks [ 1 ] ) . getObject ( ) ; if ( webUserName == null || webCredential == null ) { setAuthenticated ( false ) ; return isAuthenticated ( ) ; } // Please see the following stackoverflow article
// http : / / security . stackexchange . com / questions / 6713 / ldap - security - problems
// Some LDAP implementation " MAY " accept empty password as a sign of anonymous connection and thus
// return " true " for the authentication request .
if ( ( webCredential instanceof String ) && ( ( String ) webCredential ) . isEmpty ( ) ) { LOG . info ( "Ignoring login request for user {} as the password is empty." , webUserName ) ; setAuthenticated ( false ) ; return isAuthenticated ( ) ; } if ( _forceBindingLogin ) { return bindingLogin ( webUserName , webCredential ) ; } // This sets read and the credential
UserInfo userInfo = getUserInfo ( webUserName ) ; if ( userInfo == null ) { setAuthenticated ( false ) ; return false ; } JAASUserInfo jaasUserInfo = new JAASUserInfo ( userInfo ) ; jaasUserInfo . fetchRoles ( ) ; setCurrentUser ( jaasUserInfo ) ; if ( webCredential instanceof String ) { return credentialLogin ( Credential . getCredential ( ( String ) webCredential ) ) ; } return credentialLogin ( webCredential ) ; } catch ( UnsupportedCallbackException e ) { throw new LoginException ( "Error obtaining callback information." ) ; } catch ( IOException e ) { LOG . error ( "IO Error performing login" , e ) ; } catch ( Exception e ) { LOG . error ( "IO Error performing login" , e ) ; } return false ;


signature:
public class LdapLoginModule { /** * password supplied authentication check
 * @ param webCredential
 * @ return true if authenticated
 * @ throws LoginException */
protected boolean credentialLogin ( Object webCredential ) throws LoginException { } }

implementation: boolean credResult = getCurrentUser ( ) . checkCredential ( webCredential ) ; setAuthenticated ( credResult ) ; if ( ! credResult ) { LOG . warn ( "Authentication failed - Possibly the user password is wrong" ) ; } return isAuthenticated ( ) ;


signature:
public class LdapLoginModule { /** * binding authentication check
 * This method of authentication works only if the user branch of the DIT ( ldap tree )
 * has an ACI ( access control instruction ) that allow the access to any user or at least
 * for the user that logs in .
 * @ param username
 * @ param password
 * @ return true always
 * @ throws LoginException */
public boolean bindingLogin ( String username , Object password ) throws Exception { } }

implementation:
if ( StringUtils . isBlank ( _userObjectClass ) || StringUtils . isBlank ( _userIdAttribute ) || StringUtils . isBlank ( _userBaseDn ) ) { LOG . error ( "Failed to get user because at least one of the following is null : " + "[_userObjectClass, _userIdAttribute, _userBaseDn ]" ) ; return false ; } LdapEntry userEntry = authenticate ( username , password ) ; if ( userEntry == null ) { return false ; } // If authenticated by LDAP server , the returned LdapEntry contains full DN of the user
String userDn = userEntry . getDn ( ) ; if ( userDn == null ) { // This shouldn ' t happen if LDAP server is configured properly .
LOG . error ( "userDn is found null for the user {}" , username ) ; return false ; } List < String > roles = getUserRoles ( username , userDn ) ; // Authentication already succeeded . We won ' t store user password so passing empty credential
UserInfo userInfo = new UserInfo ( username , Credential . getCredential ( "" ) , roles ) ; JAASUserInfo jaasUserInfo = new JAASUserInfo ( userInfo ) ; jaasUserInfo . fetchRoles ( ) ; setCurrentUser ( jaasUserInfo ) ; setAuthenticated ( true ) ; return true ;


signature:
public class LdapLoginModule { /** * Perform authentication with given username and password .
 * Receive the result from Ldap server
 * @ param username Username that user entered to login
 * @ param password Password that user entered to login
 * @ return LdapEntry which contains all user attributes */
private LdapEntry authenticate ( String username , Object password ) { } }

implementation:
try { SearchDnResolver dnResolver = new SearchDnResolver ( new DefaultConnectionFactory ( connConfig ) ) ; dnResolver . setBaseDn ( _userBaseDn ) ; dnResolver . setSubtreeSearch ( true ) ; String userFilter = buildFilter ( _userFilter , _userObjectClass , _userIdAttribute ) ; LOG . debug ( "Searching a user with filter {} where user is {}" , userFilter , username ) ; dnResolver . setUserFilter ( userFilter ) ; // Set Authenticator with username and password . It will return the user if username / password matches .
BindAuthenticationHandler authHandler = new BindAuthenticationHandler ( new DefaultConnectionFactory ( connConfig ) ) ; Authenticator auth = new Authenticator ( dnResolver , authHandler ) ; AuthenticationRequest authRequest = new AuthenticationRequest ( ) ; authRequest . setUser ( username ) ; if ( password instanceof char [ ] ) { authRequest . setCredential ( new org . ldaptive . Credential ( new String ( ( char [ ] ) password ) ) ) ; } else if ( password instanceof String ) { authRequest . setCredential ( new org . ldaptive . Credential ( ( String ) password ) ) ; } else { LOG . error ( "Unexpected type for password '{}'" , ( password != null ) ? password . getClass ( ) : "NULL" ) ; return null ; } String [ ] userRoleAttribute = ReturnAttributes . ALL . value ( ) ; authRequest . setReturnAttributes ( userRoleAttribute ) ; LOG . debug ( "Retrieved authenticator from factory: {}" , auth ) ; LOG . debug ( "Retrieved authentication request from factory: {}" , authRequest ) ; AuthenticationResponse response = auth . authenticate ( authRequest ) ; LOG . info ( "Found user?: {}" , response . getResult ( ) ) ; if ( response . getResult ( ) ) { LdapEntry entry = response . getLdapEntry ( ) ; return entry ; } else { // User not found . Most likely username / password didn ' t match . Log the reason .
LOG . error ( "Result code: {} - {}" , response . getResultCode ( ) , response . getMessage ( ) ) ; }


signature:
public class UpgraderUtils { /** * Move a series of config values from old names to new names ( one to one correspondence ) . Config values will be
 * preserved .
 * @ param configs list of config objects ( will be modified )
 * @ param names mapping of old to new names , in the form of [ oldName1 , newName1 , oldName2 , newName2 , . . . ]
 * @ return */
public static int moveAllTo ( List < Config > configs , String ... names ) { } }

implementation: Map < String , String > nameMap = new HashMap < > ( ) ; if ( names . length % 2 == 1 ) { throw new IllegalArgumentException ( "names was of uneven length" ) ; } for ( int i = 0 ; i < names . length ; ) { nameMap . put ( names [ i ] , names [ i + 1 ] ) ; i += 2 ; } return moveAllTo ( configs , nameMap ) ;


signature:
public class UpgraderUtils { /** * Move a series of config values from old names to new names ( one to one correspondence ) . Config values will be
 * preserved .
 * @ param configs list of config objects ( will be modified )
 * @ param oldToNewNames a map of old to new config names
 * @ return the number of config objects moved from old to new names */
public static int moveAllTo ( List < Config > configs , Map < String , String > oldToNewNames ) { } }

implementation: List < Config > configsToAdd = new ArrayList < > ( ) ; List < Config > configsToRemove = new ArrayList < > ( ) ; int numMoved = 0 ; for ( Config config : configs ) { final String oldName = config . getName ( ) ; if ( oldToNewNames . containsKey ( oldName ) ) { configsToRemove . add ( config ) ; final Object value = config . getValue ( ) ; final String newName = oldToNewNames . get ( oldName ) ; configsToAdd . add ( new Config ( newName , value ) ) ; LOG . info ( String . format ( "Moving config value %s from old name %s to new name %s" , value , oldName , newName ) ) ; numMoved ++ ; } } configs . removeAll ( configsToRemove ) ; configs . addAll ( configsToAdd ) ; return numMoved ;


signature:
public class UpgraderUtils { /** * Returns a { @ link Config } object from the supplied list with the supplied name , if it exists . If a non - null
 * Config is returned , the supplied list of { @ code configs } will be modified such that it no longer contains the
 * returned value .
 * @ param configs list of config objects ( will not be modified )
 * @ param name the config to return , based on name
 * @ return the config object by the given name , if it exists , and null otherwise */
public static Config getAndRemoveConfigWithName ( List < Config > configs , String name ) { } }

implementation: final Config config = getConfigWithName ( configs , name ) ; if ( config != null ) { configs . remove ( config ) ; } return config ;


signature:
public class Login { /** * Re - login a principal . This method assumes that { @ link # login ( String ) } has happened already .
 * @ throws javax . security . auth . login . LoginException on a failure */
private synchronized void reLogin ( ) throws LoginException { } }

implementation:
if ( ! isKrbTicket ) { return ; } if ( login == null ) { throw new LoginException ( "Login must be done first" ) ; } if ( ! hasSufficientTimeElapsed ( ) ) { return ; } log . info ( "Initiating logout for {}" , principal ) ; synchronized ( Login . class ) { // register most recent relogin attempt
lastLogin = currentElapsedTime ( ) ; // clear up the kerberos state . But the tokens are not cleared ! As per
// the Java kerberos login module code , only the kerberos credentials
// are cleared
login . logout ( ) ; // login and also update the subject field of this instance to
// have the new credentials ( pass it to the LoginContext constructor )
login = new LoginContext ( loginContextName , subject ) ; log . info ( "Initiating re-login for {}" , principal ) ; login . login ( ) ; }


signature:
public class PipelineConfigurationUpgrader { /** * Upgrade whole pipeline at once and return updated variant .
 * @ param library Stage Library
 * @ param pipelineConf Existing Pipeline Configuration
 * @ param issues Issues
 * @ return Upgraded pipeline configuration or null on any error */
public PipelineConfiguration upgradeIfNecessary ( StageLibraryTask library , PipelineConfiguration pipelineConf , List < Issue > issues ) { } }

implementation:
Preconditions . checkArgument ( issues . isEmpty ( ) , "Given list of issues must be empty." ) ; boolean upgrade ; // Firstly upgrading schema if needed , then data
upgrade = needsSchemaUpgrade ( pipelineConf , issues ) ; if ( upgrade && issues . isEmpty ( ) ) { pipelineConf = upgradeSchema ( library , pipelineConf , issues ) ; } // Something went wrong with the schema upgrade
if ( ! issues . isEmpty ( ) ) { return null ; } // Upgrading data if needed
upgrade = needsUpgrade ( library , pipelineConf , issues ) ; if ( upgrade && issues . isEmpty ( ) ) { // we try to upgrade only if we have all defs for the pipelineConf
pipelineConf = upgrade ( library , pipelineConf , issues ) ; } return ( issues . isEmpty ( ) ) ? pipelineConf : null ;


signature:
public class PipelineConfigurationUpgrader { /** * Upgrade detached stage ( stage not associated directly with the pipeline ) .
 * @ param libraryTask Stage Library
 * @ param stageConf Existing stage configuration
 * @ param issues Issues
 * @ return Upgraded stage configuration or null on any error */
public StageConfiguration upgradeIfNecessary ( StageLibraryTask libraryTask , StageConfiguration stageConf , List < Issue > issues ) { } }

implementation: Preconditions . checkArgument ( issues . isEmpty ( ) , "Given list of issues must be empty." ) ; boolean upgrade = needsUpgrade ( libraryTask , stageConf , issues ) ; if ( upgrade ) { stageConf = upgradeIfNeeded ( libraryTask , stageConf , issues ) ; } return issues . isEmpty ( ) ? stageConf : null ;


signature:
public class PipelineConfigurationUpgrader { /** * Upgrade whole Stage configuration , including all services if needed . Convenience method that will lookup stage
 * definition from the library .
 * This method is idempotent . */
static StageConfiguration upgradeIfNeeded ( StageLibraryTask library , StageConfiguration conf , List < Issue > issues ) { } }

implementation: return upgradeIfNeeded ( library , library . getStage ( conf . getLibrary ( ) , conf . getStageName ( ) , false ) , conf , issues ) ;


signature:
public class PipelineConfigurationUpgrader { /** * Upgrade whole Stage configuration , including all services if needed .
 * This method is idempotent . */
static StageConfiguration upgradeIfNeeded ( StageLibraryTask library , StageDefinition def , StageConfiguration conf , List < Issue > issues ) { } }

implementation:
IssueCreator issueCreator = IssueCreator . getStage ( conf . getInstanceName ( ) ) ; int fromVersion = conf . getStageVersion ( ) ; int toVersion = def . getVersion ( ) ; try { // Firstly upgrade stage itself ( register any new services )
upgradeStageIfNeeded ( def , conf , issueCreator , issues ) ; // And then upgrade all it ' s services
conf . getServices ( ) . forEach ( serviceConf -> upgradeServicesIfNeeded ( library , conf , serviceConf , issueCreator . forService ( serviceConf . getService ( ) . getName ( ) ) , issues ) ) ; } catch ( Exception ex ) { LOG . error ( "Unknown exception during upgrade: " + ex , ex ) ; issues . add ( issueCreator . create ( ContainerError . CONTAINER_0900 , fromVersion , toVersion , ex . toString ( ) ) ) ; } return conf ;


signature:
public class PipelineConfigurationUpgrader { /** * Internal method that will upgrade service configuration if needed .
 * This method is idempotent . */
private static ServiceConfiguration upgradeServicesIfNeeded ( StageLibraryTask library , StageConfiguration stageConf , ServiceConfiguration conf , IssueCreator issueCreator , List < Issue > issues ) { } }

implementation:
ServiceDefinition def = library . getServiceDefinition ( conf . getService ( ) , false ) ; if ( def == null ) { issues . add ( issueCreator . create ( ContainerError . CONTAINER_0903 , conf . getService ( ) . getName ( ) ) ) ; } int fromVersion = conf . getServiceVersion ( ) ; int toVersion = def . getVersion ( ) ; // In case we don ' t need an upgrade
if ( ! needsUpgrade ( toVersion , fromVersion , issueCreator , issues ) ) { return conf ; } ClassLoader cl = Thread . currentThread ( ) . getContextClassLoader ( ) ; try { LOG . warn ( "Upgrading service instance from version '{}' to version '{}'" , conf . getServiceVersion ( ) , def . getVersion ( ) ) ; UpgradeContext upgradeContext = new UpgradeContext ( "" , def . getName ( ) , stageConf . getInstanceName ( ) , fromVersion , toVersion ) ; List < Config > configs = def . getUpgrader ( ) . upgrade ( conf . getConfiguration ( ) , upgradeContext ) ; if ( ! upgradeContext . registeredServices . isEmpty ( ) ) { throw new StageException ( ContainerError . CONTAINER_0904 ) ; } conf . setServiceVersion ( toVersion ) ; conf . setConfig ( configs ) ; } catch ( StageException ex ) { issues . add ( issueCreator . create ( ex . getErrorCode ( ) , ex . getParams ( ) ) ) ; } finally { Thread . currentThread ( ) . setContextClassLoader ( cl ) ; } return conf ;


signature:
public class PipelineConfigurationUpgrader { /** * Internal method that will upgrade only Stage configuration - not the associated services - and only if needed .
 * This method is idempotent . */
static private void upgradeStageIfNeeded ( StageDefinition def , StageConfiguration conf , IssueCreator issueCreator , List < Issue > issues ) { } }

implementation:
int fromVersion = conf . getStageVersion ( ) ; int toVersion = def . getVersion ( ) ; // In case we don ' t need an upgrade
if ( ! needsUpgrade ( toVersion , fromVersion , IssueCreator . getStage ( conf . getInstanceName ( ) ) , issues ) ) { return ; } ClassLoader cl = Thread . currentThread ( ) . getContextClassLoader ( ) ; try { Thread . currentThread ( ) . setContextClassLoader ( def . getStageClassLoader ( ) ) ; LOG . warn ( "Upgrading stage instance '{}' from version '{}' to version '{}'" , conf . getInstanceName ( ) , fromVersion , toVersion ) ; UpgradeContext upgradeContext = new UpgradeContext ( def . getLibrary ( ) , def . getName ( ) , conf . getInstanceName ( ) , fromVersion , toVersion ) ; List < Config > configs = def . getUpgrader ( ) . upgrade ( conf . getConfiguration ( ) , upgradeContext ) ; conf . setStageVersion ( def . getVersion ( ) ) ; conf . setConfig ( configs ) ; // Propagate newly registered services to the StageConfiguration
if ( ! upgradeContext . registeredServices . isEmpty ( ) ) { List < ServiceConfiguration > services = new ArrayList < > ( ) ; services . addAll ( conf . getServices ( ) ) ; // Version - 1 is special to note that this version has been created by stage and not by the service itself
upgradeContext . registeredServices . forEach ( ( s , c ) -> services . add ( new ServiceConfiguration ( s , - 1 , c ) ) ) ; conf . setServices ( services ) ; } } catch ( StageException ex ) { issues . add ( issueCreator . create ( ex . getErrorCode ( ) , ex . getParams ( ) ) ) ; } finally { Thread . currentThread ( ) . setContextClassLoader ( cl ) ; }


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

implementation:
Utils . checkNotNull ( offsets , "offsets" ) ; LOG . trace ( "setOffsets()" ) ; // We look for created directory paths here
findCreatedDirectories ( ) ; // we look for new files only here
findNewFileContexts ( ) ; // we purge file only here
purge ( ) ; super . setOffsets ( offsets ) ; startNewLoop ( ) ;


signature:
public class FieldValueReplacerProcessor { /** * should be replaced by null . */
private List < String > getFieldsToNull ( List < NullReplacerConditionalConfig > nullReplacerConditionalConfigs , Set < String > fieldsThatDoNotExist , Set < String > fieldPaths , Record record ) throws OnRecordErrorException { } }

implementation:
// Gather in this all fields to null
List < String > fieldsToNull = new ArrayList < > ( ) ; for ( NullReplacerConditionalConfig nullReplacerConditionalConfig : nullReplacerConditionalConfigs ) { List < String > fieldNamesToNull = nullReplacerConditionalConfig . fieldsToNull ; // Gather fieldsPathsToNull for this nullReplacerConditionalConfig
List < String > fieldPathsToNull = new ArrayList < > ( ) ; // Gather existing paths for each nullReplacerConditionalConfig
// And if field does not exist gather them in fieldsThatDoNotExist
for ( String fieldNameToNull : fieldNamesToNull ) { try { final List < String > matchingPaths = FieldPathExpressionUtil . evaluateMatchingFieldPaths ( fieldNameToNull , fieldPathEval , fieldPathVars , record , fieldPaths ) ; if ( matchingPaths . isEmpty ( ) ) { // FieldPathExpressionUtil . evaluateMatchingFieldPaths does NOT return the supplied param in its result
// regardless , like FieldRegexUtil # getMatchingFieldPaths did , so we add manually here
fieldsThatDoNotExist . add ( fieldNameToNull ) ; } else { for ( String matchingField : matchingPaths ) { if ( record . has ( matchingField ) ) { fieldPathsToNull . add ( matchingField ) ; } else { fieldsThatDoNotExist . add ( matchingField ) ; } } } } catch ( ELEvalException e ) { LOG . error ( "Error evaluating condition: " + nullReplacerConditionalConfig . condition , e ) ; throw new OnRecordErrorException ( record , Errors . VALUE_REPLACER_07 , fieldNameToNull , e . toString ( ) , e ) ; } } // Now evaluate the condition in nullReplacerConditionalConfig
// If it empty or condition evaluates to true , add all the gathered fields in fieldsPathsToNull
// for this nullReplacerConditionalConfig to fieldsToNull
try { boolean evaluatedCondition = true ; // If it is empty we assume it is true .
if ( ! StringUtils . isEmpty ( nullReplacerConditionalConfig . condition ) ) { evaluatedCondition = nullConditionELEval . eval ( nullConditionELVars , nullReplacerConditionalConfig . condition , Boolean . class ) ; } if ( evaluatedCondition ) { fieldsToNull . addAll ( fieldPathsToNull ) ; } } catch ( ELEvalException e ) { LOG . error ( "Error evaluating condition: " + nullReplacerConditionalConfig . condition , e ) ; throw new OnRecordErrorException ( record , Errors . VALUE_REPLACER_06 , nullReplacerConditionalConfig . condition , e . toString ( ) ) ; } } return fieldsToNull ;


signature:
public class SystemPackage { /** * Checks if a class should be included as a system class .
 * A class is a system class if and only if it matches one of the positive
 * patterns and none of the negative ones .
 * @ param name the class name to check
 * @ param packageList a list of system class configurations .
 * @ return true if the class is a system class */
private static boolean isSystemClass ( String name , List < String > packageList ) { } }

implementation:
boolean result = false ; if ( packageList != null ) { String canonicalName = ClassLoaderUtil . canonicalizeClassOrResource ( name ) ; for ( String c : packageList ) { boolean shouldInclude = true ; if ( c . startsWith ( "-" ) ) { c = c . substring ( 1 ) ; shouldInclude = false ; } if ( canonicalName . startsWith ( c ) ) { if ( c . endsWith ( "." ) // package
|| canonicalName . length ( ) == c . length ( ) // class
|| canonicalName . length ( ) > c . length ( ) // nested
&& canonicalName . charAt ( c . length ( ) ) == '$' ) { if ( shouldInclude ) { result = true ; } else { return false ; } } } } } return result ;


signature:
public class SparkStreamingBinding { /** * Visible For Testing ( TODO - Import guava and make sure it doesn ' t conflict with spark ' s rootclassloader ) */
CheckpointPath getCheckPointPath ( String topic , String consumerGroup ) { } }

implementation: return new CheckpointPath . Builder ( CHECKPOINT_BASE_DIR ) . sdcId ( Utils . getPropertyNotNull ( properties , SDC_ID ) ) . topic ( topic ) . consumerGroup ( consumerGroup ) . pipelineName ( Utils . getPropertyNotNull ( properties , ClusterModeConstants . CLUSTER_PIPELINE_NAME ) ) . build ( ) ;


signature:
public class XEvictingQueue { /** * Adds the given element to this queue . If the queue is currently full , the element at the head
 * of the queue is evicted to make room .
 * @ return { @ code true } always */
@ Override public boolean add ( E e ) { } }

implementation:
checkNotNull ( e ) ; // check before removing
if ( maxSize == 0 ) { return true ; } if ( size ( ) == maxSize ) { delegate . remove ( ) ; } delegate . add ( e ) ; return true ;


signature:
public class XEvictingQueue { /** * Adds the given element to this queue . If the queue is currently full , the element at the head
 * of the queue is evicted to make room and returns the evicted element .
 * @ return the evicted element , < code > NULL < / code > if none . */
public E addAndGetEvicted ( E e ) { } }

implementation:
checkNotNull ( e ) ; // check before removing
if ( maxSize == 0 ) { return null ; } E evicted = null ; if ( size ( ) == maxSize ) { evicted = delegate . remove ( ) ; } delegate . add ( e ) ; return evicted ;


signature:
public class OptionalLoadingCache { /** * Internal method to ensure that we return and not cache the default value if needed */
private Optional < Value > valueOrDefault ( Key key , Optional < Value > value ) { } }

implementation:
// If value is present simply return it
if ( value . isPresent ( ) ) { return value ; } if ( ! cacheMissingValues ) { delegate . invalidate ( key ) ; } return defaultValue ;


signature:
public class PreviewPipelineRunner { /** * Preview only returns data associated with batches , however errors are reported outside of batch context for
 * multi - threaded pipelines . Thus we ' emulate ' the behavior by simply adding into the current batch all ' so - far '
 * reported errors . */
private List < StageOutput > addReportedErrorsIfNeeded ( List < StageOutput > snapshotsOfAllStagesOutput ) { } }

implementation: synchronized ( this . reportedErrors ) { if ( reportedErrors . isEmpty ( ) ) { return snapshotsOfAllStagesOutput ; } try { return snapshotsOfAllStagesOutput . stream ( ) . map ( so -> new StageOutput ( so . getInstanceName ( ) , so . getOutput ( ) , so . getErrorRecords ( ) , reportedErrors . get ( so . getInstanceName ( ) ) , so . getEventRecords ( ) ) ) . collect ( Collectors . toList ( ) ) ; } finally { reportedErrors . clear ( ) ; } }


signature:
public class ProtobufTypeUtil { /** * Returns a protobuf descriptor instance from the provided descriptor file .
 * @ param context Stage context used for finding the SDC resources directory
 * @ param protoDescriptorFile Path to descriptor file relative to SDC _ RESOURCES
 * @ param messageType The name of the message to decode
 * @ param messageTypeToExtensionMap Map of protobuf extensions required for decoding
 * @ param defaultValueMap Map of default values to use for the message
 * @ return protobuf descriptor instance
 * @ throws StageException */
public static Descriptors . Descriptor getDescriptor ( ProtoConfigurableEntity . Context context , String protoDescriptorFile , String messageType , Map < String , Set < Descriptors . FieldDescriptor > > messageTypeToExtensionMap , Map < String , Object > defaultValueMap ) throws StageException { } }

implementation:
File descriptorFileHandle = new File ( context . getResourcesDirectory ( ) , protoDescriptorFile ) ; try ( FileInputStream fin = new FileInputStream ( descriptorFileHandle ) ; ) { DescriptorProtos . FileDescriptorSet set = DescriptorProtos . FileDescriptorSet . parseFrom ( fin ) ; // Iterate over all the file descriptor set computed above and cache dependencies and all encountered
// file descriptors
// this map holds all the dependencies that a given file descriptor has .
// This cached map will be looked up while building FileDescriptor instances
Map < String , Set < Descriptors . FileDescriptor > > fileDescriptorDependentsMap = new HashMap < > ( ) ; // All encountered FileDescriptor instances cached based on their name .
Map < String , Descriptors . FileDescriptor > fileDescriptorMap = new HashMap < > ( ) ; ProtobufTypeUtil . getAllFileDescriptors ( set , fileDescriptorDependentsMap , fileDescriptorMap ) ; // Get the descriptor for the expected message type
Descriptors . Descriptor descriptor = ProtobufTypeUtil . getDescriptor ( set , fileDescriptorMap , protoDescriptorFile , messageType ) ; // Compute and cache all extensions defined for each message type
ProtobufTypeUtil . populateDefaultsAndExtensions ( fileDescriptorMap , messageTypeToExtensionMap , defaultValueMap ) ; return descriptor ; } catch ( FileNotFoundException e ) { throw new StageException ( Errors . PROTOBUF_06 , descriptorFileHandle . getAbsolutePath ( ) , e ) ; } catch ( IOException e ) { throw new StageException ( Errors . PROTOBUF_08 , e . toString ( ) , e ) ; }


signature:
public class ProtobufTypeUtil { /** * Loads a Protobuf file descriptor set into an ubermap of file descriptors .
 * @ param set FileDescriptorSet
 * @ param dependenciesMap FileDescriptor dependency map
 * @ param fileDescriptorMap The populated map of FileDescriptors
 * @ throws StageException */
public static void getAllFileDescriptors ( DescriptorProtos . FileDescriptorSet set , Map < String , Set < Descriptors . FileDescriptor > > dependenciesMap , Map < String , Descriptors . FileDescriptor > fileDescriptorMap ) throws StageException { } }

implementation: List < DescriptorProtos . FileDescriptorProto > fileList = set . getFileList ( ) ; try { for ( DescriptorProtos . FileDescriptorProto fdp : fileList ) { if ( ! fileDescriptorMap . containsKey ( fdp . getName ( ) ) ) { Set < Descriptors . FileDescriptor > dependencies = dependenciesMap . get ( fdp . getName ( ) ) ; if ( dependencies == null ) { dependencies = new LinkedHashSet < > ( ) ; dependenciesMap . put ( fdp . getName ( ) , dependencies ) ; dependencies . addAll ( getDependencies ( dependenciesMap , fileDescriptorMap , fdp , set ) ) ; } Descriptors . FileDescriptor fileDescriptor = Descriptors . FileDescriptor . buildFrom ( fdp , dependencies . toArray ( new Descriptors . FileDescriptor [ dependencies . size ( ) ] ) ) ; fileDescriptorMap . put ( fdp . getName ( ) , fileDescriptor ) ; } } } catch ( Descriptors . DescriptorValidationException e ) { throw new StageException ( Errors . PROTOBUF_07 , e . getDescription ( ) , e ) ; }


signature:
public class ProtobufTypeUtil { /** * Populates a map of protobuf extensions and map with the default values for
 * each message field from a map of file descriptors .
 * @ param fileDescriptorMap Map of file descriptors
 * @ param typeToExtensionMap Map of extensions to populate
 * @ param defaultValueMap Map of default values to populate */
public static void populateDefaultsAndExtensions ( Map < String , Descriptors . FileDescriptor > fileDescriptorMap , Map < String , Set < Descriptors . FieldDescriptor > > typeToExtensionMap , Map < String , Object > defaultValueMap ) { } }

implementation:
for ( Descriptors . FileDescriptor f : fileDescriptorMap . values ( ) ) { // go over every file descriptor and look for extensions and default values of those extensions
for ( Descriptors . FieldDescriptor fieldDescriptor : f . getExtensions ( ) ) { String containingType = fieldDescriptor . getContainingType ( ) . getFullName ( ) ; Set < Descriptors . FieldDescriptor > fieldDescriptors = typeToExtensionMap . get ( containingType ) ; if ( fieldDescriptors == null ) { fieldDescriptors = new LinkedHashSet < > ( ) ; typeToExtensionMap . put ( containingType , fieldDescriptors ) ; } fieldDescriptors . add ( fieldDescriptor ) ; if ( fieldDescriptor . hasDefaultValue ( ) ) { defaultValueMap . put ( containingType + "." + fieldDescriptor . getName ( ) , fieldDescriptor . getDefaultValue ( ) ) ; } } // go over messages within file descriptor and look for all fields and extensions and their defaults
for ( Descriptors . Descriptor d : f . getMessageTypes ( ) ) { addDefaultsAndExtensions ( typeToExtensionMap , defaultValueMap , d ) ; } }


signature:
public class ProtobufTypeUtil { /** * Generates a protobuf descriptor instance from a FileDescriptor set .
 * @ param set set of file descriptors
 * @ param fileDescriptorMap map of message types to file descriptors
 * @ param descriptorFile descriptor file for message to be decoded
 * @ param qualifiedMessageType the name of the message to be decoded
 * @ return protobuf descriptor instance
 * @ throws StageException */
public static Descriptors . Descriptor getDescriptor ( DescriptorProtos . FileDescriptorSet set , Map < String , Descriptors . FileDescriptor > fileDescriptorMap , String descriptorFile , String qualifiedMessageType ) throws StageException { } }

implementation:
// find the FileDescriptorProto which contains the message type
// IF cannot find , then bail out
String packageName = null ; String messageType = qualifiedMessageType ; int lastIndex = qualifiedMessageType . lastIndexOf ( '.' ) ; if ( lastIndex != - 1 ) { packageName = qualifiedMessageType . substring ( 0 , lastIndex ) ; messageType = qualifiedMessageType . substring ( lastIndex + 1 ) ; } DescriptorProtos . FileDescriptorProto file = getFileDescProtoForMsgType ( packageName , messageType , set ) ; if ( file == null ) { // could not find the message type from all the proto files contained in the descriptor file
throw new StageException ( Errors . PROTOBUF_00 , qualifiedMessageType , descriptorFile ) ; } // finally get the FileDescriptor for the message type
Descriptors . FileDescriptor fileDescriptor = fileDescriptorMap . get ( file . getName ( ) ) ; // create builder using the FileDescriptor
// this can only find the top level message types
return fileDescriptor . findMessageTypeByName ( messageType ) ;


signature:
public class ProtobufTypeUtil { /** * Converts a protobuf message to an SDC Record Field .
 * @ param record SDC Record to add field to
 * @ param fieldPath location in record where to insert field .
 * @ param descriptor protobuf descriptor instance
 * @ param messageTypeToExtensionMap protobuf extensions map
 * @ param message message to decode and insert into the specified field path
 * @ return new Field instance representing the decoded message
 * @ throws DataParserException */
public static Field protobufToSdcField ( Record record , String fieldPath , Descriptors . Descriptor descriptor , Map < String , Set < Descriptors . FieldDescriptor > > messageTypeToExtensionMap , Object message ) throws DataParserException { } }

implementation:
LinkedHashMap < String , Field > sdcRecordMapFieldValue = new LinkedHashMap < > ( ) ; // get all the expected fields from the proto file
Map < String , Descriptors . FieldDescriptor > protobufFields = new LinkedHashMap < > ( ) ; for ( Descriptors . FieldDescriptor fieldDescriptor : descriptor . getFields ( ) ) { protobufFields . put ( fieldDescriptor . getName ( ) , fieldDescriptor ) ; } // get all fields in the read message
Map < Descriptors . FieldDescriptor , Object > values = ( ( DynamicMessage ) message ) . getAllFields ( ) ; // for every field present in the proto definition create an sdc field .
for ( Descriptors . FieldDescriptor fieldDescriptor : protobufFields . values ( ) ) { Object value = values . get ( fieldDescriptor ) ; sdcRecordMapFieldValue . put ( fieldDescriptor . getName ( ) , createField ( record , fieldPath , fieldDescriptor , messageTypeToExtensionMap , value ) ) ; } // handle applicable extensions for this message type
if ( messageTypeToExtensionMap . containsKey ( descriptor . getFullName ( ) ) ) { for ( Descriptors . FieldDescriptor fieldDescriptor : messageTypeToExtensionMap . get ( descriptor . getFullName ( ) ) ) { if ( values . containsKey ( fieldDescriptor ) ) { Object value = values . get ( fieldDescriptor ) ; sdcRecordMapFieldValue . put ( fieldDescriptor . getName ( ) , createField ( record , fieldPath , fieldDescriptor , messageTypeToExtensionMap , value ) ) ; } } } // handle unknown fields
// unknown fields can go into the record header
UnknownFieldSet unknownFields = ( ( DynamicMessage ) message ) . getUnknownFields ( ) ; if ( ! unknownFields . asMap ( ) . isEmpty ( ) ) { ByteArrayOutputStream bOut = new ByteArrayOutputStream ( ) ; try { unknownFields . writeDelimitedTo ( bOut ) ; bOut . flush ( ) ; bOut . close ( ) ; } catch ( IOException e ) { throw new DataParserException ( Errors . PROTOBUF_10 , e . toString ( ) , e ) ; } String path = fieldPath . isEmpty ( ) ? FORWARD_SLASH : fieldPath ; byte [ ] bytes = org . apache . commons . codec . binary . Base64 . encodeBase64 ( bOut . toByteArray ( ) ) ; record . getHeader ( ) . setAttribute ( PROTOBUF_UNKNOWN_FIELDS_PREFIX + path , new String ( bytes , StandardCharsets . UTF_8 ) ) ; } return Field . createListMap ( sdcRecordMapFieldValue ) ;


signature:
public class ProtobufTypeUtil { /** * Creates an SDC Record Field from the provided protobuf message and descriptor .
 * @ param record record to be augmented
 * @ param fieldPath field path in the record to insert new Field
 * @ param fieldDescriptor descriptor for this message
 * @ param messageTypeToExtensionMap protobuf type extensions
 * @ param message protobuf message to decode
 * @ return reference to the Field added to the record .
 * @ throws DataParserException */
@ SuppressWarnings ( "unchecked" ) private static Field createField ( Record record , String fieldPath , Descriptors . FieldDescriptor fieldDescriptor , Map < String , Set < Descriptors . FieldDescriptor > > messageTypeToExtensionMap , Object message ) throws DataParserException { } }

implementation:
Field newField ; if ( message == null ) { // If the message does not contain required fields then builder . build ( ) throws UninitializedMessageException
Object defaultValue = null ; Descriptors . FieldDescriptor . JavaType javaType = fieldDescriptor . getJavaType ( ) ; // get default values only for optional fields and non - message types
if ( fieldDescriptor . isOptional ( ) && fieldDescriptor . getJavaType ( ) != Descriptors . FieldDescriptor . JavaType . MESSAGE ) { defaultValue = fieldDescriptor . getDefaultValue ( ) ; // Default value for byte string should be converted to byte array
if ( javaType == Descriptors . FieldDescriptor . JavaType . BYTE_STRING && defaultValue instanceof ByteString ) { defaultValue = ( ( ByteString ) defaultValue ) . toByteArray ( ) ; } } newField = Field . create ( getFieldType ( javaType ) , defaultValue ) ; } else if ( fieldDescriptor . isMapField ( ) ) { // Map entry ( protobuf 3 map )
Map < String , Field > sdcMapFieldValues = new HashMap < > ( ) ; Collection < DynamicMessage > mapEntries = ( Collection < DynamicMessage > ) message ; // MapEntry
for ( DynamicMessage dynamicMessage : mapEntries ) { // MapEntry has 2 fields , key and value
Map < Descriptors . FieldDescriptor , Object > kv = dynamicMessage . getAllFields ( ) ; String key = null ; Object value = null ; Descriptors . FieldDescriptor valueDescriptor = null ; for ( Map . Entry < Descriptors . FieldDescriptor , Object > entry : kv . entrySet ( ) ) { switch ( entry . getKey ( ) . getName ( ) ) { case KEY : key = entry . getValue ( ) . toString ( ) ; break ; case VALUE : value = entry . getValue ( ) ; valueDescriptor = entry . getKey ( ) ; break ; default : throw new DataParserException ( Errors . PROTOBUF_09 , entry . getKey ( ) . getName ( ) ) ; } } if ( key != null && valueDescriptor != null ) { sdcMapFieldValues . put ( key , createSdcField ( record , fieldPath , valueDescriptor , messageTypeToExtensionMap , value ) ) ; } } newField = Field . create ( sdcMapFieldValues ) ; } else if ( fieldDescriptor . isRepeated ( ) ) { // List entry ( repeated )
List < ? > list = ( List < ? > ) message ; List < Field > listField = new ArrayList < > ( ) ; for ( int i = 0 ; i < list . size ( ) ; i ++ ) { if ( fieldDescriptor . getJavaType ( ) == Descriptors . FieldDescriptor . JavaType . MESSAGE ) { listField . add ( protobufToSdcField ( record , fieldPath + "[" + i + "]" , fieldDescriptor . getMessageType ( ) , messageTypeToExtensionMap , list . get ( i ) ) ) ; } else { listField . add ( createSdcField ( record , fieldPath + "[" + i + "]" , fieldDescriptor , messageTypeToExtensionMap , list . get ( i ) ) ) ; } } newField = Field . create ( listField ) ; } else { // normal entry
newField = createSdcField ( record , fieldPath , fieldDescriptor , messageTypeToExtensionMap , message ) ; } return newField ;


signature:
public class ProtobufTypeUtil { /** * Serializes a record to a protobuf message using the specified descriptor .
 * @ param record Record to serialize
 * @ param desc Protobuf descriptor
 * @ param messageTypeToExtensionMap Protobuf extension map
 * @ param defaultValueMap Protobuf default field values
 * @ return serialized message
 * @ throws DataGeneratorException */
public static DynamicMessage sdcFieldToProtobufMsg ( Record record , Descriptors . Descriptor desc , Map < String , Set < Descriptors . FieldDescriptor > > messageTypeToExtensionMap , Map < String , Object > defaultValueMap ) throws DataGeneratorException { } }

implementation: return sdcFieldToProtobufMsg ( record , record . get ( ) , "" , desc , messageTypeToExtensionMap , defaultValueMap ) ;


signature:
public class ProtobufTypeUtil { /** * Serializes a field path in a record to a protobuf message using the specified descriptor .
 * @ param record Record with the field to serialize
 * @ param field The field to serialize
 * @ param fieldPath The field path of the specified field
 * @ param desc Protobuf descriptor
 * @ param messageTypeToExtensionMap Protobuf extension map
 * @ param defaultValueMap Protobuf default field values
 * @ return serialized message
 * @ throws DataGeneratorException */
private static DynamicMessage sdcFieldToProtobufMsg ( Record record , Field field , String fieldPath , Descriptors . Descriptor desc , Map < String , Set < Descriptors . FieldDescriptor > > messageTypeToExtensionMap , Map < String , Object > defaultValueMap ) throws DataGeneratorException { } }

implementation:
if ( field == null ) { return null ; } // compute all fields to look for including extensions
DynamicMessage . Builder builder = DynamicMessage . newBuilder ( desc ) ; List < Descriptors . FieldDescriptor > fields = new ArrayList < > ( ) ; fields . addAll ( desc . getFields ( ) ) ; if ( messageTypeToExtensionMap . containsKey ( desc . getFullName ( ) ) ) { fields . addAll ( messageTypeToExtensionMap . get ( desc . getFullName ( ) ) ) ; } // root field is always a Map in a record representing protobuf data
Map < String , Field > valueAsMap = field . getValueAsMap ( ) ; for ( Descriptors . FieldDescriptor f : fields ) { Field mapField = valueAsMap . get ( f . getName ( ) ) ; // Repeated field
if ( f . isMapField ( ) ) { handleMapField ( record , mapField , fieldPath , messageTypeToExtensionMap , defaultValueMap , f , builder ) ; } else if ( f . isRepeated ( ) ) { if ( mapField != null ) { handleRepeatedField ( record , mapField , fieldPath , messageTypeToExtensionMap , defaultValueMap , f , builder ) ; } } else { // non repeated field
handleNonRepeatedField ( record , valueAsMap , fieldPath , messageTypeToExtensionMap , defaultValueMap , desc , f , builder ) ; } } // if record has unknown fields for this field path , handle it
try { handleUnknownFields ( record , fieldPath , builder ) ; } catch ( IOException e ) { throw new DataGeneratorException ( Errors . PROTOBUF_05 , e . toString ( ) , e ) ; } return builder . build ( ) ;


signature:
public class JerseyClientUtil { /** * Helper method to upgrade both HTTP stages to the JerseyConfigBean
 * @ param configs List of configs to upgrade . */
public static void upgradeToJerseyConfigBean ( List < Config > configs ) { } }

implementation: List < Config > configsToAdd = new ArrayList < > ( ) ; List < Config > configsToRemove = new ArrayList < > ( ) ; List < String > movedConfigs = ImmutableList . of ( "conf.requestTimeoutMillis" , "conf.numThreads" , "conf.authType" , "conf.oauth" , "conf.basicAuth" , "conf.useProxy" , "conf.proxy" , "conf.sslConfig" ) ; for ( Config config : configs ) { if ( hasPrefixIn ( movedConfigs , config . getName ( ) ) ) { configsToRemove . add ( config ) ; configsToAdd . add ( new Config ( config . getName ( ) . replace ( "conf." , "conf.client." ) , config . getValue ( ) ) ) ; } } configsToAdd . add ( new Config ( "conf.client.transferEncoding" , RequestEntityProcessing . CHUNKED ) ) ; configs . removeAll ( configsToRemove ) ; configs . addAll ( configsToAdd ) ;


signature:
public class KuduUtils { /** * Check network connection to the kudu master .
 * @ param kuduClient AsyncKuduClient
 * @ param context
 * @ param KUDU _ MASTER
 * @ param issues */
public static void checkConnection ( AsyncKuduClient kuduClient , Context context , String KUDU_MASTER , final List < Stage . ConfigIssue > issues ) { } }

implementation: try { kuduClient . getTablesList ( ) . join ( ) ; } catch ( Exception ex ) { issues . add ( context . createConfigIssue ( Groups . KUDU . name ( ) , KuduLookupConfig . CONF_PREFIX + KUDU_MASTER , Errors . KUDU_00 , ex . toString ( ) , ex ) ) ; }


signature:
public class KuduUtils { /** * Convert from Kudu type to SDC Field type
 * @ param kuduType
 * @ return Field . Type */
public static Field . Type convertFromKuduType ( Type kuduType ) { } }

implementation: switch ( kuduType ) { case BINARY : return Field . Type . BYTE_ARRAY ; case BOOL : return Field . Type . BOOLEAN ; case DOUBLE : return Field . Type . DOUBLE ; case FLOAT : return Field . Type . FLOAT ; case INT8 : return Field . Type . BYTE ; case INT16 : return Field . Type . SHORT ; case INT32 : return Field . Type . INTEGER ; case INT64 : return Field . Type . LONG ; case STRING : return Field . Type . STRING ; case UNIXTIME_MICROS : return Field . Type . DATETIME ; default : if ( "DECIMAL" . equals ( kuduType . name ( ) ) ) { return Field . Type . DECIMAL ; } throw new UnsupportedOperationException ( "Unknown data type: " + kuduType . getName ( ) ) ; }


signature:
public class KuduUtils { /** * Create a field and assign a value off of RowResult .
 * @ param result Result obtained from scan
 * @ param fieldName Field name to create
 * @ param type Kudu Type for the field
 * @ return Generated field
 * @ throws StageException */
public static Field createField ( RowResult result , String fieldName , Type type ) throws StageException { } }

implementation:
switch ( type ) { case INT8 : return Field . create ( Field . Type . BYTE , result . getByte ( fieldName ) ) ; case INT16 : return Field . create ( Field . Type . SHORT , result . getShort ( fieldName ) ) ; case INT32 : return Field . create ( Field . Type . INTEGER , result . getInt ( fieldName ) ) ; case INT64 : return Field . create ( Field . Type . LONG , result . getLong ( fieldName ) ) ; case BINARY : try { return Field . create ( Field . Type . BYTE_ARRAY , result . getBinary ( fieldName ) ) ; } catch ( IllegalArgumentException ex ) { throw new OnRecordErrorException ( Errors . KUDU_35 , fieldName ) ; } case STRING : return Field . create ( Field . Type . STRING , result . getString ( fieldName ) ) ; case BOOL : return Field . create ( Field . Type . BOOLEAN , result . getBoolean ( fieldName ) ) ; case FLOAT : return Field . create ( Field . Type . FLOAT , result . getFloat ( fieldName ) ) ; case DOUBLE : return Field . create ( Field . Type . DOUBLE , result . getDouble ( fieldName ) ) ; case UNIXTIME_MICROS : // UNIXTIME _ MICROS is in microsecond
return Field . create ( Field . Type . DATETIME , new Date ( result . getLong ( fieldName ) / 1000L ) ) ; default : if ( "DECIMAL" . equals ( type . name ( ) ) ) { return Field . create ( Field . Type . DECIMAL , result . getDecimal ( fieldName ) ) ; } throw new StageException ( Errors . KUDU_10 , fieldName , type . getName ( ) ) ; }


signature:
public class FullPipeBatch { /** * Intercept given records with all the interceptors .
 * We ' re not cloning records during interception as we aim at changing their original form . */
private List < Record > intercept ( List < Record > records , List < ? extends Interceptor > interceptors ) throws StageException { } }

implementation: for ( Interceptor interceptor : interceptors ) { records = interceptor . intercept ( records ) ; } return records ;


signature:
public class FileLine { /** * Returns the text of the line .
 * @ return the text of the line . */
public String getText ( ) { } }

implementation: if ( line == null ) { line = new String ( buffer , offsetInChunk , length , charset ) ; } return line ;


signature:
public class ProtoRunner { /** * Configuration */
private boolean isConfigurationActive ( ConfigDef configDef , Map < String , Object > configuration ) { } }

implementation: String dependsOn = configDef . dependsOn ( ) ; if ( ! dependsOn . isEmpty ( ) ) { Object dependsOnValue = configuration . get ( dependsOn ) ; if ( dependsOnValue != null ) { String valueStr = dependsOnValue . toString ( ) ; for ( String trigger : configDef . triggeredByValue ( ) ) { if ( valueStr . equals ( trigger ) ) { return true ; } } return false ; } return false ; } return true ;


signature:
public class ValidationUtil { /** * Resolve stage aliases ( e . g . when a stage is renamed ) . */
public static void resolveStageAlias ( StageLibraryTask stageLibrary , StageConfiguration stageConf ) { } }

implementation: String aliasKey = Joiner . on ( "," ) . join ( stageConf . getLibrary ( ) , stageConf . getStageName ( ) ) ; String aliasValue = Strings . nullToEmpty ( stageLibrary . getStageNameAliases ( ) . get ( aliasKey ) ) ; if ( LOG . isTraceEnabled ( ) ) { for ( String key : stageLibrary . getStageNameAliases ( ) . keySet ( ) ) { LOG . trace ( "Stage Lib Alias: {} => {}" , key , stageLibrary . getStageNameAliases ( ) . get ( key ) ) ; } LOG . trace ( "Looking for '{}' and found '{}'" , aliasKey , aliasValue ) ; } if ( ! aliasValue . isEmpty ( ) ) { List < String > alias = Splitter . on ( "," ) . splitToList ( aliasValue ) ; if ( alias . size ( ) == 2 ) { LOG . debug ( "Converting '{}' to '{}'" , aliasKey , aliasValue ) ; stageConf . setLibrary ( alias . get ( 0 ) ) ; stageConf . setStageName ( alias . get ( 1 ) ) ; } else { LOG . error ( "Malformed stage alias: '{}'" , aliasValue ) ; } }


signature:
public class ValidationUtil { /** * Resolve all stage library relevant aliases - this includes :
 * * Change in stage library name - e . g . when a stage moves from let say cdh5 _ cluster _ 2 to cdh _ 5 _ spark2)
 * * Change in stage name - e . g . when a stage class is renamed */
public static boolean resolveLibraryAliases ( StageLibraryTask stageLibrary , List < StageConfiguration > stageConfigurations ) { } }

implementation: for ( StageConfiguration stageConf : stageConfigurations ) { String name = stageConf . getLibrary ( ) ; if ( stageLibrary . getLibraryNameAliases ( ) . containsKey ( name ) ) { stageConf . setLibrary ( stageLibrary . getLibraryNameAliases ( ) . get ( name ) ) ; } resolveStageAlias ( stageLibrary , stageConf ) ; } return true ;


signature:
public class ValidationUtil { /** * Add any missing configs to the stage configuration . */
public static void addMissingConfigsToStage ( StageLibraryTask stageLibrary , StageConfiguration stageConf ) { } }

implementation: StageDefinition stageDef = stageLibrary . getStage ( stageConf . getLibrary ( ) , stageConf . getStageName ( ) , false ) ; if ( stageDef != null ) { for ( ConfigDefinition configDef : stageDef . getConfigDefinitions ( ) ) { String configName = configDef . getName ( ) ; Config config = stageConf . getConfig ( configName ) ; if ( config == null ) { Object defaultValue = configDef . getDefaultValue ( ) ; LOG . warn ( "Stage '{}' missing configuration '{}', adding with '{}' as default" , stageConf . getInstanceName ( ) , configName , defaultValue ) ; config = new Config ( configName , defaultValue ) ; stageConf . addConfig ( config ) ; } } }


signature:
public class ValidationUtil { /** * Validate given stage configuration . */
public static boolean validateStageConfiguration ( StageLibraryTask stageLibrary , boolean shouldBeSource , StageConfiguration stageConf , boolean notOnMainCanvas , IssueCreator issueCreator , boolean isPipelineFragment , Map < String , Object > constants , List < Issue > issues ) { } }

implementation:
boolean preview = true ; StageDefinition stageDef = stageLibrary . getStage ( stageConf . getLibrary ( ) , stageConf . getStageName ( ) , false ) ; if ( stageDef == null ) { // stage configuration refers to an undefined stage definition
issues . add ( issueCreator . create ( stageConf . getInstanceName ( ) , ValidationError . VALIDATION_0006 , stageConf . getLibrary ( ) , stageConf . getStageName ( ) , stageConf . getStageVersion ( ) ) ) ; preview = false ; } else { if ( shouldBeSource ) { if ( stageDef . getType ( ) != StageType . SOURCE && ! isPipelineFragment ) { // first stage must be a Source
issues . add ( issueCreator . create ( stageConf . getInstanceName ( ) , ValidationError . VALIDATION_0003 ) ) ; preview = false ; } } else { if ( ! stageLibrary . isMultipleOriginSupported ( ) && stageDef . getType ( ) == StageType . SOURCE ) { // no stage other than first stage can be a Source
issues . add ( issueCreator . create ( stageConf . getInstanceName ( ) , ValidationError . VALIDATION_0004 ) ) ; preview = false ; } } if ( ! stageConf . isSystemGenerated ( ) && ! TextUtils . isValidName ( stageConf . getInstanceName ( ) ) ) { // stage instance name has an invalid name ( it must match ' [ 0-9A - Za - z _ ] + ' )
issues . add ( issueCreator . create ( stageConf . getInstanceName ( ) , ValidationError . VALIDATION_0016 , stageConf . getInstanceName ( ) , TextUtils . VALID_NAME ) ) ; preview = false ; } // Hidden stages can ' t appear on the main canvas
if ( ! notOnMainCanvas && ! stageDef . getHideStage ( ) . isEmpty ( ) ) { issues . add ( issueCreator . create ( stageConf . getInstanceName ( ) , ValidationError . VALIDATION_0037 ) ) ; preview = false ; } for ( String lane : stageConf . getInputLanes ( ) ) { if ( ! TextUtils . isValidName ( lane ) ) { // stage instance input lane has an invalid name ( it must match ' [ 0-9A - Za - z _ ] + ' )
issues . add ( issueCreator . create ( stageConf . getInstanceName ( ) , ValidationError . VALIDATION_0017 , lane , TextUtils . VALID_NAME ) ) ; preview = false ; } } for ( String lane : stageConf . getOutputLanes ( ) ) { if ( ! TextUtils . isValidName ( lane ) ) { // stage instance output lane has an invalid name ( it must match ' [ 0-9A - Za - z _ ] + ' )
issues . add ( issueCreator . create ( stageConf . getInstanceName ( ) , ValidationError . VALIDATION_0018 , lane , TextUtils . VALID_NAME ) ) ; preview = false ; } } for ( String lane : stageConf . getEventLanes ( ) ) { if ( ! TextUtils . isValidName ( lane ) ) { // stage instance output lane has an invalid name ( it must match ' [ 0-9A - Za - z _ ] + ' )
issues . add ( issueCreator . create ( stageConf . getInstanceName ( ) , ValidationError . VALIDATION_0100 , lane , TextUtils . VALID_NAME ) ) ; preview = false ; } } // Special validation for stage exposed limit of input lanes
// -1 : Means that framework is fully in control on how many input lanes should be present
// 0 : Means that the stage supports unlimited number of input lanes
// > 0 : Means that stage needs exactly that amount of lanes which we will validate here
if ( stageDef . getInputStreams ( ) > 0 ) { if ( stageDef . getInputStreams ( ) != stageConf . getInputLanes ( ) . size ( ) ) { issues . add ( issueCreator . create ( stageConf . getInstanceName ( ) , ValidationError . VALIDATION_0094 , stageDef . getInputStreams ( ) , stageConf . getInputLanes ( ) . size ( ) ) ) ; } } // Validate proper input / output lane configuration
switch ( stageDef . getType ( ) ) { case SOURCE : if ( ! stageConf . getInputLanes ( ) . isEmpty ( ) ) { // source stage cannot have input lanes
issues . add ( issueCreator . create ( stageConf . getInstanceName ( ) , ValidationError . VALIDATION_0012 , stageDef . getType ( ) , stageConf . getInputLanes ( ) ) ) ; preview = false ; } if ( ! notOnMainCanvas && ! stageDef . isVariableOutputStreams ( ) ) { // source stage must match the output stream defined in StageDef
if ( stageDef . getOutputStreams ( ) != stageConf . getOutputLanes ( ) . size ( ) ) { issues . add ( issueCreator . create ( stageConf . getInstanceName ( ) , ValidationError . VALIDATION_0015 , stageDef . getOutputStreams ( ) , stageConf . getOutputLanes ( ) . size ( ) ) ) ; } } else if ( ! notOnMainCanvas && stageConf . getOutputLanes ( ) . isEmpty ( ) ) { // source stage must have at least one output lane
issues . add ( issueCreator . create ( stageConf . getInstanceName ( ) , ValidationError . VALIDATION_0032 ) ) ; } break ; case PROCESSOR : if ( ! notOnMainCanvas && stageConf . getInputLanes ( ) . isEmpty ( ) && ! isPipelineFragment ) { // processor stage must have at least one input lane
issues . add ( issueCreator . create ( stageConf . getInstanceName ( ) , ValidationError . VALIDATION_0014 , "Processor" ) ) ; preview = false ; } if ( ! notOnMainCanvas ) { if ( ! stageDef . isVariableOutputStreams ( ) ) { // processor stage must match the output stream defined in StageDef
if ( stageDef . getOutputStreams ( ) != stageConf . getOutputLanes ( ) . size ( ) ) { issues . add ( issueCreator . create ( stageConf . getInstanceName ( ) , ValidationError . VALIDATION_0015 , stageDef . getOutputStreams ( ) , stageConf . getOutputLanes ( ) . size ( ) ) ) ; } } else if ( stageConf . getOutputLanes ( ) . isEmpty ( ) ) { // processor stage must have at least one output lane
issues . add ( issueCreator . create ( stageConf . getInstanceName ( ) , ValidationError . VALIDATION_0032 ) ) ; } } break ; case EXECUTOR : case TARGET : // Normal target stage must have at least one input lane
if ( ! notOnMainCanvas && stageConf . getInputLanes ( ) . isEmpty ( ) && ! isPipelineFragment ) { issues . add ( issueCreator . create ( stageConf . getInstanceName ( ) , ValidationError . VALIDATION_0014 , "Target" ) ) ; preview = false ; } // Error / Stats / Pipeline lifecycle must not have an input lane
if ( notOnMainCanvas && ! stageConf . getInputLanes ( ) . isEmpty ( ) ) { issues . add ( issueCreator . create ( stageConf . getInstanceName ( ) , ValidationError . VALIDATION_0012 , "Error/Stats/Lifecycle" , stageConf . getInputLanes ( ) ) ) ; preview = false ; } if ( ! stageConf . getOutputLanes ( ) . isEmpty ( ) ) { // target stage cannot have output lanes
issues . add ( issueCreator . create ( stageConf . getInstanceName ( ) , ValidationError . VALIDATION_0013 , stageDef . getType ( ) , stageConf . getOutputLanes ( ) ) ) ; preview = false ; } if ( notOnMainCanvas && ! stageConf . getEventLanes ( ) . isEmpty ( ) ) { issues . add ( issueCreator . create ( stageConf . getInstanceName ( ) , ValidationError . VALIDATION_0036 , stageDef . getType ( ) , stageConf . getEventLanes ( ) ) ) ; preview = false ; } break ; default : throw new IllegalStateException ( "Unexpected stage type " + stageDef . getType ( ) ) ; } // Validate proper event configuration
if ( ! notOnMainCanvas && stageConf . getEventLanes ( ) . size ( ) > 1 ) { issues . add ( issueCreator . create ( stageConf . getInstanceName ( ) , ValidationError . VALIDATION_0101 ) ) ; preview = false ; } if ( ! notOnMainCanvas && ! stageDef . isProducingEvents ( ) && stageConf . getEventLanes ( ) . size ( ) > 0 ) { issues . add ( issueCreator . create ( stageConf . getInstanceName ( ) , ValidationError . VALIDATION_0102 ) ) ; preview = false ; } // Validate stage owns configuration
preview &= validateComponentConfigs ( stageConf , stageDef . getConfigDefinitions ( ) , stageDef . getConfigDefinitionsMap ( ) , stageDef . getHideConfigs ( ) , stageDef . hasPreconditions ( ) , constants , issueCreator , issues ) ; // Validate service definitions
Set < String > expectedServices = stageDef . getServices ( ) . stream ( ) . map ( service -> service . getService ( ) . getName ( ) ) . collect ( Collectors . toSet ( ) ) ; Set < String > configuredServices = stageConf . getServices ( ) . stream ( ) . map ( service -> service . getService ( ) . getName ( ) ) . collect ( Collectors . toSet ( ) ) ; if ( ! expectedServices . equals ( configuredServices ) ) { issues . add ( issueCreator . create ( stageConf . getInstanceName ( ) , ValidationError . VALIDATION_0200 , StringUtils . join ( expectedServices , "," ) , StringUtils . join ( configuredServices , "," ) ) ) ; preview = false ; } else { // Validate all services
for ( ServiceConfiguration serviceConf : stageConf . getServices ( ) ) { ServiceDefinition serviceDef = stageLibrary . getServiceDefinition ( serviceConf . getService ( ) , false ) ; preview &= validateComponentConfigs ( serviceConf , serviceDef . getConfigDefinitions ( ) , serviceDef . getConfigDefinitionsMap ( ) , Collections . emptySet ( ) , false , constants , issueCreator . forService ( serviceConf . getService ( ) . getName ( ) ) , issues ) ; } } } return preview ;


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

implementation:
List < ConfigIssue > issues = super . init ( ) ; errorRecordHandler = new DefaultErrorRecordHandler ( getContext ( ) ) ; // NOSONAR
double rateLimit = conf . rateLimit > 0 ? ( 1000.0 / conf . rateLimit ) : Double . MAX_VALUE ; rateLimiter = RateLimiter . create ( rateLimit ) ; httpClientCommon . init ( issues , getContext ( ) ) ; conf . dataFormatConfig . init ( getContext ( ) , conf . dataFormat , Groups . HTTP . name ( ) , HttpClientCommon . DATA_FORMAT_CONFIG_PREFIX , issues ) ; bodyVars = getContext ( ) . createELVars ( ) ; bodyEval = getContext ( ) . createELEval ( REQUEST_BODY_CONFIG_NAME ) ; if ( issues . isEmpty ( ) ) { parserFactory = conf . dataFormatConfig . getParserFactory ( ) ; } return issues ;


signature:
public class HttpProcessor { /** * { @ inheritDoc } */
@ Override public void process ( Batch batch , SingleLaneBatchMaker batchMaker ) throws StageException { } }

implementation:
List < Future < Response > > responses = new ArrayList < > ( ) ; resolvedRecords . clear ( ) ; Iterator < Record > records = batch . getRecords ( ) ; while ( records . hasNext ( ) ) { Record record = records . next ( ) ; String resolvedUrl = httpClientCommon . getResolvedUrl ( conf . resourceUrl , record ) ; WebTarget target = httpClientCommon . getClient ( ) . target ( resolvedUrl ) ; LOG . debug ( "Resolved HTTP Client URL: '{}'" , resolvedUrl ) ; // If the request ( headers or body ) contain a known sensitive EL and we ' re not using https then fail the request .
if ( httpClientCommon . requestContainsSensitiveInfo ( conf . headers , conf . requestBody ) && ! target . getUri ( ) . getScheme ( ) . toLowerCase ( ) . startsWith ( "https" ) ) { throw new StageException ( Errors . HTTP_07 ) ; } // from HttpStreamConsumer
final MultivaluedMap < String , Object > resolvedHeaders = httpClientCommon . resolveHeaders ( conf . headers , record ) ; String contentType = HttpStageUtil . getContentTypeWithDefault ( resolvedHeaders , conf . defaultRequestContentType ) ; final AsyncInvoker asyncInvoker = target . request ( ) . property ( OAuth1ClientSupport . OAUTH_PROPERTY_ACCESS_TOKEN , httpClientCommon . getAuthToken ( ) ) . headers ( resolvedHeaders ) . async ( ) ; HttpMethod method = httpClientCommon . getHttpMethod ( conf . httpMethod , conf . methodExpression , record ) ; rateLimiter . acquire ( ) ; if ( conf . requestBody != null && ! conf . requestBody . isEmpty ( ) && method != HttpMethod . GET ) { RecordEL . setRecordInContext ( bodyVars , record ) ; final String requestBody = bodyEval . eval ( bodyVars , conf . requestBody , String . class ) ; resolvedRecords . put ( record , new HeadersAndBody ( resolvedHeaders , requestBody , contentType , method , target ) ) ; responses . add ( asyncInvoker . method ( method . getLabel ( ) , Entity . entity ( requestBody , contentType ) ) ) ; } else { resolvedRecords . put ( record , new HeadersAndBody ( resolvedHeaders , null , null , method , target ) ) ; responses . add ( asyncInvoker . method ( method . getLabel ( ) ) ) ; } } records = batch . getRecords ( ) ; int recordNum = 0 ; while ( records . hasNext ( ) ) { try { Record record = processResponse ( records . next ( ) , responses . get ( recordNum ) , conf . maxRequestCompletionSecs , false ) ; if ( record != null ) { batchMaker . addRecord ( record ) ; } } catch ( OnRecordErrorException e ) { errorRecordHandler . onError ( e ) ; } finally { ++ recordNum ; } } if ( ! resolvedRecords . isEmpty ( ) ) { reprocessIfRequired ( batchMaker ) ; }


signature:
public class HttpProcessor { /** * 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 )
 * @ return parsed record from the request
 * @ throws StageException if the request fails , times out , or cannot be parsed */
private Record processResponse ( Record record , Future < Response > responseFuture , long maxRequestCompletionSecs , boolean failOn403 ) throws StageException { } }

implementation: Response response = null ; try { response = responseFuture . get ( maxRequestCompletionSecs , TimeUnit . SECONDS ) ; InputStream responseBody = null ; if ( response . hasEntity ( ) ) { responseBody = response . readEntity ( InputStream . class ) ; } int responseStatus = response . getStatus ( ) ; if ( conf . client . useOAuth2 && response . getStatus ( ) == 403 && ! failOn403 ) { HttpStageUtil . getNewOAuth2Token ( conf . client . oauth2 , httpClientCommon . getClient ( ) ) ; return null ; } else if ( responseStatus < 200 || responseStatus >= 300 ) { resolvedRecords . remove ( record ) ; throw new OnRecordErrorException ( record , Errors . HTTP_01 , response . getStatus ( ) , response . getStatusInfo ( ) . getReasonPhrase ( ) + " " + responseBody ) ; } resolvedRecords . remove ( record ) ; Record parsedResponse = parseResponse ( responseBody ) ; if ( parsedResponse != null ) { record . set ( conf . outputField , parsedResponse . get ( ) ) ; addResponseHeaders ( record , response ) ; } else if ( responseBody == null && responseStatus != 204 ) { throw new OnRecordErrorException ( record , Errors . HTTP_34 ) ; } return record ; } catch ( InterruptedException | ExecutionException e ) { LOG . error ( Errors . HTTP_03 . getMessage ( ) , e . toString ( ) , e ) ; throw new OnRecordErrorException ( record , Errors . HTTP_03 , e . toString ( ) ) ; } catch ( TimeoutException e ) { LOG . error ( "HTTP request future timed out" , e . toString ( ) , e ) ; throw new OnRecordErrorException ( record , Errors . HTTP_03 , e . toString ( ) ) ; } finally { if ( response != null ) { response . close ( ) ; } }


signature:
public class HttpProcessor { /** * Parses the HTTP response text from a request into SDC Records
 * @ param response HTTP response
 * @ return an SDC record resulting from the response text
 * @ throws StageException if the response could not be parsed */
private Record parseResponse ( InputStream response ) throws StageException { } }

implementation:
Record record = null ; if ( conf . httpMethod == HttpMethod . HEAD ) { // Head will have no body so can ' t be parsed . Return an empty record .
record = getContext ( ) . createRecord ( "" ) ; record . set ( Field . create ( new HashMap ( ) ) ) ; } else if ( response != null ) { try ( DataParser parser = parserFactory . getParser ( "" , response , "0" ) ) { // A response may only contain a single record , so we only parse it once .
record = parser . parse ( ) ; if ( conf . dataFormat == DataFormat . TEXT ) { // Output is placed in a field " / text " so we remove it here .
record . set ( record . get ( "/text" ) ) ; } } catch ( IOException | DataParserException e ) { errorRecordHandler . onError ( Errors . HTTP_00 , e . toString ( ) , e ) ; } } return record ;


signature:
public class HttpProcessor { /** * Populates HTTP response headers to the configured location
 * @ param record current record to populate
 * @ param response HTTP response
 * @ throws StageException when writing headers to a field path that already exists */
private void addResponseHeaders ( Record record , Response response ) throws StageException { } }

implementation: if ( conf . headerOutputLocation == HeaderOutputLocation . NONE ) { return ; } Record . Header header = record . getHeader ( ) ; if ( conf . headerOutputLocation == HeaderOutputLocation . FIELD ) { writeResponseHeaderToField ( record , response ) ; } else if ( conf . headerOutputLocation == HeaderOutputLocation . HEADER ) { writeResponseHeaderToRecordHeader ( response , header ) ; }


signature:
public class HttpProcessor { /** * Writes HTTP response headers to the SDC Record at the configured field path .
 * @ param record Record to populate with response headers .
 * @ param response HTTP response
 * @ throws StageException if the field path already exists */
private void writeResponseHeaderToField ( Record record , Response response ) throws StageException { } }

implementation: if ( record . has ( conf . headerOutputField ) ) { throw new StageException ( Errors . HTTP_11 , conf . headerOutputField ) ; } Map < String , Field > headers = new HashMap < > ( response . getStringHeaders ( ) . size ( ) ) ; for ( Map . Entry < String , List < String > > entry : response . getStringHeaders ( ) . entrySet ( ) ) { if ( ! entry . getValue ( ) . isEmpty ( ) ) { String firstValue = entry . getValue ( ) . get ( 0 ) ; headers . put ( entry . getKey ( ) , Field . create ( firstValue ) ) ; } } record . set ( conf . headerOutputField , Field . create ( headers ) ) ;


signature:
public class HttpProcessor { /** * Writes HTTP response headers to the SDC Record header with the configured optional prefix .
 * @ param response HTTP response
 * @ param header SDC Record header */
private void writeResponseHeaderToRecordHeader ( Response response , Record . Header header ) { } }

implementation: for ( Map . Entry < String , List < String > > entry : response . getStringHeaders ( ) . entrySet ( ) ) { if ( ! entry . getValue ( ) . isEmpty ( ) ) { String firstValue = entry . getValue ( ) . get ( 0 ) ; header . setAttribute ( conf . headerAttributePrefix + entry . getKey ( ) , firstValue ) ; } }


signature:
public class BlobStoreTaskImpl { /** * Commit metadata content to a file to disk .
 * This method does three - phased commit :
 * 1 ) New content is written into a new temporary file .
 * 2 ) Old metadata is dropped
 * 3 ) Rename from new to old is done */
synchronized private void saveMetadata ( ) throws StageException { } }

implementation:
// 0 ) Validate pre - conditions
if ( Files . exists ( newMetadataFile ) ) { throw new StageException ( BlobStoreError . BLOB_STORE_0010 ) ; } // 1 ) New content is written into a new temporary file .
try ( OutputStream os = Files . newOutputStream ( newMetadataFile , StandardOpenOption . CREATE , StandardOpenOption . TRUNCATE_EXISTING ) ) { jsonMapper . writeValue ( os , metadata ) ; } catch ( IOException e ) { throw new StageException ( BlobStoreError . BLOB_STORE_0001 , e . toString ( ) , e ) ; } // 2 ) Old metadata is dropped
try { if ( Files . exists ( metadataFile ) ) { Files . delete ( metadataFile ) ; } } catch ( IOException e ) { throw new StageException ( BlobStoreError . BLOB_STORE_0011 , e . toString ( ) , e ) ; } // 3 ) Rename from new to old is done
try { Files . move ( newMetadataFile , metadataFile ) ; } catch ( IOException e ) { throw new StageException ( BlobStoreError . BLOB_STORE_0012 , e . toString ( ) , e ) ; }


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

implementation:
List < ConfigIssue > issues = super . init ( ) ; errorRecordHandler = new DefaultErrorRecordHandler ( getContext ( ) ) ; // NOSONAR
conf . basic . init ( getContext ( ) , Groups . HTTP . name ( ) , BASIC_CONFIG_PREFIX , issues ) ; conf . dataFormatConfig . init ( getContext ( ) , conf . dataFormat , Groups . HTTP . name ( ) , DATA_FORMAT_CONFIG_PREFIX , issues ) ; conf . init ( getContext ( ) , Groups . HTTP . name ( ) , "conf." , issues ) ; if ( conf . client . tlsConfig . isEnabled ( ) ) { conf . client . tlsConfig . init ( getContext ( ) , Groups . TLS . name ( ) , TLS_CONFIG_PREFIX , issues ) ; } resourceVars = getContext ( ) . createELVars ( ) ; resourceEval = getContext ( ) . createELEval ( RESOURCE_CONFIG_NAME ) ; bodyVars = getContext ( ) . createELVars ( ) ; bodyEval = getContext ( ) . createELEval ( REQUEST_BODY_CONFIG_NAME ) ; Calendar calendar = Calendar . getInstance ( TimeZone . getTimeZone ( ZoneId . of ( conf . timeZoneID ) ) ) ; TimeEL . setCalendarInContext ( bodyVars , calendar ) ; headerVars = getContext ( ) . createELVars ( ) ; headerEval = getContext ( ) . createELEval ( HEADER_CONFIG_NAME ) ; stopVars = getContext ( ) . createELVars ( ) ; stopEval = getContext ( ) . createELEval ( STOP_CONFIG_NAME ) ; next = null ; haveMorePages = false ; if ( conf . responseStatusActionConfigs != null ) { final String cfgName = "conf.responseStatusActionConfigs" ; final EnumSet < ResponseAction > backoffRetries = EnumSet . of ( ResponseAction . RETRY_EXPONENTIAL_BACKOFF , ResponseAction . RETRY_LINEAR_BACKOFF ) ; for ( HttpResponseActionConfigBean actionConfig : conf . responseStatusActionConfigs ) { final HttpResponseActionConfigBean prevAction = statusToActionConfigs . put ( actionConfig . getStatusCode ( ) , actionConfig ) ; if ( prevAction != null ) { issues . add ( getContext ( ) . createConfigIssue ( Groups . HTTP . name ( ) , cfgName , Errors . HTTP_17 , actionConfig . getStatusCode ( ) ) ) ; } if ( backoffRetries . contains ( actionConfig . getAction ( ) ) && actionConfig . getBackoffInterval ( ) <= 0 ) { issues . add ( getContext ( ) . createConfigIssue ( Groups . HTTP . name ( ) , cfgName , Errors . HTTP_15 ) ) ; } if ( actionConfig . getStatusCode ( ) >= 200 && actionConfig . getStatusCode ( ) < 300 ) { issues . add ( getContext ( ) . createConfigIssue ( Groups . HTTP . name ( ) , cfgName , Errors . HTTP_16 ) ) ; } } } this . timeoutActionConfig = conf . responseTimeoutActionConfig ; // Validation succeeded so configure the client .
if ( issues . isEmpty ( ) ) { try { configureClient ( issues ) ; } catch ( StageException e ) { // should not happen on initial connect
ExceptionUtils . throwUndeclared ( e ) ; } } return issues ;


signature:
public class HttpClientSource { /** * Helper method to apply Jersey client configuration properties . */
private void configureClient ( List < ConfigIssue > issues ) throws StageException { } }

implementation: clientCommon . init ( issues , getContext ( ) ) ; if ( issues . isEmpty ( ) ) { client = clientCommon . getClient ( ) ; parserFactory = conf . dataFormatConfig . getParserFactory ( ) ; }


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

implementation:
long start = System . currentTimeMillis ( ) ; int chunksToFetch = Math . min ( conf . basic . maxBatchSize , maxBatchSize ) ; Optional < String > newSourceOffset = Optional . empty ( ) ; recordCount = 0 ; setPageOffset ( lastSourceOffset ) ; setResolvedUrl ( resolveInitialUrl ( lastSourceOffset ) ) ; WebTarget target = client . target ( getResolvedUrl ( ) ) ; // If the request ( headers or body ) contain a known sensitive EL and we ' re not using https then fail the request .
if ( requestContainsSensitiveInfo ( ) && ! target . getUri ( ) . getScheme ( ) . toLowerCase ( ) . startsWith ( "https" ) ) { LOG . error ( Errors . HTTP_07 . getMessage ( ) ) ; throw new StageException ( Errors . HTTP_07 ) ; } boolean uninterrupted = true ; while ( ! waitTimeExpired ( start ) && uninterrupted && ( recordCount < chunksToFetch ) ) { if ( parser != null ) { // We already have an response that we haven ' t finished reading .
newSourceOffset = Optional . of ( parseResponse ( start , chunksToFetch , batchMaker ) ) ; } else if ( shouldMakeRequest ( ) ) { if ( conf . pagination . mode != PaginationMode . NONE ) { target = client . target ( resolveNextPageUrl ( newSourceOffset . orElse ( null ) ) ) ; // Pause between paging requests so we don ' t get rate limited .
uninterrupted = ThreadUtil . sleep ( conf . pagination . rateLimit ) ; } makeRequest ( target ) ; if ( lastRequestTimedOut ) { String actionName = conf . responseTimeoutActionConfig . getAction ( ) . name ( ) ; LOG . warn ( "HTTPClient timed out after waiting {} ms for response from server;" + " reconnecting client and proceeding as per configured {} action" , conf . client . readTimeoutMillis , actionName ) ; reconnectClient ( ) ; return nonTerminating ( lastSourceOffset ) ; } else { newSourceOffset = processResponse ( start , chunksToFetch , batchMaker ) ; } } else if ( conf . httpMode == HttpClientMode . BATCH ) { // We are done .
return null ; } else { // In polling mode , waiting for the next polling interval .
uninterrupted = ThreadUtil . sleep ( SLEEP_TIME_WAITING_FOR_BATCH_SIZE_MS ) ; } } return newSourceOffset . orElse ( lastSourceOffset ) ;


signature:
public class HttpClientSource { /** * Returns the URL of the next page to fetch when paging is enabled . Otherwise
 * returns the previously configured URL .
 * @ param sourceOffset current source offset
 * @ return next URL to fetch
 * @ throws ELEvalException if the resource expression cannot be evaluated */
@ VisibleForTesting String resolveNextPageUrl ( String sourceOffset ) throws ELEvalException { } }

implementation: String url ; if ( LINK_PAGINATION . contains ( conf . pagination . mode ) && next != null ) { url = next . getUri ( ) . toString ( ) ; setResolvedUrl ( url ) ; } else if ( conf . pagination . mode == PaginationMode . BY_OFFSET || conf . pagination . mode == PaginationMode . BY_PAGE ) { if ( sourceOffset != null ) { setPageOffset ( sourceOffset ) ; } url = resourceEval . eval ( resourceVars , conf . resourceUrl , String . class ) ; } else { url = getResolvedUrl ( ) ; } return url ;


signature:
public class HttpClientSource { /** * Sets the startAt EL variable in scope for the resource and request body .
 * If the source offset is null ( origin was reset ) then the initial value
 * from the user provided configuration is used .
 * @ param sourceOffset source offset to parse for startAt variable . */
private void setPageOffset ( String sourceOffset ) { } }

implementation: if ( conf . pagination . mode == PaginationMode . NONE ) { return ; } int startAt = conf . pagination . startAt ; if ( StringUtils . isNotEmpty ( sourceOffset ) ) { startAt = HttpSourceOffset . fromString ( sourceOffset ) . getStartAt ( ) ; } resourceVars . addVariable ( START_AT , startAt ) ; bodyVars . addVariable ( START_AT , startAt ) ;


signature:
public class HttpClientSource { /** * Helper method to construct an HTTP request and fetch a response .
 * @ param target the target url to fetch .
 * @ throws StageException if an unhandled error is encountered */
private void makeRequest ( WebTarget target ) throws StageException { } }

implementation:
hasher = HF . newHasher ( ) ; MultivaluedMap < String , Object > resolvedHeaders = resolveHeaders ( ) ; final Invocation . Builder invocationBuilder = target . request ( ) . property ( OAuth1ClientSupport . OAUTH_PROPERTY_ACCESS_TOKEN , authToken ) . headers ( resolvedHeaders ) ; boolean keepRequesting = ! getContext ( ) . isStopped ( ) ; boolean gotNewToken = false ; while ( keepRequesting ) { long startTime = System . currentTimeMillis ( ) ; try { if ( conf . requestBody != null && ! conf . requestBody . isEmpty ( ) && conf . httpMethod != HttpMethod . GET ) { final String requestBody = bodyEval . eval ( bodyVars , conf . requestBody , String . class ) ; final String contentType = HttpStageUtil . getContentTypeWithDefault ( resolvedHeaders , conf . defaultRequestContentType ) ; hasher . putString ( requestBody , Charset . forName ( conf . dataFormatConfig . charset ) ) ; setResponse ( invocationBuilder . method ( conf . httpMethod . getLabel ( ) , Entity . entity ( requestBody , contentType ) ) ) ; } else { setResponse ( invocationBuilder . method ( conf . httpMethod . getLabel ( ) ) ) ; } LOG . debug ( "Retrieved response in {} ms" , System . currentTimeMillis ( ) - startTime ) ; lastRequestTimedOut = false ; final int status = response . getStatus ( ) ; final boolean statusOk = status >= 200 && status < 300 ; if ( conf . client . useOAuth2 && status == 403 ) { // Token may have expired
if ( gotNewToken ) { LOG . error ( HTTP_21 . getMessage ( ) ) ; throw new StageException ( HTTP_21 ) ; } gotNewToken = HttpStageUtil . getNewOAuth2Token ( conf . client . oauth2 , client ) ; } else if ( ! statusOk && this . statusToActionConfigs . containsKey ( status ) ) { final HttpResponseActionConfigBean actionConf = this . statusToActionConfigs . get ( status ) ; final boolean statusChanged = lastStatus != status || lastRequestTimedOut ; keepRequesting = applyResponseAction ( actionConf , statusChanged , input -> { final StageException stageException = new StageException ( Errors . HTTP_14 , status , response . readEntity ( String . class ) ) ; LOG . error ( stageException . getMessage ( ) ) ; return stageException ; } ) ; } else { keepRequesting = false ; retryCount = 0 ; } lastStatus = status ; } catch ( Exception e ) { LOG . debug ( "Request failed after {} ms" , System . currentTimeMillis ( ) - startTime ) ; final Throwable cause = e . getCause ( ) ; if ( cause != null && ( cause instanceof TimeoutException || cause instanceof SocketTimeoutException ) ) { LOG . warn ( "{} attempting to read response in HttpClientSource: {}" , cause . getClass ( ) . getSimpleName ( ) , e . getMessage ( ) , e ) ; // read timeout ; consult configured action to decide on backoff and retry strategy
if ( this . timeoutActionConfig != null ) { final HttpResponseActionConfigBean actionConf = this . timeoutActionConfig ; final boolean firstTimeout = ! lastRequestTimedOut ; applyResponseAction ( actionConf , firstTimeout , input -> { LOG . error ( Errors . HTTP_18 . getMessage ( ) ) ; return new StageException ( Errors . HTTP_18 ) ; } ) ; } lastRequestTimedOut = true ; keepRequesting = false ; } else if ( cause != null && cause instanceof InterruptedException ) { LOG . error ( String . format ( "InterruptedException attempting to make request in HttpClientSource; stopping: %s" , e . getMessage ( ) ) , e ) ; keepRequesting = false ; } else { LOG . error ( String . format ( "ProcessingException attempting to make request in HttpClientSource: %s" , e . getMessage ( ) ) , e ) ; Throwable reportEx = cause != null ? cause : e ; final StageException stageException = new StageException ( Errors . HTTP_32 , reportEx . toString ( ) , reportEx ) ; LOG . error ( stageException . getMessage ( ) ) ; throw stageException ; } } keepRequesting &= ! getContext ( ) . isStopped ( ) ; } // Calculate request parameter hash
currentParameterHash = hasher . hash ( ) . toString ( ) ;


signature:
public class HttpClientSource { /** * Determines whether or not we should continue making additional HTTP requests
 * in the current produce ( ) call or whether to return the current batch .
 * @ return true if we should make additional HTTP requests for this batch */
private boolean shouldMakeRequest ( ) { } }

implementation: final long now = System . currentTimeMillis ( ) ; boolean shouldMakeRequest = lastRequestCompletedTime == - 1 ; shouldMakeRequest |= lastRequestTimedOut ; shouldMakeRequest |= next != null ; shouldMakeRequest |= ( haveMorePages && conf . pagination . mode != PaginationMode . LINK_HEADER ) ; shouldMakeRequest |= now > lastRequestCompletedTime + conf . pollingInterval && conf . httpMode == HttpClientMode . POLLING ; shouldMakeRequest |= now > lastRequestCompletedTime && conf . httpMode == HttpClientMode . STREAMING && conf . httpMethod != HttpMethod . HEAD ; return shouldMakeRequest ;


signature:
public class HttpClientSource { /** * Parses the response of a completed request into records and adds them to the batch .
 * If more records are available in the response than we can add to the batch , the
 * response is not closed and parsing will continue on the next batch .
 * @ param maxRecords maximum number of records to add to the batch .
 * @ param batchMaker batch to add records to .
 * @ return the next source offset to commit
 * @ throws StageException if an unhandled error is encountered */
@ VisibleForTesting String parseResponse ( long start , int maxRecords , BatchMaker batchMaker ) throws StageException { } }

implementation:
HttpSourceOffset sourceOffset = new HttpSourceOffset ( getResolvedUrl ( ) , currentParameterHash , System . currentTimeMillis ( ) , getCurrentPage ( ) ) ; InputStream in = null ; if ( parser == null ) { // Only get a new parser if we are done with the old one .
in = getResponse ( ) . readEntity ( InputStream . class ) ; try { parser = parserFactory . getParser ( sourceOffset . toString ( ) , in , "0" ) ; } catch ( DataParserException e ) { if ( e . getErrorCode ( ) == JSON_PARSER_00 ) { LOG . warn ( "No data returned in HTTP response body." , e ) ; return sourceOffset . toString ( ) ; } LOG . warn ( "Error parsing response" , e ) ; throw e ; } } Record record = null ; int subRecordCount = 0 ; try { do { record = parser . parse ( ) ; if ( record == null ) { break ; } // LINK _ FIELD pagination
if ( conf . pagination . mode == PaginationMode . LINK_FIELD ) { // evaluate stopping condition
RecordEL . setRecordInContext ( stopVars , record ) ; haveMorePages = ! stopEval . eval ( stopVars , conf . pagination . stopCondition , Boolean . class ) ; if ( haveMorePages ) { next = Link . fromUri ( record . get ( conf . pagination . nextPageFieldPath ) . getValueAsString ( ) ) . build ( ) ; } else { next = null ; } } if ( conf . pagination . mode != PaginationMode . NONE && record . has ( conf . pagination . resultFieldPath ) ) { subRecordCount = parsePaginatedResult ( batchMaker , sourceOffset . toString ( ) , record ) ; recordCount += subRecordCount ; } else { addResponseHeaders ( record . getHeader ( ) ) ; batchMaker . addRecord ( record ) ; ++ recordCount ; } } while ( recordCount < maxRecords && ! waitTimeExpired ( start ) ) ; } catch ( IOException e ) { LOG . error ( Errors . HTTP_00 . getMessage ( ) , e . toString ( ) , e ) ; errorRecordHandler . onError ( Errors . HTTP_00 , e . toString ( ) , e ) ; } finally { try { if ( record == null ) { cleanupResponse ( in ) ; } if ( subRecordCount != 0 ) { incrementSourceOffset ( sourceOffset , subRecordCount ) ; } } catch ( IOException e ) { LOG . warn ( Errors . HTTP_28 . getMessage ( ) , e . toString ( ) , e ) ; errorRecordHandler . onError ( Errors . HTTP_28 , e . toString ( ) , e ) ; } } return sourceOffset . toString ( ) ;


signature:
public class HttpClientSource { /** * Used only for HEAD requests . Sets up a record for output based on headers only
 * with an empty body .
 * @ param batchMaker batch to add records to .
 * @ return the next source offset to commit
 * @ throws StageException if an unhandled error is encountered */
String parseHeadersOnly ( BatchMaker batchMaker ) throws StageException { } }

implementation: HttpSourceOffset sourceOffset = new HttpSourceOffset ( getResolvedUrl ( ) , currentParameterHash , System . currentTimeMillis ( ) , getCurrentPage ( ) ) ; Record record = getContext ( ) . createRecord ( sourceOffset + "::0" ) ; addResponseHeaders ( record . getHeader ( ) ) ; record . set ( Field . create ( new HashMap ( ) ) ) ; batchMaker . addRecord ( record ) ; recordCount ++ ; incrementSourceOffset ( sourceOffset , 1 ) ; lastRequestCompletedTime = System . currentTimeMillis ( ) ; return sourceOffset . toString ( ) ;


signature:
public class HttpClientSource { /** * Increments the current source offset ' s startAt portion by the specified amount .
 * This is the number of records parsed when paging BY _ OFFSET or 1 if incrementing
 * BY _ PAGE .
 * @ param sourceOffset the source offset
 * @ param increment the amount to increment the offset by */
private void incrementSourceOffset ( HttpSourceOffset sourceOffset , int increment ) { } }

implementation: if ( conf . pagination . mode == PaginationMode . BY_PAGE ) { sourceOffset . incrementStartAt ( 1 ) ; } else if ( conf . pagination . mode == PaginationMode . BY_OFFSET ) { sourceOffset . incrementStartAt ( increment ) ; }