signature:
public class StageRuntime { /** * For all PushSource callbacks we have to make sure that we get back to a security context
 * of SDC container module , otherwise we won ' t be able to update state files with new offsets
 * and other stuff . */
@ Override public final BatchContext startBatch ( ) { } }

implementation: return ( BatchContext ) AccessController . doPrivileged ( new PrivilegedAction ( ) { public Object run ( ) { try { Thread . currentThread ( ) . setContextClassLoader ( mainClassLoader ) ; return pushSourceContextDelegate . startBatch ( ) ; } finally { Thread . currentThread ( ) . setContextClassLoader ( getDefinition ( ) . getStageClassLoader ( ) ) ; } } } ) ;


signature:
public class HttpSourceOffset { /** * Parses an HttpSourceOffset from a string , e . g . from lastSourceOffset in
 * { @ link HttpClientSource # produce ( String , int , BatchMaker ) }
 * @ param s lastSourceOffset to parse
 * @ return new instance of HttpSourceOffset */
static HttpSourceOffset fromString ( String s ) { } }

implementation: LOG . debug ( "Parsing HttpSourceOffset from '{}'" , s ) ; String [ ] parts = s . split ( "::" ) ; if ( parts . length < 8 ) { throw new IllegalArgumentException ( "Offset must have at least 8 parts" ) ; } return new HttpSourceOffset ( parts [ 1 ] , parts [ 3 ] , Long . parseLong ( parts [ 5 ] ) , Integer . parseInt ( parts [ 7 ] ) ) ;


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

implementation: underlyingConfig = new TlsConfigBean ( ) ; underlyingConfig . tlsEnabled = true ; if ( useMutualAuth ) { underlyingConfig . keyStorePassword = keyStorePassword ; underlyingConfig . keyStoreFilePath = keyStoreFilePath ; underlyingConfig . keyStoreAlgorithm = keyStoreAlgorithm ; underlyingConfig . keyStoreType = keyStoreType ; underlyingConfig . init ( context , "TLS" , prefix , issues ) ; LOG . debug ( "Initialized Mutual Authentication config with {} keystore file {}" , underlyingConfig . keyStoreType , underlyingConfig . keyStoreFilePath ) ; }


signature:
public class SolrTarget { /** * Checks whether the record contains solr fields in solrFieldsMap or not .
 * @ param recordFieldMap Fields contained in the reocrd
 * @ param record The whole record
 * @ param solrFieldsMap Solr fields that will be searched in record fields
 * @ param errorToThrow error to use if missing Solr fields are detected
 * @ return true if record contains all solr fields in solrFieldsMap otherwise false
 * @ throws StageException Exception thrown by handleError method call */
private boolean checkRecordContainsSolrFields ( Map < String , Field > recordFieldMap , Record record , List < String > solrFieldsMap , Errors errorToThrow ) throws StageException { } }

implementation:
// for ( Map . Entry < String , Field > recordFieldMapEntry : recordFieldMap . entrySet ( ) )
List < String > fieldsFound = new ArrayList < > ( ) ; recordFieldMap . keySet ( ) . forEach ( recordFieldKey -> { if ( solrFieldsMap . contains ( recordFieldKey ) ) { fieldsFound . add ( recordFieldKey ) ; } } ) ; // if record does not contain solr fields then process error accordingly
if ( solrFieldsMap . size ( ) != fieldsFound . size ( ) ) { Set < String > missingFields = new HashSet < > ( ) ; solrFieldsMap . forEach ( requiredField -> { if ( ! fieldsFound . contains ( requiredField ) ) { missingFields . add ( requiredField ) ; } } ) ; handleError ( record , errorToThrow , Joiner . on ( "," ) . join ( missingFields ) ) ; return false ; } return true ;


signature:
public class SolrTarget { /** * Filter auto - generated fields from the list passed as argument .
 * @ param fieldNames List of Solr field names .
 * @ return A copy of { @ code fieldNames } with all the auto - generated fields removed . */
private List < String > filterAutogeneratedFieldNames ( List < String > fieldNames ) { } }

implementation: List < String > result = new ArrayList < > ( ) ; fieldNames . forEach ( name -> { if ( ! autogeneratedFieldNamesMap . contains ( name ) ) { result . add ( name ) ; } } ) ; return result ;


signature:
public class SolrTarget { /** * Check that the list of mappings in { @ code mappings } covers all the fields in { @ code solrFields } .
 * @ param mappings List of record field name to Solr field name mappings .
 * @ param solrFields List of Solr field names to check .
 * @ return A list with the missing Solr field names in { @ code mappings } according to { @ code solrFields } . */
private List < String > checkMissingFields ( List < SolrFieldMappingConfig > mappings , List < String > solrFields ) { } }

implementation: List < String > missingFields = new ArrayList < > ( solrFields ) ; for ( SolrFieldMappingConfig map : mappings ) { missingFields . remove ( map . solrFieldName ) ; } return missingFields ;


signature:
public class SolrTarget { /** * Handles an error that occurred when processing a record . The error can either be logged or thrown in an exception .
 * @ param record The record which was being processed
 * @ param errorTemplate The error template to be thrown if required
 * @ param errorMessage The error specific message
 * @ throws StageException Exception thrown if missingFieldAction indicates to stop the pipeline or send the record
 * to error . If missingFieldAction value is unknown an exception is also thrown */
private void handleError ( Record record , Errors errorTemplate , String errorMessage ) throws StageException { } }

implementation: handleError ( record , errorTemplate , new String [ ] { errorMessage } ) ;


signature:
public class SolrTarget { /** * Handles an error that occurred when processing a record . The error can either be logged or thrown in an exception .
 * @ param record The record which was being processed
 * @ param errorTemplate The error template to be thrown if required
 * @ param errorArguments The error specific arguments
 * @ throws StageException Exception thrown if missingFieldAction indicates to stop the pipeline or send the record
 * to error . If missingFieldAction value is unknown an exception is also thrown */
private void handleError ( Record record , Errors errorTemplate , String ... errorArguments ) throws StageException { } }

implementation:
switch ( missingFieldAction ) { case DISCARD : LOG . debug ( errorTemplate . getMessage ( ) , errorArguments ) ; break ; case STOP_PIPELINE : throw new StageException ( errorTemplate , errorArguments ) ; case TO_ERROR : throw new OnRecordErrorException ( record , errorTemplate , errorArguments ) ; default : // unknown operation
LOG . debug ( "Sending record to error due to unknown operation {}" , missingFieldAction ) ; throw new OnRecordErrorException ( record , errorTemplate , errorArguments ) ; }


signature:
public class SolrTarget { /** * Send exception ex to errorRecordHandler in order to let the handler process it .
 * @ param record The record which was being processed when the exception was thrown
 * @ param error The error template
 * @ param ex The exception that was thrown
 * @ throws StageException Exception thrown by the errorRecordHandler when handling the exception ex */
private void sendOnRecordErrorExceptionToHandler ( Record record , Errors error , StageException ex ) throws StageException { } }

implementation: errorRecordHandler . onError ( new OnRecordErrorException ( record , error , record . getHeader ( ) . getSourceId ( ) , ex . toString ( ) , ex ) ) ;


signature:
public class DependencyParser { /** * Generate dependency from a jar file name . */
public static Optional < Dependency > parseJarName ( String sourceName , String jarName ) { } }

implementation:
if ( SPECIAL_CASES . containsKey ( jarName ) ) { Dependency specialCase = SPECIAL_CASES . get ( jarName ) ; return Optional . of ( new Dependency ( sourceName , specialCase . getName ( ) , specialCase . getVersion ( ) ) ) ; } // Go over all known patterns
for ( Pattern p : PATTERNS ) { Matcher m = p . matcher ( jarName ) ; if ( m . matches ( ) ) { LOG . trace ( "Applied pattern '{}' to {}" , p . pattern ( ) , jarName ) ; return Optional . of ( new Dependency ( sourceName , m . group ( 1 ) , m . group ( 2 ) ) ) ; } } // Otherwise this jar name is unknown to us
return Optional . empty ( ) ;


signature:
public class DependencyParser { /** * Generate dependency from a URL . */
public static Optional < Dependency > parseURL ( URL url ) { } }

implementation: return parseJarName ( url . toString ( ) , Paths . get ( url . getPath ( ) ) . getFileName ( ) . toString ( ) ) ;


signature:
public class CouchbaseConnector { /** * Maintains a singleton instance of the CouchbaseConnector object per pipeline
 * @ param config the stage configuration
 * @ param issues the list of configuration issues
 * @ param context the context for the stage
 * @ return a singleton instance for Couchbase connections */
public static synchronized CouchbaseConnector getInstance ( BaseCouchbaseConfig config , List < Stage . ConfigIssue > issues , Stage . Context context ) { } }

implementation: Map < String , Object > runnerSharedMap = context . getStageRunnerSharedMap ( ) ; if ( runnerSharedMap . containsKey ( INSTANCE ) ) { LOG . debug ( "Using existing instance of CouchbaseConnector" ) ; } else { LOG . debug ( "CouchbaseConnector not yet instantiated. Creating new instance" ) ; validateConfig ( config , issues , context ) ; if ( issues . isEmpty ( ) ) { runnerSharedMap . put ( INSTANCE , new CouchbaseConnector ( config , issues , context ) ) ; } } return ( CouchbaseConnector ) runnerSharedMap . get ( INSTANCE ) ;


signature:
public class CouchbaseConnector { /** * Disconnects from Couchbase and releases all resources */
public synchronized void close ( ) { } }

implementation:
if ( ! isClosed ) { if ( bucket != null ) { LOG . debug ( "Closing Couchbase bucket" ) ; bucket . close ( ) ; } if ( cluster != null ) { LOG . debug ( "Disconnecting Couchbase cluster" ) ; cluster . disconnect ( ) ; } if ( env != null ) { LOG . debug ( "Shutting down Couchbase environment" ) ; env . shutdown ( ) ; } // Explicitly shutdown the RxJava scheduler threads . Not doing so will leak threads when a pipeline stops .
// Note : this disallows restarting scheduler threads without also explicitly calling Schedulers . start ( )
// LOG . debug ( " Stopping RxJava schedulers " ) ;
// Schedulers . shutdown ( ) ;
isClosed = true ; }


signature:
public class CouchbaseConnector { /** * Validates connection configurations that don ' t require runtime exception handling
 * @ param config the stage configuration
 * @ param issues the list of configuration issues
 * @ param context the context for the stage */
private static void validateConfig ( BaseCouchbaseConfig config , List < Stage . ConfigIssue > issues , Stage . Context context ) { } }

implementation: if ( config . couchbase . nodes == null ) { issues . add ( context . createConfigIssue ( Groups . COUCHBASE . name ( ) , "config.couchbase.nodes" , Errors . COUCHBASE_29 ) ) ; } if ( config . couchbase . kvTimeout < 0 ) { issues . add ( context . createConfigIssue ( Groups . COUCHBASE . name ( ) , "config.couchbase.kvTimeout" , Errors . COUCHBASE_30 ) ) ; } if ( config . couchbase . connectTimeout < 0 ) { issues . add ( context . createConfigIssue ( Groups . COUCHBASE . name ( ) , "config.couchbase.connectTimeout" , Errors . COUCHBASE_31 ) ) ; } if ( config . couchbase . disconnectTimeout < 0 ) { issues . add ( context . createConfigIssue ( Groups . COUCHBASE . name ( ) , "config.couchbase.disconnectTimeout" , Errors . COUCHBASE_32 ) ) ; } if ( config . couchbase . tls . tlsEnabled ) { config . couchbase . tls . init ( context , Groups . COUCHBASE . name ( ) , "config.couchbase.tls." , issues ) ; } if ( config . credentials . version == null ) { issues . add ( context . createConfigIssue ( Groups . CREDENTIALS . name ( ) , "config.credentials.version" , Errors . COUCHBASE_33 ) ) ; } if ( config . credentials . version == AuthenticationType . USER ) { if ( config . credentials . userName == null ) { issues . add ( context . createConfigIssue ( Groups . CREDENTIALS . name ( ) , "config.credentials.userName" , Errors . COUCHBASE_34 ) ) ; } if ( config . credentials . userPassword == null ) { issues . add ( context . createConfigIssue ( Groups . CREDENTIALS . name ( ) , "config.credentials.userPassword" , Errors . COUCHBASE_35 ) ) ; } }


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

implementation:
// Offsets can vary depending on the data source . Here we use an integer as an example only .
long nextSourceOffset = 0 ; lastSourceOffset = lastSourceOffset == null ? "" : lastSourceOffset ; if ( ! lastSourceOffset . equals ( "" ) ) { nextSourceOffset = Long . parseLong ( lastSourceOffset ) ; } int recordCounter = 0 ; long startTime = System . currentTimeMillis ( ) ; int maxRecords = Math . min ( maxBatchSize , conf . maxBatchSize ) ; while ( recordCounter < maxRecords && ( startTime + conf . maxWaitTime ) > System . currentTimeMillis ( ) ) { String message = buffer . poll ( ) ; if ( null == message ) { try { Thread . sleep ( 100 ) ; } catch ( Exception e ) { LOG . debug ( e . getMessage ( ) , e ) ; break ; } } else { List < Record > records = processRedisMessage ( "id::" + nextSourceOffset , message ) ; for ( Record record : records ) { batchMaker . addRecord ( record ) ; } recordCounter += records . size ( ) ; ++ nextSourceOffset ; } } return lastSourceOffset ;


signature:
public class OperationType { /** * Convert from code in String type to label
 * @ param code
 * @ return */
public static String getLabelFromStringCode ( String code ) throws NumberFormatException { } }

implementation: try { int intCode = Integer . parseInt ( code ) ; return getLabelFromIntCode ( intCode ) ; } catch ( NumberFormatException ex ) { throw new NumberFormatException ( String . format ( "%s but received '%s'" , "operation code must be numeric" , code ) ) ; }


signature:
public class ConfigDefinitionExtractor { /** * Verify that the config definition ' s dependency actually maps to a valid config definition */
private void verifyDependencyExists ( Map < String , ConfigDefinition > definitionsMap , ConfigDefinition def , String dependsOnKey , Object contextMsg ) { } }

implementation: Preconditions . checkState ( definitionsMap . containsKey ( dependsOnKey ) , Utils . format ( "Error while processing {} ConfigDef='{}'. Dependency='{}' does not exist." , contextMsg , def . getName ( ) , dependsOnKey ) ) ;


signature:
public class ConfigDefinitionExtractor { /** * Returns true if child creates a dependency with any member ( s ) of dependencyAncestors .
 * Also adds the stringified cycle to the cycles list */
private boolean detectCycle ( LinkedHashSet < String > dependencyAncestors , Set < String > cycles , final String child ) { } }

implementation:
if ( dependencyAncestors . contains ( child ) ) { // Find index of the child in the ancestors list
int index = - 1 ; for ( String s : dependencyAncestors ) { index ++ ; if ( s . equals ( child ) ) { break ; } } // The cycle starts from the first time the child is seen in the ancestors list
// and continues till the end of the list , followed by the child again .
cycles . add ( Joiner . on ( " -> " ) . join ( Iterables . skip ( dependencyAncestors , index ) ) + " -> " + child ) ; return true ; } return false ;


signature:
public class S3Accessor { /** * visible for testing */
AWSCredentialsProvider createCredentialsProvider ( ) throws StageException { } }

implementation: String accessKey = credentialConfigs . getAccessKey ( ) . get ( ) ; String secretKey = credentialConfigs . getSecretKey ( ) . get ( ) ; if ( accessKey != null && ! accessKey . isEmpty ( ) && secretKey != null && ! secretKey . isEmpty ( ) ) { return new AWSStaticCredentialsProvider ( new BasicAWSCredentials ( credentialConfigs . getAccessKey ( ) . get ( ) , credentialConfigs . getSecretKey ( ) . get ( ) ) ) ; } return null ;


signature:
public class S3Accessor { /** * visible for testing */
ClientConfiguration createClientConfiguration ( ) throws StageException { } }

implementation: ClientConfiguration clientConfig = new ClientConfiguration ( ) ; clientConfig . setConnectionTimeout ( connectionConfigs . getConnectionTimeoutMillis ( ) ) ; clientConfig . setSocketTimeout ( connectionConfigs . getSocketTimeoutMillis ( ) ) ; clientConfig . withMaxErrorRetry ( connectionConfigs . getMaxErrorRetry ( ) ) ; if ( connectionConfigs . isProxyEnabled ( ) ) { clientConfig . setProxyHost ( connectionConfigs . getProxyHost ( ) ) ; clientConfig . setProxyPort ( connectionConfigs . getProxyPort ( ) ) ; if ( connectionConfigs . isProxyAuthenticationEnabled ( ) ) { clientConfig . setProxyUsername ( connectionConfigs . getProxyUser ( ) . get ( ) ) ; clientConfig . setProxyPassword ( connectionConfigs . getProxyPassword ( ) . get ( ) ) ; } } return clientConfig ;


signature:
public class S3Accessor { /** * visible for testing */
AmazonS3Client createS3Client ( ) throws StageException { } }

implementation:
AmazonS3ClientBuilder builder = createAmazonS3ClientBuilder ( ) . withClientConfiguration ( createClientConfiguration ( ) ) . withChunkedEncodingDisabled ( connectionConfigs . isChunkedEncodingEnabled ( ) ) . withPathStyleAccessEnabled ( true ) ; AWSCredentialsProvider awsCredentialsProvider = getCredentialsProvider ( ) ; // If we don ' t call build . withCredentials ( . . . ) then we will not overwrite the default credentials provider
// already set in the builder when doing AmazonS3ClientBuilder . standard ( ) so only calling build . withCredentials ( . . . )
// if our own provider exists
if ( awsCredentialsProvider != null ) { builder . withCredentials ( awsCredentialsProvider ) ; } String region = ( connectionConfigs . getRegion ( ) == null || connectionConfigs . getRegion ( ) . isEmpty ( ) ) ? null : connectionConfigs . getRegion ( ) ; if ( connectionConfigs . isUseEndpoint ( ) ) { builder . withEndpointConfiguration ( new AwsClientBuilder . EndpointConfiguration ( connectionConfigs . getEndpoint ( ) , region ) ) ; } else if ( region != null ) { builder . withRegion ( connectionConfigs . getRegion ( ) ) ; } else { builder . withRegion ( AwsRegion . US_WEST_1 . getId ( ) ) ; builder . withForceGlobalBucketAccessEnabled ( true ) ; } return ( AmazonS3Client ) builder . build ( ) ;


signature:
public class S3Accessor { /** * visible for testing */
TransferManager createTransferManager ( AmazonS3 s3Client ) throws StageException { } }

implementation: return createTransferManagerBuilder ( ) . withS3Client ( s3Client ) . withExecutorFactory ( createExecutorFactory ( transferManagerConfigs . getThreads ( ) ) ) . withShutDownThreadPools ( true ) . withMinimumUploadPartSize ( transferManagerConfigs . getMinimumUploadPartSize ( ) ) . withMultipartUploadThreshold ( transferManagerConfigs . getMultipartUploadThreshold ( ) ) . build ( ) ;


signature:
public class Vault { /** * The user - id portion of Vault ' s app - auth should be machine specific and at least
 * somewhat obfuscated to make it more difficult to derive . We use the sha256 hash of
 * the MAC address of the first interface found by InetAddress . getLocalHost ( ) .
 * This provides a way for administrators to compute the value itself during / after
 * deployment and authorize the app - id , user - id pair out of band .
 * @ return String representation of the user - id portion of an auth token . */
static String calculateUserId ( String csId ) { } }

implementation:
try { // Try to hash based on default interface
InetAddress ip = InetAddress . getLocalHost ( ) ; NetworkInterface netIf = NetworkInterface . getByInetAddress ( ip ) ; byte [ ] mac = netIf . getHardwareAddress ( ) ; if ( mac == null ) { // In some cases the default interface may be a tap / tun device which has no MAC
// instead pick the first available interface .
Enumeration < NetworkInterface > netIfs = NetworkInterface . getNetworkInterfaces ( ) ; while ( netIfs . hasMoreElements ( ) && mac == null ) { netIf = netIfs . nextElement ( ) ; mac = netIf . getHardwareAddress ( ) ; } } if ( mac == null ) { throw new IllegalStateException ( "Could not find network interface with MAC address." ) ; } Hasher hasher = HASH_FUNCTION . newHasher ( 6 ) ; // MAC is 6 bytes .
return hasher . putBytes ( mac ) . hash ( ) . toString ( ) ; } catch ( IOException e ) { LOG . error ( "CredentialStore '{}' Vault, could not compute Vault user-id: '{}'" , csId , e ) ; throw new VaultRuntimeException ( "Could not compute Vault user-id: " + e . toString ( ) ) ; }


signature:
public class Vault { /** * Reads a secret from the local cache if it hasn ' t expired and returns the value for the specified key .
 * If the secret isn ' t cached or has expired , it requests it from Vault again .
 * This version of the method will also add the specified delay in milliseconds before returning , but only
 * if the value wasn ' t already locally cached . This is because certain backends such as AWS will return
 * a secret ( access keys for example ) before they have propagated to all AWS services . For AWS a delay of up to 5
 * or 10 seconds may be necessary . If you receive a 403 from AWS services you probably need to increase the delay .
 * @ param path path in Vault to read
 * @ param key key of the property of the secret represented by the path to return
 * @ param delay delay in milliseconds to wait before returning the value if it wasn ' t already cached .
 * @ return value of the specified key for the requested secret . */
public String read ( String path , String key , long delay ) { } }

implementation:
if ( ! secrets . containsKey ( path ) ) { VaultClient vault = new VaultClient ( getConfig ( ) ) ; Secret secret ; try { secret = vault . logical ( ) . read ( path ) ; } catch ( VaultException e ) { LOG . error ( e . toString ( ) , e ) ; throw new VaultRuntimeException ( e . toString ( ) ) ; } // Record the expiration date of this lease
String leaseId ; if ( secret . isRenewable ( ) ) { // Only renewable secrets seem to have a leaseId
leaseId = secret . getLeaseId ( ) ; } else { // So for non - renewable secrets we ' ll store the path with an extra / so that we can purge them correctly .
leaseId = path + "/" ; } leases . put ( leaseId , System . currentTimeMillis ( ) + ( secret . getLeaseDuration ( ) * 1000 ) ) ; secrets . put ( path , secret ) ; try { Thread . sleep ( delay ) ; } catch ( InterruptedException e ) { Thread . currentThread ( ) . interrupt ( ) ; } } Map < String , Object > data = secrets . get ( path ) . getData ( ) ; String value = getSecretValue ( data , key ) . orElseThrow ( ( ) -> new VaultRuntimeException ( "Value not found for key" ) ) ; LOG . trace ( "CredentialStore '{}' Vault, retrieved value for key '{}'" , csId , key ) ; return value ;


signature:
public class CouchbaseTarget { /** * Evaluates the sdc . operation . type header for a record and returns the equivalent Couchbase write operation type .
 * @ param record the record being evaluated
 * @ param key the document key , used for logging
 * @ return the equivalent WriteOperationType or the default write operation if sdc . operation . type is not configured . */
private WriteOperationType getOperationFromHeader ( Record record , String key ) { } }

implementation: String op = record . getHeader ( ) . getAttribute ( OperationType . SDC_OPERATION_TYPE ) ; if ( op == null || op . isEmpty ( ) ) { return config . defaultWriteOperation ; } int opCode ; try { opCode = Integer . parseInt ( op ) ; } catch ( NumberFormatException e ) { LOG . debug ( "Unparsable CDC operation. Sending record to error." ) ; handleError ( record , Errors . COUCHBASE_08 , e ) ; return null ; } switch ( opCode ) { case OperationType . INSERT_CODE : return WriteOperationType . INSERT ; case OperationType . UPDATE_CODE : return WriteOperationType . REPLACE ; case OperationType . UPSERT_CODE : return WriteOperationType . UPSERT ; case OperationType . DELETE_CODE : return WriteOperationType . DELETE ; default : switch ( config . unsupportedOperation ) { case DISCARD : LOG . debug ( "Unsupported CDC operation for key: {}. Discarding record per configuration." , key ) ; return null ; case TOERROR : LOG . debug ( "Unsupported CDC operation for key: {}. Sending record to error configuration." , key ) ; handleError ( record , Errors . COUCHBASE_09 , new RuntimeException ( ) ) ; return null ; default : LOG . debug ( "Unsupported CDC operation for key: {}. Using default write operation per configuration." , key ) ; return config . defaultWriteOperation ; } }


signature:
public class CouchbaseTarget { /** * Executes a document write operation .
 * @ param key the document key
 * @ param ttl the document expiry ttl
 * @ param cas the compare - and - swap ( CAS ) value to apply to the write
 * @ param baos the raw document content
 * @ param record the record being written
 * @ return an observable for the document write or an empty observable on error */
private Observable < AbstractDocument > writeDoc ( String key , int ttl , long cas , ByteArrayOutputStream baos , Record record ) { } }

implementation: WriteOperationType opType = getOperationFromHeader ( record , key ) ; if ( opType == null ) { return Observable . empty ( ) ; } AbstractDocument doc ; if ( config . dataFormat == DataFormat . JSON ) { try { doc = JsonDocument . create ( key , ttl , JsonObject . fromJson ( baos . toString ( config . dataFormatConfig . charset ) ) , cas ) ; } catch ( Exception e ) { return handleError ( record , Errors . COUCHBASE_10 , e ) ; } } else { doc = ByteArrayDocument . create ( key , ttl , baos . toByteArray ( ) , cas ) ; } switch ( opType ) { case DELETE : { LOG . debug ( "DELETE key: {}, TTL: {}, CAS: {}" , key , ttl , cas ) ; return connector . bucket ( ) . remove ( doc , config . persistTo , config . replicateTo ) . timeout ( config . couchbase . kvTimeout , TimeUnit . MILLISECONDS ) ; } case INSERT : { LOG . debug ( "INSERT key: {}, TTL: {}, CAS: {}" , key , ttl , cas ) ; return connector . bucket ( ) . insert ( doc , config . persistTo , config . replicateTo ) . timeout ( config . couchbase . kvTimeout , TimeUnit . MILLISECONDS ) ; } case REPLACE : { LOG . debug ( "REPLACE key: {}, TTL: {}, CAS: {}" , key , ttl , cas ) ; return connector . bucket ( ) . replace ( doc , config . persistTo , config . replicateTo ) . timeout ( config . couchbase . kvTimeout , TimeUnit . MILLISECONDS ) ; } case UPSERT : { LOG . debug ( "UPSERT key: {}, TTL: {}, CAS: {}" , key , ttl , cas ) ; return connector . bucket ( ) . upsert ( doc , config . persistTo , config . replicateTo ) . timeout ( config . couchbase . kvTimeout , TimeUnit . MILLISECONDS ) ; } default : return Observable . empty ( ) ; }


signature:
public class CouchbaseTarget { /** * Executes a sub - document write operation .
 * @ param key the document key
 * @ param ttl the document expiry ttl
 * @ param cas the compare - and - swap ( CAS ) value to apply to the write
 * @ param baos the raw document content
 * @ param subdocPath the path within the parent document in which to write the subdoc
 * @ param record the record being written
 * @ return an observable for the subdoc mutation or an empty observable on error */
private Observable < DocumentFragment < Mutation > > writeSubdoc ( String key , int ttl , long cas , ByteArrayOutputStream baos , String subdocPath , Record record ) { } }

implementation:
JsonObject frag ; try { frag = JsonObject . fromJson ( baos . toString ( config . dataFormatConfig . charset ) ) ; } catch ( IOException e ) { return handleError ( record , Errors . COUCHBASE_12 , e ) ; } AsyncMutateInBuilder mutation = connector . bucket ( ) . mutateIn ( key ) ; SubdocOptionsBuilder options = new SubdocOptionsBuilder ( ) . createPath ( true ) ; String subdocOpType ; try { subdocOpType = subdocOperationELEval . eval ( elVars , config . subdocOperationEL , String . class ) ; } catch ( ELEvalException e ) { return handleError ( record , Errors . COUCHBASE_13 , e ) ; } if ( config . allowSubdoc && ! subdocOpType . isEmpty ( ) ) { switch ( subdocOpType . toUpperCase ( ) ) { case "DELETE" : LOG . debug ( "Sub-document DELETE key: {}, sub-document path: {}, TTL: {}, CAS: {}" , key , subdocPath , ttl , cas ) ; return buildSubdocMutation ( mutation . remove ( subdocPath , options ) , ttl , cas , false ) ; case "INSERT" : LOG . debug ( "Sub-document INSERT key: {}, sub-document path: {}, TTL: {}, CAS: {}" , key , subdocPath , ttl , cas ) ; return buildSubdocMutation ( mutation . insert ( subdocPath , frag , options ) , ttl , cas , true ) ; case "REPLACE" : LOG . debug ( "Sub-document REPLACE key: {}, sub-document path: {}, TTL: {}, CAS: {}" , key , subdocPath , ttl , cas ) ; return buildSubdocMutation ( mutation . replace ( subdocPath , frag ) , ttl , cas , false ) ; case "UPSERT" : LOG . debug ( "Sub-document UPSERT key: {}, sub-document path: {}, TTL: {}, CAS: {}" , key , subdocPath , ttl , cas ) ; return buildSubdocMutation ( mutation . upsert ( subdocPath , frag , options ) , ttl , cas , true ) ; case "ARRAY_PREPEND" : LOG . debug ( "Sub-document ARRAY_PREPEND key: {}, sub-document path: {}, TTL: {}, CAS: {}" , key , subdocPath , ttl , cas ) ; return buildSubdocMutation ( mutation . arrayPrepend ( subdocPath , frag , options ) , ttl , cas , true ) ; case "ARRAY_APPEND" : LOG . debug ( "Sub-document ARRAY_APPEND key: {}, sub-document path: {}, TTL: {}, CAS: {}" , key , subdocPath , ttl , cas ) ; return buildSubdocMutation ( mutation . arrayAppend ( subdocPath , frag , options ) , ttl , cas , true ) ; case "ARRAY_ADD_UNIQUE" : LOG . debug ( "Sub-document ARRAY_ADD_UNIQUE key: {}, sub-document path: {}, TTL: {}, CAS: {}" , key , subdocPath , ttl , cas ) ; return buildSubdocMutation ( mutation . arrayAddUnique ( subdocPath , frag , options ) , ttl , cas , true ) ; default : switch ( config . unsupportedOperation ) { case DISCARD : LOG . debug ( "Unsupported sub-document operation: {} for key: {}. Discarding record per configuration." , subdocOpType , key ) ; return Observable . empty ( ) ; case TOERROR : LOG . debug ( "Unsupported sub-document operation: {} for key: {}. Sending record to error per configuration." , subdocOpType , key ) ; return handleError ( record , Errors . COUCHBASE_14 , new RuntimeException ( ) ) ; default : LOG . debug ( "Unsupported sub-document operation: {} for key: {}. Using default write operation per configuration." , subdocOpType , key ) ; // Fall through
// Inherit the CDC or default operation
} } } WriteOperationType opType = getOperationFromHeader ( record , key ) ; if ( opType == null ) { return Observable . empty ( ) ; } switch ( opType ) { case DELETE : LOG . debug ( "Sub-document DELETE key: {}, sub-document path: {}, TTL: {}, CAS: {}" , key , subdocPath , ttl , cas ) ; return buildSubdocMutation ( mutation . remove ( subdocPath , options ) , ttl , cas , false ) ; case INSERT : LOG . debug ( "Sub-document INSERT key: {}, sub-document path: {}, TTL: {}, CAS: {}" , key , subdocPath , ttl , cas ) ; return buildSubdocMutation ( mutation . insert ( subdocPath , frag , options ) , ttl , cas , true ) ; case REPLACE : LOG . debug ( "Sub-document REPLACE key: {}, sub-document path: {}, TTL: {}, CAS: {}" , key , subdocPath , ttl , cas ) ; return buildSubdocMutation ( mutation . replace ( subdocPath , frag ) , ttl , cas , false ) ; case UPSERT : LOG . debug ( "Sub-document UPSERT key: {}, sub-document path: {}, TTL: {}, CAS: {}" , key , subdocPath , ttl , cas ) ; return buildSubdocMutation ( mutation . upsert ( subdocPath , frag , options ) , ttl , cas , true ) ; default : return Observable . empty ( ) ; }


signature:
public class CouchbaseTarget { /** * Applies standard options to sub - document mutations
 * @ param mutation the mutation to be augmented
 * @ param ttl the time to live to be set on the parent document
 * @ param cas the compare - and - swap ( CAS ) value to apply to the write
 * @ param upsertDoc defines whether the parent document should be created if it does not exist
 * @ return an augmented mutation */
private Observable < DocumentFragment < Mutation > > buildSubdocMutation ( AsyncMutateInBuilder mutation , int ttl , long cas , boolean upsertDoc ) { } }

implementation: return mutation . upsertDocument ( upsertDoc ) . withExpiry ( ttl ) . withCas ( cas ) . withDurability ( config . persistTo , config . replicateTo ) . execute ( ) . timeout ( config . couchbase . kvTimeout , TimeUnit . MILLISECONDS ) ;


signature:
public class SecurityUtil { /** * Influenced by javax . security . auth . Subject . # doAs */
public static < T > T doAs ( Subject subject , PrivilegedExceptionAction < T > privilegedExceptionAction ) throws PrivilegedActionException { } }

implementation: checkDoAsPermission ( ) ; if ( privilegedExceptionAction == null ) { throw new RuntimeException ( "No privileged exception action provided" ) ; } return AccessController . doPrivileged ( privilegedExceptionAction , createContext ( subject , AccessController . getContext ( ) ) ) ;


signature:
public class SecurityUtil { /** * Influenced by javax . security . auth . Subject . # doAs */
public static < T > T doAs ( final Subject subject , final PrivilegedAction < T > privilegedAction ) { } }

implementation: checkDoAsPermission ( ) ; if ( privilegedAction == null ) { throw new RuntimeException ( "No privileged action provided" ) ; } return AccessController . doPrivileged ( privilegedAction , createContext ( subject , AccessController . getContext ( ) ) ) ;


signature:
public class WholeFileFormatFsHelper { /** * because the path is going to be used only once ( because only one record is used ) . */
@ Override public Path getPath ( FileSystem fs , Date recordDate , Record record ) throws StageException , IOException { } }

implementation:
// Check whether the real file already exists
Path path = new Path ( mgr . getDirPath ( recordDate , record ) , getTempFile ( recordDate , record ) ) ; // this will check the file exists .
Path renamableFinalPath = getRenamablePath ( fs , path ) ; updateFsPermissionsIfNeeded ( record ) ; wholeFileEventRecord = createWholeFileEventRecord ( record , renamableFinalPath ) ; return path ;


signature:
public class H2SchemaWriter { /** * H2 wants an ALTER TABLE command per column */
@ Override protected String makeAlterTableSqlString ( String schema , String tableName , LinkedHashMap < String , JdbcTypeInfo > columnDiff ) { } }

implementation: String tableSchema = ( schema == null ) ? getDefaultSchema ( ) : schema ; StringBuilder sqlString = new StringBuilder ( ) ; boolean first = true ; for ( Map . Entry < String , JdbcTypeInfo > entry : columnDiff . entrySet ( ) ) { if ( first ) { first = false ; } else { sqlString . append ( "\n" ) ; } sqlString . append ( ALTER_TABLE ) . append ( " " ) ; if ( tableSchema != null ) { sqlString . append ( tableSchema ) ; sqlString . append ( "." ) ; } sqlString . append ( tableName ) . append ( " " ) . append ( "ADD COLUMN" ) . append ( " " ) . append ( entry . getKey ( ) ) . append ( " " ) . append ( entry . getValue ( ) . toString ( ) ) . append ( ";" ) ; } return sqlString . toString ( ) ;


signature:
public class DelimitedLengthFieldBasedFrameDecoder { /** * Create a frame out of the { @ link ByteBuf } and return it .
 * @ param ctx the { @ link ChannelHandlerContext } which this { @ link ByteToMessageDecoder } belongs to
 * @ param in the { @ link ByteBuf } from which to read data
 * @ return frame the { @ link ByteBuf } which represent the frame or { @ code null } if no frame could
 * be created . */
protected Object decode ( ChannelHandlerContext ctx , ByteBuf in ) throws Exception { } }

implementation:
if ( discardingTooLongFrame ) { long bytesToDiscard = this . bytesToDiscard ; int localBytesToDiscard = ( int ) Math . min ( bytesToDiscard , in . readableBytes ( ) ) ; in . skipBytes ( localBytesToDiscard ) ; bytesToDiscard -= localBytesToDiscard ; this . bytesToDiscard = bytesToDiscard ; failIfNecessary ( false ) ; return null ; } if ( consumingLength ) { int delimIndex = indexOf ( in , delimiter ) ; if ( delimIndex < 0 ) { return null ; } final String lengthStr = in . toString ( in . readerIndex ( ) , delimIndex , lengthFieldCharset ) ; try { frameLength = Long . parseLong ( trimLengthString ? lengthStr . trim ( ) : lengthStr ) ; } catch ( NumberFormatException e ) { throw new CorruptedFrameException ( String . format ( "Invalid length field decoded (in %s charset): %s" , lengthFieldCharset . name ( ) , lengthStr ) , e ) ; } if ( frameLength < 0 ) { throw new CorruptedFrameException ( "negative pre-adjustment length field: " + frameLength ) ; } frameLength += lengthAdjustment ; // consume length field and delimiter bytes
in . skipBytes ( delimIndex + delimiter . capacity ( ) ) ; // consume delimiter bytes
consumingLength = false ; } if ( frameLength > maxFrameLength ) { long discard = frameLength - in . readableBytes ( ) ; tooLongFrameLength = frameLength ; if ( discard < 0 ) { // buffer contains more bytes then the frameLength so we can discard all now
in . skipBytes ( ( int ) frameLength ) ; } else { // Enter the discard mode and discard everything received so far .
discardingTooLongFrame = true ; consumingLength = true ; bytesToDiscard = discard ; in . skipBytes ( in . readableBytes ( ) ) ; } failIfNecessary ( true ) ; return null ; } // never overflows because it ' s less than maxFrameLength
int frameLengthInt = ( int ) frameLength ; if ( in . readableBytes ( ) < frameLengthInt ) { // need more bytes available to read actual frame
return null ; } // the frame is now entirely present , reset state vars
consumingLength = true ; frameLength = 0 ; // extract frame
int readerIndex = in . readerIndex ( ) ; int actualFrameLength = frameLengthInt ; // - initialBytesToStrip ;
ByteBuf frame = extractFrame ( ctx , in , readerIndex , actualFrameLength ) ; in . readerIndex ( readerIndex + actualFrameLength ) ; return frame ;


signature:
public class DelimitedLengthFieldBasedFrameDecoder { /** * Extract the sub - region of the specified buffer .
 * If you are sure that the frame and its content are not accessed after
 * the current { @ link # decode ( ChannelHandlerContext , ByteBuf ) }
 * call returns , you can even avoid memory copy by returning the sliced
 * sub - region ( i . e . < tt > return buffer . slice ( index , length ) < / tt > ) .
 * It ' s often useful when you convert the extracted frame into an object .
 * Refer to the source code of { @ link ObjectDecoder } to see how this method
 * is overridden to avoid memory copy . */
protected ByteBuf extractFrame ( ChannelHandlerContext ctx , ByteBuf buffer , int index , int length ) { } }

implementation: return buffer . slice ( index , length ) . retain ( ) ;


signature:
public class KerberosLogin { /** * Login constructor . The constructor starts the thread used
 * to periodically re - login to the Kerberos Ticket Granting Server .
 * @ param loginContextName
 * name of section in JAAS file that will be use to login .
 * Passed as first param to javax . security . auth . login . LoginContext ( ) .
 * @ param configs configure Login with the given key - value pairs .
 * @ throws javax . security . auth . login . LoginException
 * Thrown if authentication fails . */
public void configure ( Map < String , ? > configs , final String loginContextName ) { } }

implementation: super . configure ( configs , loginContextName ) ; this . loginContextName = loginContextName ; this . ticketRenewWindowFactor = ( Double ) configs . get ( SaslConfigs . SASL_KERBEROS_TICKET_RENEW_WINDOW_FACTOR ) ; this . ticketRenewJitter = ( Double ) configs . get ( SaslConfigs . SASL_KERBEROS_TICKET_RENEW_JITTER ) ; this . minTimeBeforeRelogin = ( Long ) configs . get ( SaslConfigs . SASL_KERBEROS_MIN_TIME_BEFORE_RELOGIN ) ; this . kinitCmd = ( String ) configs . get ( SaslConfigs . SASL_KERBEROS_KINIT_CMD ) ; this . serviceName = getServiceName ( configs , loginContextName ) ;


signature:
public class KerberosLogin { /** * 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 ( loginContext == null ) { throw new LoginException ( "Login must be done first" ) ; } if ( ! hasSufficientTimeElapsed ( ) ) { return ; } log . info ( "Initiating logout for {}" , principal ) ; synchronized ( KerberosLogin . 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
loginContext . logout ( ) ; // login and also update the subject field of this instance to
// have the new credentials ( pass it to the LoginContext constructor )
loginContext = new LoginContext ( loginContextName , subject ) ; log . info ( "Initiating re-login for {}" , principal ) ; loginContext . login ( ) ; }


signature:
public class AbstractTableJdbcSource { /** * Checks whether any of the { @ link TableJdbcRunnable } workers completed
 * and whether there is any error that needs to be handled from them .
 * @ param completionService { @ link ExecutorCompletionService } used to detect completion
 * @ throws StageException if { @ link StageException } is thrown by the workers ( if the error handling is stop pipeline ) */
private void checkWorkerStatus ( ExecutorCompletionService < Future > completionService ) throws StageException { } }

implementation: Future future = completionService . poll ( ) ; if ( future != null ) { try { future . get ( ) ; } catch ( InterruptedException e ) { LOG . error ( "Thread interrupted" , e ) ; } catch ( ExecutionException e ) { Throwable cause = Throwables . getRootCause ( e ) ; if ( cause != null && cause instanceof StageException ) { throw ( StageException ) cause ; } else { LOG . error ( "Internal Error" , e ) ; throw new StageException ( JdbcErrors . JDBC_75 , e . toString ( ) , e ) ; } } }


signature:
public class Consumer { /** * Consumes messages off the queue . Returns null when the producer
 * has indicated it is complete and throws an exception
 * when the consumer producer has indicated it is in error . */
public OffsetAndResult < Map . Entry > take ( ) { } }

implementation:
if ( producerError != null ) { throw new RuntimeException ( Utils . format ( "Producer encountered error: {}" , producerError ) , producerError ) ; } if ( consumerError != null ) { throw new RuntimeException ( Utils . format ( "Consumer encountered error: {}" , consumerError ) , consumerError ) ; } try { Utils . checkState ( batchCommitted , "Cannot take messages when last batch is uncommitted" ) ; while ( running ) { for ( ControlChannel . Message controlMessage : controlChannel . getConsumerMessages ( ) ) { switch ( controlMessage . getType ( ) ) { case PRODUCER_COMPLETE : // producer is complete , empty channel and afterwards return null
running = false ; break ; case PRODUCER_ERROR : running = false ; Throwable throwable = ( Throwable ) controlMessage . getPayload ( ) ; producerError = throwable ; throw new ProducerRuntimeException ( Utils . format ( "Producer encountered error: {}" , throwable ) , throwable ) ; default : String msg = Utils . format ( "Illegal control message type: '{}'" , controlMessage . getType ( ) ) ; throw new IllegalStateException ( msg ) ; } } OffsetAndResult < Map . Entry > batch = dataChannel . take ( 10 , TimeUnit . MILLISECONDS ) ; LOG . trace ( "Received batch: {}" , batch ) ; if ( batch != null ) { batchCommitted = false ; // got a new batch
return batch ; } } LOG . trace ( "Returning null" ) ; return null ; } catch ( Throwable throwable ) { if ( ! ( throwable instanceof ProducerRuntimeException ) ) { String msg = "Error caught in consumer: " + throwable ; LOG . error ( msg , throwable ) ; error ( throwable ) ; } throw Throwables . propagate ( throwable ) ; }


signature:
public class Consumer { /** * Commit the offset . Required after take has returned a non - null value . */
public void commit ( String offset ) { } }

implementation: batchCommitted = true ; LOG . trace ( "Last committed offset '{}', attempting to commit '{}'" , lastCommittedOffset , offset ) ; Utils . checkState ( null != lastCommittedOffset , "Last committed offset cannot be null" ) ; controlChannel . consumerCommit ( offset ) ; lastCommittedOffset = offset ;


signature:
public class Consumer { /** * Send a control message indicating the consumer has encountered an error . */
public void error ( Throwable throwable ) { } }

implementation: if ( consumerError == null ) { consumerError = throwable ; controlChannel . consumerError ( throwable ) ; }


signature:
public class FuzzyMatch { /** * t0 = [ SORTED _ INTERSECTION ]
 * t1 = [ SORTED _ INTERSECTION ] + [ SORTED _ REST _ OF _ STRING1]
 * t2 = [ SORTED _ INTERSECTION ] + [ SORTED _ REST _ OF _ STRING2]
 * outcome = max ( t0 , t1 , t2) */
public static int getRatio ( String s1 , String s2 ) { } }

implementation:
if ( s1 . length ( ) >= s2 . length ( ) ) { // We need to swap s1 and s2
String temp = s2 ; s2 = s1 ; s1 = temp ; } // Get alpha numeric characters
Set < String > set1 = tokenizeString ( escapeString ( s1 ) ) ; Set < String > set2 = tokenizeString ( escapeString ( s2 ) ) ; SetView < String > intersection = Sets . intersection ( set1 , set2 ) ; TreeSet < String > sortedIntersection = Sets . newTreeSet ( intersection ) ; if ( LOG . isTraceEnabled ( ) ) { StringBuilder sortedSb = new StringBuilder ( ) ; for ( String s : sortedIntersection ) { sortedSb . append ( s ) . append ( " " ) ; } LOG . trace ( "Sorted intersection --> {}" , sortedSb . toString ( ) ) ; } // Find out difference of sets set1 and intersection of set1 , set2
SetView < String > restOfSet1 = Sets . symmetricDifference ( set1 , intersection ) ; // Sort it
TreeSet < String > sortedRestOfSet1 = Sets . newTreeSet ( restOfSet1 ) ; SetView < String > restOfSet2 = Sets . symmetricDifference ( set2 , intersection ) ; TreeSet < String > sortedRestOfSet2 = Sets . newTreeSet ( restOfSet2 ) ; if ( LOG . isTraceEnabled ( ) ) { StringBuilder sb1 = new StringBuilder ( ) ; for ( String s : sortedRestOfSet1 ) { sb1 . append ( s ) . append ( " " ) ; } LOG . trace ( "Sorted rest of 1 --> {}" , sb1 . toString ( ) ) ; StringBuilder sb2 = new StringBuilder ( ) ; for ( String s : sortedRestOfSet1 ) { sb2 . append ( s ) . append ( " " ) ; } LOG . trace ( "Sorted rest of 2 --> {}" , sb2 . toString ( ) ) ; } StringBuilder t0Builder = new StringBuilder ( "" ) ; StringBuilder t1Builder = new StringBuilder ( "" ) ; StringBuilder t2Builder = new StringBuilder ( "" ) ; for ( String s : sortedIntersection ) { t0Builder . append ( " " ) . append ( s ) ; } String t0 = t0Builder . toString ( ) . trim ( ) ; Set < String > setT1 = Sets . union ( sortedIntersection , sortedRestOfSet1 ) ; for ( String s : setT1 ) { t1Builder . append ( " " ) . append ( s ) ; } String t1 = t1Builder . toString ( ) . trim ( ) ; Set < String > setT2 = Sets . union ( intersection , sortedRestOfSet2 ) ; for ( String s : setT2 ) { t2Builder . append ( " " ) . append ( s ) ; } String t2 = t2Builder . toString ( ) . trim ( ) ; int amt1 = calculateLevenshteinDistance ( t0 , t1 ) ; int amt2 = calculateLevenshteinDistance ( t0 , t2 ) ; int amt3 = calculateLevenshteinDistance ( t1 , t2 ) ; LOG . trace ( "t0 = {} --> {}" , t0 , amt1 ) ; LOG . trace ( "t1 = {} --> {}" , t1 , amt2 ) ; LOG . trace ( "t2 = {} --> {}" , t2 , amt3 ) ; return Math . max ( Math . max ( amt1 , amt2 ) , amt3 ) ;


signature:
public class DataLakeDataGenerator { /** * Private Methods */
private void close ( boolean idleClosed ) throws IOException , StageException { } }

implementation:
closeLock . writeLock ( ) . lock ( ) ; LOG . debug ( "Path[{}] - Closing" , filePath ) ; try { throwIfIdleClosed ( ) ; // If this was closed previously , just return
if ( isClosed ( ) ) { return ; } if ( generator != null ) { generator . close ( ) ; } this . idleClosed = idleClosed ; // writers can never be null , except in tests
if ( idleClosed && outputStreamHelper != null ) { // writers . release ( this , false ) ;
outputStreamHelper . commitFile ( filePath ) ; } } finally { closeLock . writeLock ( ) . unlock ( ) ; // Gracefully Shutdown the thread , so rename goes through without glitch .
if ( ! idleClosed ) { idleCloseExecutor . shutdown ( ) ; } }


signature:
public class DetachedStage { /** * Create a new instance of a stage that does not directly live in the pipeline canvas . */
public < S > DetachedStageRuntime < ? extends S > createDetachedStage ( String jsonDefinition , StageLibraryTask stageLibrary , String pipelineId , String pipelineTitle , String rev , Stage . UserContext userContext , MetricRegistry metrics , ExecutionMode executionMode , DeliveryGuarantee deliveryGuarantee , RuntimeInfo runtimeInfo , EmailSender emailSender , Configuration configuration , long startTime , LineagePublisherDelegator lineagePublisherDelegator , Class < S > klass , List < Issue > errors ) { } }

implementation: DetachedStageConfiguration stageConf ; try { ObjectMapper objectMapper = ObjectMapperFactory . get ( ) ; DetachedStageConfigurationJson stageConfJson = objectMapper . readValue ( jsonDefinition , DetachedStageConfigurationJson . class ) ; stageConf = stageConfJson . getDetachedStageConfiguration ( ) ; } catch ( IOException e ) { LOG . error ( CreationError . CREATION_0900 . getMessage ( ) , e . toString ( ) , e ) ; errors . add ( IssueCreator . getPipeline ( ) . create ( CreationError . CREATION_0900 , e . toString ( ) ) ) ; return null ; } return createDetachedStage ( stageConf , stageLibrary , pipelineId , pipelineTitle , rev , userContext , metrics , executionMode , deliveryGuarantee , runtimeInfo , emailSender , configuration , startTime , lineagePublisherDelegator , klass , errors ) ;


signature:
public class DetachedStage { /** * Create a new instance of a stage that does not directly live in the pipeline canvas . */
public < S > DetachedStageRuntime < ? extends S > createDetachedStage ( DetachedStageConfiguration stageConf , StageLibraryTask stageLibrary , String pipelineId , String pipelineTitle , String rev , Stage . UserContext userContext , MetricRegistry metrics , ExecutionMode executionMode , DeliveryGuarantee deliveryGuarantee , RuntimeInfo runtimeInfo , EmailSender emailSender , Configuration configuration , long startTime , LineagePublisherDelegator lineagePublisherDelegator , Class < S > klass , List < Issue > errors ) { } }

implementation:
// Firstly validate that the configuration is correct and up to date
DetachedStageValidator validator = new DetachedStageValidator ( stageLibrary , stageConf ) ; DetachedStageConfiguration detachedStageConfiguration = validator . validate ( ) ; // If the stage is not valid , we can ' t create instance of it
if ( detachedStageConfiguration . getIssues ( ) . hasIssues ( ) ) { errors . addAll ( detachedStageConfiguration . getIssues ( ) . getIssues ( ) ) ; return null ; } // Then stageBean that will create new instance and properly propagate all the
StageBean stageBean = PipelineBeanCreator . get ( ) . createStageBean ( true , stageLibrary , stageConf . getStageConfiguration ( ) , false , false , false , Collections . emptyMap ( ) , null , errors ) ; if ( ! errors . isEmpty ( ) ) { return null ; } // Stage . Info and Stage . Context
Stage . Info stageInfo = new Stage . Info ( ) { @ Override public String getName ( ) { return stageBean . getDefinition ( ) . getName ( ) ; } @ Override public int getVersion ( ) { return stageBean . getDefinition ( ) . getVersion ( ) ; } @ Override public String getInstanceName ( ) { return stageBean . getConfiguration ( ) . getInstanceName ( ) ; } @ Override public String getLabel ( ) { return stageBean . getConfiguration ( ) . getInstanceName ( ) ; } } ; StageContext context = new StageContext ( pipelineId , pipelineTitle , null , // TODO . Will need to set here if this stage needs to publish lineage events
rev , null , // TODO . Will need to set here if this stage needs to publish lineage events
Collections . emptyList ( ) , userContext , stageBean . getDefinition ( ) . getType ( ) , 0 , false , metrics , stageBean . getDefinition ( ) . getConfigDefinitions ( ) , stageBean . getSystemConfigs ( ) . stageOnRecordError , Collections . emptyList ( ) , Collections . emptyMap ( ) , stageInfo , executionMode , deliveryGuarantee , runtimeInfo , emailSender , configuration , Collections . emptyMap ( ) , startTime , lineagePublisherDelegator , Collections . emptyMap ( ) , false ) ; return DetachedStageRuntime . create ( stageBean , stageInfo , context , klass ) ;


signature:
public class ELEvaluator { /** * stuff a function into the namespace / name dual level map structure */
private void registerFunction ( ElFunctionDefinition function ) { } }

implementation: String namespace ; String functionName ; if ( function . getName ( ) . contains ( ":" ) ) { String [ ] tokens = function . getName ( ) . split ( ":" ) ; namespace = tokens [ 0 ] ; functionName = tokens [ 1 ] ; } else { namespace = "" ; functionName = function . getName ( ) ; } Map < String , Method > namespaceFunctions = functionsByNamespace . get ( namespace ) ; if ( namespaceFunctions == null ) { namespaceFunctions = new HashMap < > ( ) ; functionsByNamespace . put ( namespace , namespaceFunctions ) ; } namespaceFunctions . put ( functionName , function . getMethod ( ) ) ;


signature:
public class ConfigInjector { /** * Inject config values to given Stage .
 * @ param stage Stage instance
 * @ param stageDef Definition for given stage
 * @ param stageConf Actual configuration values
 * @ param constants Pipeline constants ( parameters )
 * @ param issues List into which issues will be added */
public void injectStage ( Object stage , StageDefinition stageDef , StageConfiguration stageConf , Map < String , Object > constants , List < Issue > issues ) { } }

implementation: injectConfigsToObject ( stage , new StageInjectorContext ( stageDef , stageConf , constants , issues ) ) ;


signature:
public class StageContext { /** * Processor . Context */
@ Override public Record createRecord ( Record originatorRecord ) { } }

implementation: Preconditions . checkNotNull ( originatorRecord , "originatorRecord cannot be null" ) ; RecordImpl record = new RecordImpl ( stageInfo . getInstanceName ( ) , originatorRecord , null , null ) ; HeaderImpl header = record . getHeader ( ) ; header . setStagesPath ( "" ) ; return record ;


signature:
public class StageContext { /** * Processor . Context */
@ Override public Record createRecord ( Record originatorRecord , String sourceIdPostfix ) { } }

implementation: Preconditions . checkNotNull ( originatorRecord , "originatorRecord cannot be null" ) ; RecordImpl record = new RecordImpl ( stageInfo . getInstanceName ( ) , originatorRecord , null , null ) ; HeaderImpl header = record . getHeader ( ) ; header . setSourceId ( header . getSourceId ( ) + "_" + sourceIdPostfix ) ; header . setStagesPath ( "" ) ; return record ;


signature:
public class StageContext { /** * Processor . Context */
@ Override public Record createRecord ( Record originatorRecord , byte [ ] raw , String rawMime ) { } }

implementation: return new RecordImpl ( stageInfo . getInstanceName ( ) , originatorRecord , raw , rawMime ) ;


signature:
public class StageContext { /** * Processor . Context */
@ Override public Record cloneRecord ( Record record ) { } }

implementation: RecordImpl clonedRecord = ( ( RecordImpl ) record ) . clone ( ) ; HeaderImpl header = clonedRecord . getHeader ( ) ; header . setStagesPath ( "" ) ; return clonedRecord ;


signature:
public class StageContext { /** * Processor . Context */
@ Override public Record cloneRecord ( Record record , String sourceIdPostfix ) { } }

implementation: RecordImpl clonedRecord = ( ( RecordImpl ) record ) . clone ( ) ; HeaderImpl header = clonedRecord . getHeader ( ) ; header . setSourceId ( header . getSourceId ( ) + "_" + sourceIdPostfix ) ; header . setStagesPath ( "" ) ; return clonedRecord ;


signature:
public class HMSCache { /** * Returns if the { @ link HMSCache } has the corresponding { @ link HMSCacheSupport . HMSCacheInfo } and qualified table name .
 * @ param hmsCacheType { @ link HMSCacheType }
 * @ param qualifiedTableName qualified table name
 * @ param < T > { @ link HMSCacheSupport . HMSCacheInfo }
 * @ return Corresponding { @ link HMSCacheSupport . HMSCacheInfo } for the qualified table name .
 * @ throws StageException if the { @ link HMSCacheType } is not supported by { @ link HMSCache } */
@ SuppressWarnings ( "unchecked" ) public < T extends HMSCacheSupport . HMSCacheInfo > T getIfPresent ( HMSCacheType hmsCacheType , String qualifiedTableName ) throws StageException { } }

implementation: if ( ! cacheMap . containsKey ( hmsCacheType ) ) { throw new StageException ( Errors . HIVE_16 , hmsCacheType ) ; } Optional < HMSCacheSupport . HMSCacheInfo > ret = cacheMap . get ( hmsCacheType ) . getIfPresent ( qualifiedTableName ) ; return ret == null ? null : ( T ) ret . orNull ( ) ;


signature:
public class HMSCache { /** * Returns if the { @ link HMSCache } has the corresponding { @ link HMSCacheSupport . HMSCacheInfo }
 * and qualified table name . This method is safe to call from multiple threads - it will block and serialize
 * multiple readers . It guarantees that each value will be loaded at most once to limit load on HS2 as much
 * as possible .
 * If it is not there load it using corresponding { @ link HMSCacheSupport . HMSCacheLoader }
 * @ param < T > { @ link HMSCacheSupport . HMSCacheInfo }
 * @ param hmsCacheType { @ link HMSCacheType }
 * @ param qualifiedTableName qualified table name
 * @ return Corresponding { @ link HMSCacheSupport . HMSCacheInfo } for the qualified table name .
 * @ throws StageException if the { @ link HMSCacheType } is not supported by { @ link HMSCache } */
@ SuppressWarnings ( "unchecked" ) public < T extends HMSCacheSupport . HMSCacheInfo > T getOrLoad ( HMSCacheType hmsCacheType , String qualifiedTableName , HiveQueryExecutor queryExecutor ) throws StageException { } }

implementation:
if ( ! cacheMap . containsKey ( hmsCacheType ) ) { throw new StageException ( Errors . HIVE_16 , hmsCacheType ) ; } // Firstly validate if the data already exists to avoid locking
T cacheValue = getIfPresent ( hmsCacheType , qualifiedTableName ) ; if ( cacheValue != null ) { return cacheValue ; } // For altering operation , get exclusive lock
Lock lock = null ; try { lock = tableLocks . get ( keyForLockMap ( hmsCacheType , qualifiedTableName ) ) ; lock . lock ( ) ; // Check the presence again as another thread could load the value before we got the lock
cacheValue = getIfPresent ( hmsCacheType , qualifiedTableName ) ; if ( cacheValue != null ) { return cacheValue ; } // Load the value from Hive
return ( T ) ( cacheMap . get ( hmsCacheType ) . get ( qualifiedTableName , ( ) -> hmsCacheType . getSupport ( ) . newHMSCacheLoader ( queryExecutor ) . load ( qualifiedTableName ) ) ) . orNull ( ) ; } catch ( ExecutionException e ) { throw new StageException ( Errors . HIVE_01 , e ) ; } finally { if ( lock != null ) { lock . unlock ( ) ; } }


signature:
public class HMSCache { /** * Puts / updates the { @ link HMSCache } with { @ link HMSCacheSupport . HMSCacheInfo } for corresponding
 * { @ link HMSCacheType } and qualified table name .
 * This method skips any locking provided by getOrLoad ( ) and hence the caller code should not mix getOrLoad ( ) and
 * put ( ) calls in the same code .
 * @ param cacheType { @ link HMSCacheType }
 * @ param qualifiedTableName qualified table name
 * @ param hmsCacheInfo { @ link HMSCacheSupport . HMSCacheInfo }
 * @ param < T > { @ link HMSCacheSupport . HMSCacheInfo }
 * @ throws StageException if the { @ link HMSCacheType } is not supported by { @ link HMSCache } */
public < T extends HMSCacheSupport . HMSCacheInfo > void put ( HMSCacheType cacheType , String qualifiedTableName , T hmsCacheInfo ) throws StageException { } }

implementation: if ( ! cacheMap . containsKey ( cacheType ) ) { throw new StageException ( Errors . HIVE_16 , cacheType ) ; } cacheMap . get ( cacheType ) . put ( qualifiedTableName , Optional . of ( hmsCacheInfo ) ) ;


signature:
public class DefinitionsApi { /** * Returns pipeline & amp ; stage configuration definitions
 * This will fetch defintions based on the hideStage filter
 * @ return DefinitionsJson */
public DefinitionsJson getDefinitions ( HideStage . Type hideStage ) throws ApiException { } }

implementation:
Object postBody = null ; byte [ ] postBinaryBody = null ; // create path and map variables
String path = "/v1/definitions" . replaceAll ( "\\{format\\}" , "json" ) ; // query params
List < Pair > queryParams = new ArrayList < Pair > ( ) ; if ( hideStage != null ) { queryParams . add ( new Pair ( "hideStage" , hideStage . name ( ) ) ) ; } Map < String , String > headerParams = new HashMap < String , String > ( ) ; Map < String , Object > formParams = new HashMap < String , Object > ( ) ; final String [ ] accepts = { "application/json" } ; final String accept = apiClient . selectHeaderAccept ( accepts ) ; final String [ ] contentTypes = { } ; final String contentType = apiClient . selectHeaderContentType ( contentTypes ) ; String [ ] authNames = new String [ ] { "basic" } ; TypeRef returnType = new TypeRef < DefinitionsJson > ( ) { } ; return apiClient . invokeAPI ( path , "GET" , queryParams , postBody , postBinaryBody , headerParams , formParams , accept , contentType , authNames , returnType ) ;


signature:
public class LineagePublisherTaskImpl { /** * Parse given configuration declaration of lineage plugin and return appropriate definition .
 * This method will throw exceptions on all error paths . */
private LineagePublisherDefinition getDefinition ( String name ) { } }

implementation:
String defConfig = LineagePublisherConstants . configDef ( name ) ; String publisherDefinition = configuration . get ( defConfig , null ) ; if ( StringUtils . isEmpty ( publisherDefinition ) ) { throw new IllegalArgumentException ( Utils . format ( "Missing definition '{}'" , defConfig ) ) ; } String [ ] lineagePluginDefs = publisherDefinition . split ( "::" ) ; if ( lineagePluginDefs . length != 2 ) { throw new IllegalStateException ( Utils . format ( "Invalid definition '{}', expected $libraryName::$publisherName" , publisherDefinition ) ) ; } LineagePublisherDefinition def = stageLibraryTask . getLineagePublisherDefinition ( lineagePluginDefs [ 0 ] , // Library
lineagePluginDefs [ 1 ] // Plugin name
) ; if ( def == null ) { throw new IllegalStateException ( Utils . format ( "Can't find publisher '{}'" , publisherDefinition ) ) ; } return def ;


signature:
public class SQLParserUtils { /** * Format column names based on whether they are case - sensitive */
private static String formatName ( String columnName , boolean caseSensitive ) { } }

implementation: String returnValue = format ( columnName ) ; if ( caseSensitive ) { return returnValue ; } return returnValue . toUpperCase ( ) ;


signature:
public class SQLParserUtils { /** * Unescapes strings and returns them . */
private static String formatValue ( String value ) { } }

implementation:
// The value can either be null ( if the IS keyword is present before it or just a NULL string with no quotes )
if ( value == null || NULL_STRING . equalsIgnoreCase ( value ) ) { return null ; } String returnValue = format ( value ) ; return returnValue . replaceAll ( "''" , "'" ) ;


signature:
public class StoreApi { /** * Find Pipeline Configuration by name and revision
 * @ param pipelineId
 * @ param rev
 * @ param get
 * @ param attachment
 * @ return PipelineConfigurationJson */
public PipelineConfigurationJson getPipelineInfo ( String pipelineId , String rev , String get , Boolean attachment ) throws ApiException { } }

implementation:
Object postBody = null ; byte [ ] postBinaryBody = null ; // verify the required parameter ' pipelineId ' is set
if ( pipelineId == null ) { throw new ApiException ( 400 , "Missing the required parameter 'pipelineId' when calling getPipelineInfo" ) ; } // create path and map variables
String path = "/v1/pipeline/{pipelineId}" . replaceAll ( "\\{format\\}" , "json" ) . replaceAll ( "\\{" + "pipelineId" + "\\}" , apiClient . escapeString ( pipelineId . toString ( ) ) ) ; // query params
List < Pair > queryParams = new ArrayList < Pair > ( ) ; Map < String , String > headerParams = new HashMap < String , String > ( ) ; Map < String , Object > formParams = new HashMap < String , Object > ( ) ; queryParams . addAll ( apiClient . parameterToPairs ( "" , "rev" , rev ) ) ; queryParams . addAll ( apiClient . parameterToPairs ( "" , "get" , get ) ) ; queryParams . addAll ( apiClient . parameterToPairs ( "" , "attachment" , attachment ) ) ; final String [ ] accepts = { "application/json" } ; final String accept = apiClient . selectHeaderAccept ( accepts ) ; final String [ ] contentTypes = { } ; final String contentType = apiClient . selectHeaderContentType ( contentTypes ) ; String [ ] authNames = new String [ ] { "basic" } ; TypeRef returnType = new TypeRef < PipelineConfigurationJson > ( ) { } ; return apiClient . invokeAPI ( path , "GET" , queryParams , postBody , postBinaryBody , headerParams , formParams , accept , contentType , authNames , returnType ) ;


signature:
public class StoreApi { /** * Add a new Pipeline Fragment Configuration to the store
 * @ param fragmentId Frgament Id
 * @ param description Description
 * @ return PipelineFragmentEnvelopeJson */
public PipelineFragmentEnvelopeJson createDraftPipelineFragment ( String fragmentId , String description , List < StageConfigurationJson > stageInstances ) throws ApiException { } }

implementation:
Object postBody = stageInstances ; byte [ ] postBinaryBody = null ; // verify the required parameter ' pipelineId ' is set
if ( fragmentId == null ) { throw new ApiException ( 400 , "Missing the required parameter 'fragmentId' when calling createPipelineFragment" ) ; } // create path and map variables
String path = "/v1/fragment/{fragmentId}" . replaceAll ( "\\{format\\}" , "json" ) . replaceAll ( "\\{" + "fragmentId" + "\\}" , apiClient . escapeString ( fragmentId . toString ( ) ) ) ; // query params
List < Pair > queryParams = new ArrayList < Pair > ( ) ; Map < String , String > headerParams = new HashMap < String , String > ( ) ; Map < String , Object > formParams = new HashMap < String , Object > ( ) ; queryParams . addAll ( apiClient . parameterToPairs ( "" , "description" , description ) ) ; queryParams . addAll ( apiClient . parameterToPairs ( "" , "draft" , true ) ) ; final String [ ] accepts = { "application/json" } ; final String accept = apiClient . selectHeaderAccept ( accepts ) ; final String [ ] contentTypes = { } ; final String contentType = apiClient . selectHeaderContentType ( contentTypes ) ; String [ ] authNames = new String [ ] { "basic" } ; TypeRef returnType = new TypeRef < PipelineFragmentEnvelopeJson > ( ) { } ; return apiClient . invokeAPI ( path , "PUT" , queryParams , postBody , postBinaryBody , headerParams , formParams , accept , contentType , authNames , returnType ) ;


signature:
public class StoreApi { /** * Returns all Pipeline Configuration Info
 * @ return List < PipelineInfoJson > */
public List < PipelineInfoJson > getPipelines ( String filterText , String label , int offset , int len , PipelineOrderByFields orderBy , Order order , boolean includeStatus ) throws ApiException { } }

implementation:
Object postBody = null ; byte [ ] postBinaryBody = null ; // create path and map variables
String path = "/v1/pipelines" . replaceAll ( "\\{format\\}" , "json" ) ; // query params
List < Pair > queryParams = new ArrayList < Pair > ( ) ; Map < String , String > headerParams = new HashMap < String , String > ( ) ; Map < String , Object > formParams = new HashMap < String , Object > ( ) ; queryParams . addAll ( apiClient . parameterToPairs ( "" , "filterText" , filterText ) ) ; queryParams . addAll ( apiClient . parameterToPairs ( "" , "label" , label ) ) ; queryParams . addAll ( apiClient . parameterToPairs ( "" , "offset" , offset ) ) ; queryParams . addAll ( apiClient . parameterToPairs ( "" , "len" , len ) ) ; queryParams . addAll ( apiClient . parameterToPairs ( "" , "orderBy" , orderBy ) ) ; queryParams . addAll ( apiClient . parameterToPairs ( "" , "order" , order ) ) ; final String [ ] accepts = { "application/json" } ; final String accept = apiClient . selectHeaderAccept ( accepts ) ; final String [ ] contentTypes = { } ; final String contentType = apiClient . selectHeaderContentType ( contentTypes ) ; String [ ] authNames = new String [ ] { "basic" } ; TypeRef returnType = new TypeRef < List < PipelineInfoJson > > ( ) { } ; return apiClient . invokeAPI ( path , "GET" , queryParams , postBody , postBinaryBody , headerParams , formParams , accept , contentType , authNames , returnType ) ;


signature:
public class StoreApi { /** * Import Pipeline Fragment Configuration & Rules
 * @ param fragmentId Fragment Id
 * @ return fragmentEnvelope */
public PipelineFragmentEnvelopeJson importPipelineFragment ( String fragmentId , boolean draft , boolean includeLibraryDefinitions , PipelineFragmentEnvelopeJson fragmentEnvelope ) throws ApiException { } }

implementation:
Object postBody = fragmentEnvelope ; byte [ ] postBinaryBody = null ; // verify the required parameter ' fragmentId ' is set
if ( fragmentId == null ) { throw new ApiException ( 400 , "Missing the required parameter 'fragmentId' when calling importPipelineFragment" ) ; } // verify the required parameter ' fragmentEnvelope ' is set
if ( fragmentEnvelope == null ) { throw new ApiException ( 400 , "Missing the required parameter 'pipelineEnvelope' when calling importPipelineFragment" ) ; } // create path and map variables
String path = "/v1/fragment/{fragmentId}/import" . replaceAll ( "\\{format\\}" , "json" ) . replaceAll ( "\\{" + "fragmentId" + "\\}" , apiClient . escapeString ( fragmentId . toString ( ) ) ) ; // query params
List < Pair > queryParams = new ArrayList < Pair > ( ) ; Map < String , String > headerParams = new HashMap < String , String > ( ) ; Map < String , Object > formParams = new HashMap < String , Object > ( ) ; queryParams . addAll ( apiClient . parameterToPairs ( "" , "draft" , draft ) ) ; queryParams . addAll ( apiClient . parameterToPairs ( "" , "includeLibraryDefinitions" , includeLibraryDefinitions ) ) ; final String [ ] accepts = { "application/json" } ; final String accept = apiClient . selectHeaderAccept ( accepts ) ; final String [ ] contentTypes = { "application/json" } ; final String contentType = apiClient . selectHeaderContentType ( contentTypes ) ; String [ ] authNames = new String [ ] { "basic" } ; TypeRef returnType = new TypeRef < PipelineFragmentEnvelopeJson > ( ) { } ; return apiClient . invokeAPI ( path , "POST" , queryParams , postBody , postBinaryBody , headerParams , formParams , accept , contentType , authNames , returnType ) ;


signature:
public class ActiveRecordWriters { /** * and the hdfsTarget ( in the pipeline runnable thread ) , calls flushAll */
public synchronized void release ( RecordWriter writer , boolean roll ) throws StageException , IOException { } }

implementation: writer . closeLock ( ) ; try { if ( roll || writer . isIdleClosed ( ) || manager . isOverThresholds ( writer ) ) { if ( IS_TRACE_ENABLED ) { LOG . trace ( "Release '{}'" , writer . getPath ( ) ) ; } writers . remove ( writer . getPath ( ) . toString ( ) ) ; manager . commitWriter ( writer ) ; } } finally { writer . closeUnlock ( ) ; } purge ( ) ;


signature:
public class DataLakeGeneratorManager { /** * Returns the temp file path to write records to
 * @ param dirPathTemplate Directory Template
 * @ param record Record for which file path is calculated
 * @ param recordTime Record time
 * @ return Temporary File Path to write records to
 * @ throws StageException */
public String getFilePath ( String dirPathTemplate , Record record , Date recordTime ) throws StageException { } }

implementation:
String dirPath ; // get directory path
if ( dirPathTemplateInHeader ) { dirPath = record . getHeader ( ) . getAttribute ( DataLakeTarget . TARGET_DIRECTORY_HEADER ) ; Utils . checkArgument ( ! ( dirPath == null || dirPath . isEmpty ( ) ) , "Directory Path cannot be null" ) ; } else { dirPath = resolvePath ( dirPathTemplateEval , dirPathTemplateVars , dirPathTemplate , recordTime , record ) ; } // SDC - 5492 : replace " / / " to " / " in file path
dirPath = dirPath . replaceAll ( "/+" , "/" ) ; if ( dirPath . endsWith ( "/" ) ) { dirPath = dirPath . substring ( 0 , dirPath . length ( ) - 1 ) ; } return outputStreamHelper . getTempFilePath ( dirPath , record , recordTime ) ;


signature:
public class DataLakeGeneratorManager { /** * Close all generators ( and underlying streams / files )
 * @ throws IOException
 * @ throws StageException */
public void closeAll ( ) throws IOException , StageException { } }

implementation: Set < String > filePathsToClose = ImmutableSet . copyOf ( tmpFilePathToGenerators . keySet ( ) ) ; for ( String filePath : filePathsToClose ) { close ( filePath ) ; }


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

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


signature:
public class JdbcRecordReader { /** * Get the numeric operation code from record header . The default code is
 * used if the operation code is not found in the header .
 * @ param record the record to find the operation code
 * @ param defaultOp the default operation
 * @ param unsupportedAction the action to take for unsupported code
 * @ param errorRecords the list to take error records
 * @ return the numeric operation code or - 1 for unsupported operation */
@ VisibleForTesting final int getOperationFromRecord ( Record record , JDBCOperationType defaultOp , UnsupportedOperationAction unsupportedAction , List < OnRecordErrorException > errorRecords ) { } }

implementation: return getOperationFromRecord ( record , defaultOp . getCode ( ) , unsupportedAction , errorRecords ) ;


signature:
public class JdbcRecordReader { /** * Get the numeric operation code from record header . The default code is
 * used if the operation code is not found in the header . This can be
 * overwritten in inherited classes .
 * @ param record the record to find the operation code
 * @ param defaultOpCode the default operation code
 * @ param unsupportedAction the action to take for unsupported code
 * @ param errorRecords the list to take error records
 * @ return the numeric operation code or - 1 for unsupported operation */
int getOperationFromRecord ( Record record , int defaultOpCode , UnsupportedOperationAction unsupportedAction , List < OnRecordErrorException > errorRecords ) { } }

implementation:
String op = record . getHeader ( ) . getAttribute ( OperationType . SDC_OPERATION_TYPE ) ; int opCode = - 1 ; // unsupported
if ( Strings . isNullOrEmpty ( op ) ) { return defaultOpCode ; } // Check if the operation code from header attribute is valid
try { opCode = JDBCOperationType . convertToIntCode ( op ) ; } catch ( NumberFormatException | UnsupportedOperationException ex ) { LOG . debug ( "Operation obtained from record is not supported. Handle by UnsupportedOperationAction {}. {}" , unsupportedAction . getLabel ( ) , ex ) ; switch ( unsupportedAction ) { case SEND_TO_ERROR : LOG . debug ( "Sending record to error due to unsupported operation {}" , op ) ; errorRecords . add ( new OnRecordErrorException ( record , JdbcErrors . JDBC_70 , op ) ) ; break ; case USE_DEFAULT : opCode = defaultOpCode ; break ; case DISCARD : default : // unknown action
LOG . debug ( "Discarding record with unsupported operation {}" , op ) ; } } return opCode ;


signature:
public class JdbcRecordReader { /** * Iterate the columnsToField map and check if the record has all the necessary columns .
 * The returned SortedMap contains columnName - param mapping which is only contained in this record .
 * For example , if a table in destination DB has a column A but record doesn ' t have a field named A ,
 * the column A is not included in the returning SortedMap .
 * @ param record Record to check
 * @ param op Operation code . Used mainly in subclass
 * @ param parameters columnName - to - value mapping
 * @ param columnsToFields columnName - to - fieldPath mapping
 * @ return */
@ VisibleForTesting SortedMap < String , String > getColumnsToParameters ( final Record record , int op , Map < String , String > parameters , Map < String , String > columnsToFields ) { } }

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


signature:
public class JdbcRecordReader { /** * This function simply returns field path in record for the corresponding column name .
 * This is needed because records generated by CDC origins store data in different location
 * for different operation . Subclasses will override this function and implement special handling .
 * @ param columnName column name
 * @ param columnsToField mapping of column name to field path
 * @ param op Not used here but is used in subclass .
 * @ return */
String getFieldPath ( String columnName , Map < String , String > columnsToField , int op ) { } }

implementation: return columnsToField . get ( columnName ) ;


signature:
public class TopologicalSorter { /** * Sorts the vertices in the topological order using { @ link # tieComparator } if needed to break ties .
 * Throws { @ link IllegalStateException } if there is a cycle detected .
 * @ return Topological sorted order for the @ { @ link DirectedGraph } */
public SortedSet < V > sort ( ) { } }

implementation:
Utils . checkState ( ! new CycleDetector < > ( directedGraph ) . isGraphCyclic ( ) , "Cycles found in the graph" ) ; final Map < V , Integer > vertexToSortedNumber = new HashMap < > ( ) ; Map < V , Integer > inEdgesCount = new TreeMap < > ( ) ; SortedSet < V > sortedSet = new TreeSet < > ( ( o1 , o2 ) -> { Integer sortedNumber1 = vertexToSortedNumber . get ( o1 ) ; Integer sortedNumber2 = vertexToSortedNumber . get ( o2 ) ; if ( sortedNumber1 . intValue ( ) == sortedNumber2 . intValue ( ) ) { // If there is no tie comparator and there is a tie , arrange o1 before o2.
return ( tieComparator != null ) ? tieComparator . compare ( o1 , o2 ) : - 1 ; } return sortedNumber1 . compareTo ( sortedNumber2 ) ; } ) ; final AtomicInteger startNumber = new AtomicInteger ( 1 ) ; directedGraph . vertices ( ) . forEach ( vertex -> { Collection < V > inwardVertices = directedGraph . getInwardEdgeVertices ( vertex ) ; inEdgesCount . put ( vertex , inwardVertices . size ( ) ) ; } ) ; while ( ! inEdgesCount . isEmpty ( ) ) { Set < V > nextVertices = nextVerticesForProcessing ( inEdgesCount ) ; nextVertices . forEach ( vertexForProcessing -> { inEdgesCount . remove ( vertexForProcessing ) ; updateCounts ( vertexForProcessing , inEdgesCount ) ; vertexToSortedNumber . put ( vertexForProcessing , startNumber . getAndIncrement ( ) ) ; } ) ; } sortedSet . addAll ( vertexToSortedNumber . keySet ( ) ) ; return sortedSet ;


signature:
public class EventBuffer { /** * Read next event from buffer with respect to maximum timeout .
 * @ param timeout timeout .
 * @ param unit timeout time unit .
 * @ return next event of null
 * @ throws StageException */
public EnrichedEvent poll ( long timeout , TimeUnit unit ) throws StageException { } }

implementation: try { return queue . poll ( timeout , unit ) ; } catch ( InterruptedException e ) { LOG . error ( Errors . MYSQL_001 . getMessage ( ) , e . toString ( ) , e ) ; Thread . currentThread ( ) . interrupt ( ) ; throw new StageException ( Errors . MYSQL_001 , e . toString ( ) , e ) ; }


signature:
public class HadoopMapReduceBinding { /** * visible for testing ( can ' t annotate as can ' t depend on Guava ) */
public static Integer getMapMemoryMb ( String javaOpts , Configuration conf ) { } }

implementation:
String [ ] javaOptsArray = javaOpts . split ( " " ) ; Integer upperLimitMemory = null ; for ( String opts : javaOptsArray ) { if ( opts . contains ( "-Xmx" ) ) { Integer memoryMb = Integer . valueOf ( opts . substring ( 4 , opts . length ( ) - 1 ) ) ; switch ( opts . charAt ( opts . length ( ) - 1 ) ) { case 'm' : case 'M' : break ; case 'k' : case 'K' : memoryMb = memoryMb / ( 1024 ) ; break ; case 'g' : case 'G' : memoryMb = memoryMb * 1024 ; break ; default : memoryMb = Integer . valueOf ( opts . substring ( 4 , opts . length ( ) ) ) / ( 1024 * 1024 ) ; break ; } // Add 25 % to Java heap as MAP _ MEMORY _ MB is the total physical memory for the map task
upperLimitMemory = ( ( int ) ( memoryMb * 0.25 ) ) + memoryMb ; // dont break as there could be multiple - Xmx , we need to honor the last
} } if ( upperLimitMemory != null ) { String defaultMapMemoryString = conf . get ( MAPREDUCE_MAP_MEMORY_MB ) ; if ( defaultMapMemoryString != null ) { Integer defaultMapMemory = Integer . valueOf ( defaultMapMemoryString ) ; upperLimitMemory = ( upperLimitMemory > defaultMapMemory ? upperLimitMemory : defaultMapMemory ) ; } } return upperLimitMemory ;


signature:
public class TarFileCreator { /** * Copied from https : / / raw . githubusercontent . com / kamranzafar / jtar / master / src / test / java / org / kamranzafar / jtar / JTarTest . java */
private static void tarFolder ( String parent , String path , TarOutputStream out ) throws IOException { } }

implementation:
BufferedInputStream src = null ; File f = new File ( path ) ; String files [ ] = f . list ( ) ; // is file
if ( files == null ) { files = new String [ 1 ] ; files [ 0 ] = f . getName ( ) ; } parent = ( ( parent == null ) ? ( f . isFile ( ) ) ? "" : f . getName ( ) + "/" : parent + f . getName ( ) + "/" ) ; for ( int i = 0 ; i < files . length ; i ++ ) { File fe = f ; if ( f . isDirectory ( ) ) { fe = new File ( f , files [ i ] ) ; } if ( fe . isDirectory ( ) ) { String [ ] fl = fe . list ( ) ; if ( fl != null && fl . length != 0 ) { tarFolder ( parent , fe . getPath ( ) , out ) ; } else { TarEntry entry = new TarEntry ( fe , parent + files [ i ] + "/" ) ; out . putNextEntry ( entry ) ; } continue ; } FileInputStream fi = new FileInputStream ( fe ) ; src = new BufferedInputStream ( fi ) ; TarEntry entry = new TarEntry ( fe , parent + files [ i ] ) ; out . putNextEntry ( entry ) ; IOUtils . copy ( src , out ) ; src . close ( ) ; out . flush ( ) ; }


signature:
public class RunnerPool { /** * Get exclusive runner for use .
 * @ return Runner that is not being used by anyone else .
 * @ throws PipelineRuntimeException Thrown in case that current thread is unexpectedly interrupted */
public T getRunner ( ) throws PipelineRuntimeException { } }

implementation: validateNotDestroyed ( ) ; try { return queue . take ( ) . runner ; } catch ( InterruptedException e ) { throw new PipelineRuntimeException ( ContainerError . CONTAINER_0801 , e ) ; } finally { runtimeStats . setAvailableRunners ( queue . size ( ) ) ; histogram . update ( queue . size ( ) ) ; }


signature:
public class RunnerPool { /** * Return a runner that haven ' t been used at least for the configured number of milliseconds .
 * @ param idleTime Number of milliseconds when a runner is considered an " idle "
 * @ return First runner that fits such criteria or null if there is no such runner */
public T getIdleRunner ( long idleTime ) { } }

implementation:
// Take the first runner
QueueItem < T > item = queue . poll ( ) ; // All runners might be currently in use , which is fine in this case .
if ( item == null ) { return null ; } // If the runner wasn ' t idle for the expected time , we need to put it back to the queue ( it will be added to the
// begging again ) .
if ( ( System . currentTimeMillis ( ) - item . timestamp ) < idleTime ) { queue . add ( item ) ; return null ; } // Otherwise we do have runner that hasn ' t been used for at least idleTime , so we can return it now
return item . runner ;


signature:
public class RunnerPool { /** * Return given runner back to the pool .
 * @ param runner Runner to be returned */
public void returnRunner ( T runner ) throws PipelineRuntimeException { } }

implementation: validateNotDestroyed ( ) ; queue . add ( new QueueItem < > ( runner ) ) ; runtimeStats . setAvailableRunners ( queue . size ( ) ) ; histogram . update ( queue . size ( ) ) ;


signature:
public class RunnerPool { /** * Destroy only the pool itself - not the individual pipe runners .
 * This method will also validate that all runners were properly returned to the pool . PipelineRuntimeException will
 * be thrown if some runners are still running .
 * @ throws PipelineRuntimeException */
public void destroy ( ) throws PipelineRuntimeException { } }

implementation:
// Firstly set this runner as destroyed
destroyed . set ( true ) ; // Validate that this thread pool have all runners back , otherwise we ' re missing something and that is sign of
// a trouble .
if ( queue . size ( ) < runtimeStats . getTotalRunners ( ) ) { throw new PipelineRuntimeException ( ContainerError . CONTAINER_0802 , queue . size ( ) , runtimeStats . getTotalRunners ( ) ) ; }


signature:
public class RunnerPool { /** * Throw an exception if the runner was already destroyed .
 * @ throws PipelineRuntimeException */
private void validateNotDestroyed ( ) throws PipelineRuntimeException { } }

implementation: if ( destroyed . get ( ) ) { throw new PipelineRuntimeException ( ContainerError . CONTAINER_0803 , queue . size ( ) , runtimeStats . getTotalRunners ( ) ) ; }


signature:
public class KinesisUtil { /** * Checks for existence of the requested stream and adds
 * any configuration issues to the list .
 * @ param conf
 * @ param streamName
 * @ param issues
 * @ param context */
public static long checkStreamExists ( ClientConfiguration awsClientConfig , KinesisConfigBean conf , String streamName , List < Stage . ConfigIssue > issues , Stage . Context context ) { } }

implementation: long numShards = 0 ; try { numShards = getShardCount ( awsClientConfig , conf , streamName ) ; } catch ( AmazonClientException | StageException e ) { LOG . error ( Errors . KINESIS_01 . getMessage ( ) , e . toString ( ) , e ) ; issues . add ( context . createConfigIssue ( Groups . KINESIS . name ( ) , KINESIS_CONFIG_BEAN + ".streamName" , Errors . KINESIS_01 , e . toString ( ) ) ) ; } return numShards ;


signature:
public class KinesisUtil { /** * Get the last shard Id in the given stream
 * In preview mode , kinesis source uses the last Shard Id to get records from kinesis
 * @ param awsClientConfig generic AWS client configuration
 * @ param conf
 * @ param streamName */
public static String getLastShardId ( ClientConfiguration awsClientConfig , KinesisConfigBean conf , String streamName ) throws StageException { } }

implementation: AmazonKinesis kinesisClient = getKinesisClient ( awsClientConfig , conf ) ; String lastShardId = null ; try { StreamDescription description ; do { if ( lastShardId == null ) { description = kinesisClient . describeStream ( streamName ) . getStreamDescription ( ) ; } else { description = kinesisClient . describeStream ( streamName , lastShardId ) . getStreamDescription ( ) ; } int pageSize = description . getShards ( ) . size ( ) ; lastShardId = description . getShards ( ) . get ( pageSize - 1 ) . getShardId ( ) ; } while ( description . getHasMoreShards ( ) ) ; return lastShardId ; } finally { kinesisClient . shutdown ( ) ; }


signature:
public class JdbcMultiRowRecordWriter { /** * Process all records in queue . All records have same operation to same table .
 * Generate a query and set parameters from each record . INSERT and DELETE can be multi - row operation
 * but UPDATE is single - row operation .
 * If maxStatement
 * @ param errorRecords
 * @ param connection
 * @ param maxRowsPerBatch
 * @ param opCode
 * @ param queue
 * @ return
 * @ throws OnRecordErrorException */
private void processQueue ( LinkedList < Record > queue , List < OnRecordErrorException > errorRecords , Connection connection , int maxRowsPerBatch , int opCode ) throws StageException { } }

implementation:
if ( queue . isEmpty ( ) ) { return ; } int rowCount = 0 ; // Assume that columns are all same for the same operation to the same table
// If some columns are missing in record , the record goes to error .
final Record first = queue . getFirst ( ) ; SortedMap < String , String > columnsToParameters = recordReader . getColumnsToParameters ( first , opCode , getColumnsToParameters ( ) , opCode == OperationType . UPDATE_CODE ? getColumnsToFieldNoPK ( ) : getColumnsToFields ( ) ) ; if ( columnsToParameters . isEmpty ( ) ) { // no parameters found for configured columns
if ( LOG . isWarnEnabled ( ) ) { LOG . warn ( "No parameters found for record with ID {}; skipping" , first . getHeader ( ) . getSourceId ( ) ) ; } return ; } String query = generateQueryForMultiRow ( opCode , columnsToParameters , getPrimaryKeyColumns ( ) , // the next batch will have either the max number of records , or however many are left .
Math . min ( maxRowsPerBatch , queue . size ( ) ) ) ; // Need to store removed records from queue , because we might need to add newly generated columns
// to records for Jdbc Tee Processor .
LinkedList < Record > removed = new LinkedList < > ( ) ; try ( PreparedStatement statement = jdbcUtil . getPreparedStatement ( getGeneratedColumnMappings ( ) , query , connection ) ) { int paramIdx = 1 ; // Start processing records in queue . All records have the same operation to the same table .
while ( ! queue . isEmpty ( ) ) { Record r = queue . removeFirst ( ) ; if ( opCode != DELETE_CODE ) { paramIdx = setParamsToStatement ( paramIdx , statement , columnsToParameters , r , connection , opCode ) ; } if ( opCode != OperationType . INSERT_CODE ) { paramIdx = setPrimaryKeys ( paramIdx , r , statement , opCode ) ; } removed . add ( r ) ; ++ rowCount ; if ( rowCount == maxRowsPerBatch ) { // time to execute the current batch
processBatch ( removed , errorRecords , statement , connection ) ; // reset our counters
rowCount = 0 ; paramIdx = 1 ; removed . clear ( ) ; } } } catch ( SQLException e ) { handleSqlException ( e , removed , errorRecords ) ; } // Process the rest of the records that are removed from queue but haven ' t processed yet
// this happens when rowCount is still less than maxRowsPerBatch .
// This is a bit of an ugly fix as its not very DRY but sufficient until there ' s a larger
// refactoring of this code .
if ( ! removed . isEmpty ( ) ) { query = generateQueryForMultiRow ( opCode , columnsToParameters , getPrimaryKeyColumns ( ) , removed . size ( ) // always the remainder
) ; try ( PreparedStatement statement = jdbcUtil . getPreparedStatement ( getGeneratedColumnMappings ( ) , query , connection ) ) { int paramIdx = 1 ; for ( Record r : removed ) { if ( opCode != DELETE_CODE ) { paramIdx = setParamsToStatement ( paramIdx , statement , columnsToParameters , r , connection , opCode ) ; } if ( opCode != OperationType . INSERT_CODE ) { paramIdx = setPrimaryKeys ( paramIdx , r , statement , opCode ) ; } } processBatch ( removed , errorRecords , statement , connection ) ; } catch ( SQLException e ) { handleSqlException ( e , removed , errorRecords ) ; } }


signature:
public class JdbcMultiRowRecordWriter { /** * Handle SQLException in a smart way , detecting if the exception is data oriented or not . */
private void handleSqlException ( SQLException exception , List < Record > inputRecords , List < OnRecordErrorException > errors ) throws StageException { } }

implementation: if ( jdbcUtil . isDataError ( getCustomDataSqlStateCodes ( ) , getConnectionString ( ) , exception ) ) { String formattedError = jdbcUtil . formatSqlException ( exception ) ; LOG . error ( JdbcErrors . JDBC_89 . getMessage ( ) , formattedError ) ; for ( Record inputRecord : inputRecords ) { errors . add ( new OnRecordErrorException ( inputRecord , JdbcErrors . JDBC_89 , formattedError ) ) ; } return ; } super . handleSqlException ( exception ) ;


signature:
public class JdbcMultiRowRecordWriter { /** * Generates a hash for the fields present in a record and their mappings .
 * A specific implementation of the hash function is not guaranteed .
 * @ param record The record to generate a hash for .
 * @ return A Guava HashCode of the fields . */
private HashCode getColumnHash ( Record record , int op ) throws OnRecordErrorException { } }

implementation: Map < String , String > parameters = getColumnsToParameters ( ) ; SortedMap < String , String > columnsToParameters = recordReader . getColumnsToParameters ( record , op , parameters , getColumnsToFields ( ) ) ; return columnHashFunction . newHasher ( ) . putObject ( columnsToParameters , stringMapFunnel ) . hash ( ) ;


signature:
public class SSOUserAuthenticator { /** * Creates Login url with request URL as parameter for a callback redirection on a successful login . */
String getLoginUrl ( HttpServletRequest request , boolean repeatedRedirect ) { } }

implementation: String requestUrl = getRequestUrl ( request , TOKEN_PARAM_SET ) . toString ( ) ; return getSsoService ( ) . createRedirectToLoginUrl ( requestUrl , repeatedRedirect ) ;


signature:
public class SSOUserAuthenticator { /** * Removes the token from request URL , redirects to the modified URL , returns the token as a header */
Authentication redirectToSelf ( HttpServletRequest httpReq , HttpServletResponse httpRes ) throws ServerAuthException { } }

implementation: String authToken = httpReq . getParameter ( SSOConstants . USER_AUTH_TOKEN_PARAM ) ; String urlWithoutToken = getRequestUrlWithoutToken ( httpReq ) ; httpRes . setHeader ( SSOConstants . X_USER_AUTH_TOKEN , authToken ) ; try { LOG . debug ( "Redirecting to self without token '{}'" , urlWithoutToken ) ; httpRes . sendRedirect ( urlWithoutToken ) ; return Authentication . SEND_CONTINUE ; } catch ( IOException ex ) { throw new ServerAuthException ( Utils . format ( "Could not redirect to '{}': {}" , urlWithoutToken , ex . toString ( ) , ex ) ) ; }


signature:
public class SSOUserAuthenticator { /** * Terminates the request with an HTTP Unauthorized response or redirects to login for page requests */
@ Override protected Authentication returnUnauthorized ( HttpServletRequest httpReq , HttpServletResponse httpRes , String principalId , String logMessageTemplate ) throws ServerAuthException { } }

implementation: Authentication ret ; httpRes . addCookie ( createAuthCookie ( httpReq , "" , 0 ) ) ; if ( httpReq . getHeader ( SSOConstants . X_REST_CALL ) != null ) { ret = super . returnUnauthorized ( httpReq , httpRes , null , logMessageTemplate ) ; } else { redirectToLogin ( httpReq , httpRes ) ; ret = Authentication . SEND_FAILURE ; } return ret ;


signature:
public class PostgresCDCWalReceiver { /** * TODO this is a great place for generic validator */
public Optional < List < ConfigIssue > > validateSchemaAndTables ( List < SchemaTableConfigBean > schemaTableConfigs ) { } }

implementation: List < ConfigIssue > issues = new ArrayList < > ( ) ; for ( SchemaTableConfigBean tables : configBean . baseConfigBean . schemaTableConfigs ) { validateSchemaAndTable ( tables ) . ifPresent ( issues :: add ) ; } return Optional . ofNullable ( issues ) ;


signature:
public class SpoolDirUtil { /** * True if f1 is " newer " than f2. */
public static boolean compareFiles ( WrappedFileSystem fs , WrappedFile f1 , WrappedFile f2 ) { } }

implementation: if ( ! fs . exists ( f2 ) ) { return true ; } try { long mtime1 = fs . getLastModifiedTime ( f1 ) ; long mtime2 = fs . getLastModifiedTime ( f2 ) ; long ctime1 = fs . getChangedTime ( f1 ) ; long ctime2 = fs . getChangedTime ( f2 ) ; long time1 = Math . max ( mtime1 , ctime1 ) ; long time2 = Math . max ( mtime2 , ctime2 ) ; int compares = Long . compare ( time1 , time2 ) ; if ( compares != 0 ) { return compares > 0 ; } } catch ( IOException ex ) { LOG . error ( "Failed to get ctime: '{}'" , f1 . getFileName ( ) , ex ) ; return false ; } return f1 . getAbsolutePath ( ) . compareTo ( f2 . getAbsolutePath ( ) ) > 0 ;


signature:
public class SpoolDirUtil { /** * Returns the { @ code DataParser } of the file , could be local file and hdfs file
 * @ return the { @ code DataParser } of the file */
public static DataParser getParser ( WrappedFileSystem fs , WrappedFile file , DataFormat dataFormat , DataParserFactory parserFactory , String offset , int wholeFileMaxObjectLen , ELEval rateLimitElEval , ELVars rateLimitElVars , String rateLimit ) throws DataParserException , ELEvalException , IOException { } }

implementation: DataParser parser ; switch ( dataFormat ) { case WHOLE_FILE : FileRef fileRef = fs . getFileRefBuilder ( ) . filePath ( file . getAbsolutePath ( ) ) . bufferSize ( wholeFileMaxObjectLen ) . rateLimit ( FileRefUtil . evaluateAndGetRateLimit ( rateLimitElEval , rateLimitElVars , rateLimit ) ) . createMetrics ( true ) . totalSizeInBytes ( file . getSize ( ) ) . build ( ) ; parser = parserFactory . getParser ( file . getFileName ( ) , file . getFileMetadata ( ) , fileRef ) ; break ; default : parser = parserFactory . getParser ( file . getFileName ( ) , file . getInputStream ( ) , offset ) ; } return parser ;


signature:
public class SpoolDirUtil { /** * Returns the absolute path from the root to the last base directory
 * Meaning the path until the last / before the first *
 * @ param directory String containing the path to the directory
 * @ return the path truncated to the las complete directory */
public static String truncateGlobPatternDirectory ( String directory ) { } }

implementation: String [ ] absolutePath = directory . split ( ESCAPED_ASTERISK ) ; String truncatedString = absolutePath [ 0 ] ; if ( lastCharacterIsAsterisk ( truncatedString ) != SLASH . toCharArray ( ) [ 0 ] ) { List < String > subDirectories = Arrays . asList ( truncatedString . split ( SLASH ) ) ; StringBuffer stringBuffer = new StringBuffer ( ) ; stringBuffer . append ( String . join ( SLASH , subDirectories . subList ( 0 , subDirectories . size ( ) - 1 ) ) ) . append ( SLASH ) ; truncatedString = stringBuffer . toString ( ) ; } LOG . debug ( String . format ( "Checking existence of path: %s" , truncatedString ) ) ; return truncatedString ;