signature:
public class HttpClientSource { /** * Cleanup the { @ link DataParser } , response ' s { @ link InputStream } and the { @ link Response } itself .
 * @ param in the InputStream we are finished with
 * @ throws IOException If a resource is not closed properly */
private void cleanupResponse ( InputStream in ) throws IOException { } }

implementation: IOException ex = null ; LOG . debug ( "Cleanup after request processing complete." ) ; lastRequestCompletedTime = System . currentTimeMillis ( ) ; if ( in != null ) { try { in . close ( ) ; } catch ( IOException e ) { LOG . warn ( "Error closing input stream" , ex ) ; ex = e ; } } getResponse ( ) . close ( ) ; setResponse ( null ) ; try { parser . close ( ) ; } catch ( IOException e ) { LOG . warn ( "Error closing parser" , ex ) ; ex = e ; } parser = null ; if ( ex != null ) { throw ex ; }


signature:
public class HttpClientSource { /** * Returns the most recently requested page number or page offset requested .
 * @ return page number or offset */
@ VisibleForTesting int getCurrentPage ( ) { } }

implementation:
// Body params take precedence , but usually only one or the other should be used .
if ( bodyVars . hasVariable ( START_AT ) ) { return ( int ) bodyVars . getVariable ( START_AT ) ; } else if ( resourceVars . hasVariable ( START_AT ) ) { return ( int ) resourceVars . getVariable ( START_AT ) ; } return 0 ;


signature:
public class HttpClientSource { /** * Parses a paginated result from the configured field .
 * @ param batchMaker batch to add records to
 * @ param sourceOffset to use as a base when creating records
 * @ param record the source record containing an array to be converted to records
 * @ return number of records parsed
 * @ throws StageException if an unhandled error is encountered */
private int parsePaginatedResult ( BatchMaker batchMaker , String sourceOffset , Record record ) throws StageException { } }

implementation: int numSubRecords = 0 ; if ( ! record . has ( conf . pagination . resultFieldPath ) ) { final StageException stageException = new StageException ( Errors . HTTP_12 , conf . pagination . resultFieldPath ) ; LOG . error ( stageException . getMessage ( ) ) ; throw stageException ; } Field resultField = record . get ( conf . pagination . resultFieldPath ) ; if ( resultField . getType ( ) != Field . Type . LIST ) { final StageException stageException = new StageException ( Errors . HTTP_08 , resultField . getType ( ) ) ; LOG . error ( stageException . getMessage ( ) ) ; throw stageException ; } List < Field > results = resultField . getValueAsList ( ) ; int subRecordIdx = 0 ; for ( Field result : results ) { Record r = getContext ( ) . createRecord ( sourceOffset + "::" + subRecordIdx ++ ) ; if ( conf . pagination . keepAllFields ) { r . set ( record . get ( ) . clone ( ) ) ; r . set ( conf . pagination . resultFieldPath , result ) ; } else { r . set ( result ) ; } addResponseHeaders ( r . getHeader ( ) ) ; batchMaker . addRecord ( r ) ; ++ numSubRecords ; } if ( conf . pagination . mode != PaginationMode . LINK_FIELD ) { haveMorePages = numSubRecords > 0 ; } return numSubRecords ;


signature:
public class HttpClientSource { /** * Adds the HTTP response headers to the record header .
 * @ param header an SDC record header to populate */
private void addResponseHeaders ( Record . Header header ) { } }

implementation: final MultivaluedMap < String , String > headers = getResponse ( ) . getStringHeaders ( ) ; if ( headers == null ) { return ; } for ( Map . Entry < String , List < String > > entry : headers . entrySet ( ) ) { if ( ! entry . getValue ( ) . isEmpty ( ) ) { String firstValue = entry . getValue ( ) . get ( 0 ) ; header . setAttribute ( entry . getKey ( ) , firstValue ) ; } }


signature:
public class HttpClientSource { /** * Resolves any expressions in the Header value entries of the request .
 * @ return map of evaluated headers to add to the request
 * @ throws StageException if an unhandled error is encountered */
private MultivaluedMap < String , Object > resolveHeaders ( ) throws StageException { } }

implementation: MultivaluedMap < String , Object > requestHeaders = new MultivaluedHashMap < > ( ) ; for ( Map . Entry < String , String > entry : conf . headers . entrySet ( ) ) { List < Object > header = new ArrayList < > ( 1 ) ; Object resolvedValue = headerEval . eval ( headerVars , entry . getValue ( ) , String . class ) ; header . add ( resolvedValue ) ; requestHeaders . put ( entry . getKey ( ) , header ) ; hasher . putString ( entry . getKey ( ) , Charset . forName ( conf . dataFormatConfig . charset ) ) ; hasher . putString ( entry . getValue ( ) , Charset . forName ( conf . dataFormatConfig . charset ) ) ; } return requestHeaders ;


signature:
public class HttpClientSource { /** * Verifies that the response was a successful one and has data and continues to parse the response .
 * @ param maxRecords maximum number of records to add to the batch
 * @ param batchMaker batch of records to populate
 * @ return a new source offset if the response was successful
 * @ throws StageException if an unhandled error is encountered */
private Optional < String > processResponse ( long start , int maxRecords , BatchMaker batchMaker ) throws StageException { } }

implementation:
Optional < String > newSourceOffset = Optional . empty ( ) ; if ( getResponse ( ) == null ) { return newSourceOffset ; } // Response was not in the OK range , so treat as an error
int status = getResponse ( ) . getStatus ( ) ; if ( status < 200 || status >= 300 ) { lastRequestCompletedTime = System . currentTimeMillis ( ) ; String reason = getResponse ( ) . getStatusInfo ( ) . getReasonPhrase ( ) ; String respString = getResponse ( ) . readEntity ( String . class ) ; getResponse ( ) . close ( ) ; setResponse ( null ) ; final String errorMsg = reason + " : " + respString ; LOG . warn ( Errors . HTTP_01 . getMessage ( ) , status , errorMsg ) ; errorRecordHandler . onError ( Errors . HTTP_01 , status , errorMsg ) ; return newSourceOffset ; } if ( conf . pagination . mode == PaginationMode . LINK_HEADER ) { next = getResponse ( ) . getLink ( "next" ) ; if ( next == null ) { haveMorePages = false ; } } if ( getResponse ( ) . hasEntity ( ) ) { newSourceOffset = Optional . of ( parseResponse ( start , maxRecords , batchMaker ) ) ; } else if ( conf . httpMethod . getLabel ( ) == "HEAD" ) { // Handle HEAD only requests , which have no body , by creating a blank record for output with headers .
newSourceOffset = Optional . of ( parseHeadersOnly ( batchMaker ) ) ; } return newSourceOffset ;


signature:
public class AvroConversionBaseMapper { /** * Return true if and only if given property is defined with non empty non default value */
protected boolean propertyDefined ( Configuration conf , String propertyName ) { } }

implementation:
String prop = conf . get ( propertyName ) ; // String property will have default empty , integer - 1 , we ' ll skip both of them
return prop != null && ! prop . isEmpty ( ) && ! prop . equals ( "-1" ) ;


signature:
public class ShimUtil { /** * Change package name for DataParserException .
 * @ param original Original exception from * . lib . * package
 * @ return New exception from * . api . * package */
public static DataParserException convert ( com . streamsets . pipeline . lib . parser . DataParserException original ) { } }

implementation: if ( original instanceof com . streamsets . pipeline . lib . parser . RecoverableDataParserException ) { return new RecoverableDataParserException ( ( ( com . streamsets . pipeline . lib . parser . RecoverableDataParserException ) original ) . getUnparsedRecord ( ) , original . getErrorCode ( ) , original . getParams ( ) ) ; } return new DataParserException ( original . getErrorCode ( ) , original . getParams ( ) ) ;


signature:
public class ShimUtil { /** * Change package name for DataGeneratorException .
 * @ param original Original exception from * . lib . * package
 * @ return New exception from * . api . * package */
public static DataGeneratorException convert ( com . streamsets . pipeline . lib . generator . DataGeneratorException original ) { } }

implementation: return new DataGeneratorException ( original . getErrorCode ( ) , original . getParams ( ) ) ;


signature:
public class GoogleCloudCredentialsConfig { /** * Tries to create a { @ link CredentialsProvider } for the appropriate type of credentials supplied .
 * @ param context stage context used for creating config issues
 * @ param issues list of issues to append to
 * @ return Optional containing a credentials provider if it was successfully created . */
public Optional < CredentialsProvider > getCredentialsProvider ( Stage . Context context , List < Stage . ConfigIssue > issues ) { } }

implementation: CredentialsProvider provider = null ; if ( credentialsProvider . equals ( CredentialsProviderType . DEFAULT_PROVIDER ) ) { return Optional . of ( SubscriptionAdminSettings . defaultCredentialsProviderBuilder ( ) . build ( ) ) ; } else if ( credentialsProvider . equals ( CredentialsProviderType . JSON_PROVIDER ) ) { Credentials credentials = getCredentials ( context , issues ) ; provider = new FixedCredentialsProvider ( ) { @ Nullable @ Override public Credentials getCredentials ( ) { return credentials ; } } ; } return Optional . ofNullable ( provider ) ;


signature:
public class GoogleCloudCredentialsConfig { /** * Reads a JSON credentials file for a service account from and returns any errors .
 * @ param issues list to append any discovered issues .
 * @ return a generic credentials object */
private Credentials getCredentials ( Stage . Context context , List < Stage . ConfigIssue > issues ) { } }

implementation: Credentials credentials = null ; File credentialsFile ; if ( Paths . get ( path ) . isAbsolute ( ) ) { credentialsFile = new File ( path ) ; } else { credentialsFile = new File ( context . getResourcesDirectory ( ) , path ) ; } if ( ! credentialsFile . exists ( ) || ! credentialsFile . isFile ( ) ) { LOG . error ( GOOGLE_01 . getMessage ( ) , credentialsFile . getPath ( ) ) ; issues . add ( context . createConfigIssue ( Groups . CREDENTIALS . name ( ) , CONF_CREDENTIALS_CREDENTIALS_PROVIDER , GOOGLE_01 , credentialsFile . getPath ( ) ) ) ; return null ; } try ( InputStream in = new FileInputStream ( credentialsFile ) ) { credentials = ServiceAccountCredentials . fromStream ( in ) ; } catch ( IOException | IllegalArgumentException e ) { LOG . error ( GOOGLE_02 . getMessage ( ) , e ) ; issues . add ( context . createConfigIssue ( Groups . CREDENTIALS . name ( ) , CONF_CREDENTIALS_CREDENTIALS_PROVIDER , GOOGLE_02 ) ) ; } return credentials ;


signature:
public class PreviewApi { /** * Run Pipeline preview
 * @ param pipelineId
 * @ param stageOutputsToOverrideJson
 * @ param rev
 * @ param batchSize
 * @ param batches
 * @ param skipTargets
 * @ param endStage
 * @ param timeout
 * @ return PreviewInfoJson */
public PreviewInfoJson previewWithOverride ( String pipelineId , List < StageOutputJson > stageOutputsToOverrideJson , String rev , Integer batchSize , Integer batches , Boolean skipTargets , String endStage , Long timeout ) throws ApiException { } }

implementation:
Object postBody = stageOutputsToOverrideJson ; byte [ ] postBinaryBody = null ; // verify the required parameter ' pipelineId ' is set
if ( pipelineId == null ) { throw new ApiException ( 400 , "Missing the required parameter 'pipelineId' when calling previewWithOverride" ) ; } // verify the required parameter ' stageOutputsToOverrideJson ' is set
if ( stageOutputsToOverrideJson == null ) { throw new ApiException ( 400 , "Missing the required parameter 'stageOutputsToOverrideJson' when calling previewWithOverride" ) ; } // create path and map variables
String path = "/v1/pipeline/{pipelineId}/preview" . 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 ( "" , "batchSize" , batchSize ) ) ; queryParams . addAll ( apiClient . parameterToPairs ( "" , "batches" , batches ) ) ; queryParams . addAll ( apiClient . parameterToPairs ( "" , "skipTargets" , skipTargets ) ) ; queryParams . addAll ( apiClient . parameterToPairs ( "" , "endStage" , endStage ) ) ; queryParams . addAll ( apiClient . parameterToPairs ( "" , "timeout" , timeout ) ) ; 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 < PreviewInfoJson > ( ) { } ; return apiClient . invokeAPI ( path , "POST" , queryParams , postBody , postBinaryBody , headerParams , formParams , accept , contentType , authNames , returnType ) ;


signature:
public class AbstractOverrunDelimitedReader { /** * we use this to trim the output in case of overruns */
protected int copyToBuffer ( StringBuilder s , int initialLen , int startChar , int currentChar ) { } }

implementation: int overrun = 0 ; int currentSize = s . length ( ) - initialLen ; int readSize = currentChar - startChar ; if ( maxLine > - 1 && currentSize + readSize > maxLine ) { int adjustedReadSize = maxLine - currentSize ; if ( adjustedReadSize > 0 ) { s . append ( cb , startChar , adjustedReadSize ) ; overrun = readSize - adjustedReadSize ; } else { overrun = readSize ; } } else { s . append ( cb , startChar , readSize ) ; } return overrun ;


signature:
public class ApplicationPackage { /** * Finds package names associated with a class loader which which are not JVM level
 * packages . Anything inside java home are specifically excluded as well as some
 * OS specific install locations ( MacOS ) . */
private static SortedSet < String > findApplicationPackageNames ( ClassLoader cl ) { } }

implementation:
SortedSet < String > packages = new TreeSet < > ( ) ; while ( cl != null ) { if ( cl instanceof URLClassLoader ) { for ( URL url : ( ( URLClassLoader ) cl ) . getURLs ( ) ) { String path = url . getPath ( ) ; if ( ! path . startsWith ( JAVA_HOME ) && ! path . startsWith ( MACOS_JAVA_EXTENSIONS_DIR ) && path . endsWith ( JAR_FILE_SUFFIX ) ) { try { try ( ZipInputStream zip = new ZipInputStream ( url . openStream ( ) ) ) { for ( ZipEntry entry = zip . getNextEntry ( ) ; entry != null ; entry = zip . getNextEntry ( ) ) { if ( ! entry . isDirectory ( ) && entry . getName ( ) . endsWith ( CLASS_FILE_SUFFIX ) ) { // This ZipEntry represents a class . Now , what class does it represent ?
String className = entry . getName ( ) . replace ( '/' , '.' ) ; // including " . class "
className = className . substring ( 0 , className . length ( ) - CLASS_FILE_SUFFIX . length ( ) ) ; if ( className . contains ( "." ) && ! className . startsWith ( STREAMSETS_PACKAGE ) ) { // must end with a . as we don ' t want o . a . h matching o . a . ha
packages . add ( className . substring ( 0 , className . lastIndexOf ( '.' ) ) + "." ) ; } } } } } catch ( IOException unlikely ) { // since these are local URL we will likely only
// hit this if there is a corrupt jar in the classpath
// which we will ignore
if ( SDCClassLoader . isDebug ( ) ) { System . err . println ( "Error opening '" + url + "' : " + unlikely ) ; unlikely . printStackTrace ( ) ; } } } } } cl = cl . getParent ( ) ; } SystemPackage systemPackage = new SystemPackage ( SDCClassLoader . SYSTEM_API_CHILDREN_CLASSES ) ; Iterator < String > iterator = packages . iterator ( ) ; while ( iterator . hasNext ( ) ) { String packageName = iterator . next ( ) ; if ( systemPackage . isSystem ( packageName ) ) { iterator . remove ( ) ; } } removeLogicalDuplicates ( packages ) ; return packages ;


signature:
public class ApplicationPackage { /** * Traverses sorted list of packages and removes logical duplicates . For example
 * if the set contains akka . , akka . io . , and akka . util . only akka . will remain .
 * Note that if the set contains only akka . io . and akka . util . both will remain .
 * Otherwise all of the org . apache . would devolve to org . */
static void removeLogicalDuplicates ( SortedSet < String > packages ) { } }

implementation: Iterator < String > iterator = packages . iterator ( ) ; if ( ! iterator . hasNext ( ) ) { return ; } String last = iterator . next ( ) ; while ( iterator . hasNext ( ) ) { String current = iterator . next ( ) ; if ( current . startsWith ( last ) ) { iterator . remove ( ) ; } else { last = current ; } }


signature:
public class HdfsTarget { /** * new data . */
protected void emptyBatch ( ) throws StageException { } }

implementation: setBatchTime ( ) ; try { hdfsTargetConfigBean . getUGI ( ) . doAs ( new PrivilegedExceptionAction < Void > ( ) { @ Override public Void run ( ) throws Exception { hdfsTargetConfigBean . getCurrentWriters ( ) . purge ( ) ; if ( hdfsTargetConfigBean . getLateWriters ( ) != null ) { hdfsTargetConfigBean . getLateWriters ( ) . purge ( ) ; } return null ; } } ) ; } catch ( Exception ex ) { throw throwStageException ( ex ) ; }


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

implementation:
// Validate configuration values and open any required resources .
List < ConfigIssue > issues = super . init ( ) ; errorRecordHandler = new DefaultErrorRecordHandler ( getContext ( ) ) ; elEvals . init ( getContext ( ) ) ; Processor . Context context = getContext ( ) ; issues . addAll ( hikariConfigBean . validateConfigs ( context , issues ) ) ; if ( issues . isEmpty ( ) && null == dataSource ) { try { dataSource = jdbcUtil . createDataSourceForWrite ( hikariConfigBean , null , null , false , issues , Collections . emptyList ( ) , getContext ( ) ) ; } catch ( RuntimeException | SQLException | StageException e ) { LOG . debug ( "Could not connect to data source" , e ) ; issues . add ( getContext ( ) . createConfigIssue ( Groups . JDBC . name ( ) , CONNECTION_STRING , JdbcErrors . JDBC_00 , e . toString ( ) ) ) ; } } if ( issues . isEmpty ( ) ) { try { schemaWriter = JdbcSchemaWriterFactory . create ( hikariConfigBean . getConnectionString ( ) , dataSource ) ; } catch ( JdbcStageCheckedException e ) { issues . add ( getContext ( ) . createConfigIssue ( Groups . JDBC . name ( ) , CONNECTION_STRING , e . getErrorCode ( ) , e . getParams ( ) ) ) ; } schemaReader = new JdbcSchemaReader ( dataSource , schemaWriter ) ; tableCache = CacheBuilder . newBuilder ( ) . maximumSize ( 50 ) . build ( new CacheLoader < Pair < String , String > , LinkedHashMap < String , JdbcTypeInfo > > ( ) { @ Override public LinkedHashMap < String , JdbcTypeInfo > load ( Pair < String , String > pair ) throws Exception { return schemaReader . getTableSchema ( pair . getLeft ( ) , pair . getRight ( ) ) ; } } ) ; } // If issues is not empty , the UI will inform the user of each configuration issue in the list .
return issues ;


signature:
public class MultiLineLiveFileReader { /** * finds the first main line in the chunk from the specified index position onwards */
int findNextMainLine ( LiveFileChunk chunk , int startIdx ) { } }

implementation: List < FileLine > lines = chunk . getLines ( ) ; int found = - 1 ; for ( int i = startIdx ; found == - 1 && i < lines . size ( ) ; i ++ ) { if ( pattern . matcher ( lines . get ( i ) . getText ( ) . trim ( ) ) . matches ( ) ) { found = i ; } } return found ;


signature:
public class MultiLineLiveFileReader { /** * it there is an incomplete multiline from a previous chunk it starts from it . */
LiveFileChunk resolveChunk ( LiveFileChunk chunk ) { } }

implementation:
List < FileLine > completeLines = new ArrayList < > ( ) ; List < FileLine > chunkLines = chunk . getLines ( ) ; if ( incompleteMultiLine . length ( ) == 0 ) { incompleteMultiLineOffset = chunk . getOffset ( ) ; incompleteMultiLineTruncated = chunk . isTruncated ( ) ; } incompleteMultiLineTruncated |= chunk . isTruncated ( ) ; int pos = 0 ; int idx = findNextMainLine ( chunk , pos ) ; // while we have main lines we keep adding / compacting into the new chunk
while ( idx > - 1 ) { // any multi lines up to the next main line belong to the previous main line
for ( int i = pos ; i < idx ; i ++ ) { incompleteMultiLine . append ( chunkLines . get ( i ) . getText ( ) ) ; } // if we have incomplete lines , at this point they are a complete multiline , compact and add to new chunk lines
if ( incompleteMultiLine . length ( ) != 0 ) { completeLines . add ( new FileLine ( incompleteMultiLineOffset , incompleteMultiLine . toString ( ) ) ) ; incompleteMultiLineOffset += incompleteMultiLine . length ( ) ; // clear the incomplete multi lines as we just used them to create a full line
incompleteMultiLine . setLength ( 0 ) ; incompleteMultiLineTruncated = false ; } // add the current main line as incomplete as we still don ' t if it is a complete line
incompleteMultiLine . append ( chunkLines . get ( idx ) . getText ( ) ) ; // find the next main line
pos = idx + 1 ; idx = findNextMainLine ( chunk , pos ) ; } // lets process the left over multi lines in the chunk after the last main line .
// if any they will kept to completed with lines from the next chunk .
for ( int i = pos ; i < chunkLines . size ( ) ; i ++ ) { incompleteMultiLine . append ( chunkLines . get ( i ) . getText ( ) ) ; } if ( completeLines . isEmpty ( ) ) { // didn ' t get a complete multi line yet , we keep storing lines but return a null chunk
chunk = null ; } else { // create a new chunk with all complete multi lines
chunk = new LiveFileChunk ( chunk . getTag ( ) , chunk . getFile ( ) , chunk . getCharset ( ) , completeLines , incompleteMultiLineTruncated ) ; } return chunk ;


signature:
public class ConnectionManager { /** * Get { @ link Connection } if cached connection is null fetch a connection from { @ link HikariDataSource }
 * @ return { @ link Connection }
 * @ throws SQLException */
public Connection getConnection ( ) throws SQLException { } }

implementation: if ( threadLocalConnection . get ( ) == null ) { threadLocalConnection . set ( getNewConnection ( ) ) ; } return threadLocalConnection . get ( ) ;


signature:
public class ConnectionManager { /** * Close the current thread ' s connection */
public void closeConnection ( ) { } }

implementation: LOGGER . debug ( "Closing connection" ) ; Connection connectionToRemove = threadLocalConnection . get ( ) ; jdbcUtil . closeQuietly ( connectionToRemove ) ; if ( connectionToRemove != null ) { synchronized ( this ) { connectionsToCloseDuringDestroy . remove ( connectionToRemove ) ; } } threadLocalConnection . set ( null ) ;


signature:
public class AbstractWrapperStream { /** * InputStream */
@ Override public int read ( byte [ ] b ) throws IOException { } }

implementation: checkState ( InputStream . class ) ; return ( ( InputStream ) stream ) . read ( b ) ;


signature:
public class AbstractWrapperStream { /** * InputStream */
@ Override public int read ( byte [ ] b , int offset , int len ) throws IOException { } }

implementation: checkState ( InputStream . class ) ; return ( ( InputStream ) stream ) . read ( b , offset , len ) ;


signature:
public class AbstractWrapperStream { /** * InputStream */
@ Override public long skip ( long n ) throws IOException { } }

implementation: checkState ( InputStream . class ) ; return ( ( InputStream ) stream ) . skip ( n ) ;


signature:
public class AbstractWrapperStream { /** * ReadableByteChannel */
@ Override public int read ( ByteBuffer dst ) throws IOException { } }

implementation: checkState ( ReadableByteChannel . class ) ; return ( ( ReadableByteChannel ) stream ) . read ( dst ) ;


signature:
public class FieldRenamerProcessor { /** * Add backslash to escape the | character within quoted sections of the
 * input string . This prevents the | from being processed as part of a regex .
 * @ param input Contains the field name to process .
 * @ return New string */
private static String escapeQuotedSubstring ( String input ) { } }

implementation: String [ ] parts = input . split ( "'" ) ; StringBuilder output = new StringBuilder ( input . length ( ) * 2 ) ; for ( int i = 0 ; i < parts . length ; i ++ ) { if ( ( i % 2 ) == 1 ) { output . append ( "'" ) . append ( parts [ i ] . replace ( "|" , "\\|" ) ) . append ( "'" ) ; } else { output . append ( parts [ i ] ) ; } } return output . toString ( ) ;


signature:
public class PubSubSource { /** * Returns a flow control setting such that a subscriber will block if it has buffered more messages than can be
 * processed in a single batch times the number of record processors . Since the flow control settings are per
 * subscriber , we should divide by the number of subscribers to avoid buffering too much data in each subscriber .
 * @ return settings based on the stage configuration . */
private FlowControlSettings getFlowControlSettings ( ) { } }

implementation: return FlowControlSettings . newBuilder ( ) . setLimitExceededBehavior ( FlowController . LimitExceededBehavior . Block ) . setMaxOutstandingElementCount ( ( long ) conf . basic . maxBatchSize * conf . maxThreads / conf . advanced . numSubscribers ) . build ( ) ;


signature:
public class PubSubSource { /** * Creates a channel provider shared by each subscriber . It is basically the default ChannelProvider with the
 * exception that it can be configured with a custom endpoint , for example when running against the PubSub Emulator .
 * @ return channel provider based on the stage configuration . */
private InstantiatingGrpcChannelProvider getChannelProvider ( ) { } }

implementation: return SubscriptionAdminSettings . defaultGrpcTransportProviderBuilder ( ) . setMaxInboundMessageSize ( MAX_INBOUND_MESSAGE_SIZE ) . setEndpoint ( Strings . isNullOrEmpty ( conf . advanced . customEndpoint ) ? SubscriptionAdminSettings . getDefaultEndpoint ( ) : conf . advanced . customEndpoint ) . build ( ) ;


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

implementation: Map < String , Field > fields = new HashMap < > ( ) ; fields . put ( HiveMetastoreUtil . TYPE , Field . create ( hiveTypeInfo . getHiveType ( ) . name ( ) ) ) ; fields . put ( HiveMetastoreUtil . EXTRA_INFO , generateExtraInfoFieldForMetadataRecord ( hiveTypeInfo ) ) ; fields . put ( HiveMetastoreUtil . COMMENT , Field . create ( Field . Type . STRING , hiveTypeInfo . getComment ( ) ) ) ; return Field . create ( fields ) ;


signature:
public class HiveTypeSupport { /** * Generate { @ link HiveTypeInfo } from the Metadata Record < br > .
 * ( Reverse of { @ link # generateHiveTypeInfoFieldForMetadataRecord ( HiveTypeInfo ) } )
 * @ throws StageException if the metadata field is not valid .
 * @ param hiveTypeInfoField hiveTypeInfoField
 * @ return { @ link HiveTypeInfo }
 * @ throws StageException If the record has invalid */
@ SuppressWarnings ( "unchecked" ) public HiveTypeInfo generateHiveTypeInfoFromMetadataField ( Field hiveTypeInfoField ) throws StageException { } }

implementation: if ( hiveTypeInfoField . getType ( ) == Field . Type . MAP ) { Map < String , Field > fields = ( Map < String , Field > ) hiveTypeInfoField . getValue ( ) ; if ( ! fields . containsKey ( HiveMetastoreUtil . TYPE ) || ! fields . containsKey ( HiveMetastoreUtil . EXTRA_INFO ) ) { throw new StageException ( Errors . HIVE_17 , HiveMetastoreUtil . TYPE_INFO ) ; } HiveType hiveType = HiveType . getHiveTypeFromString ( fields . get ( HiveMetastoreUtil . TYPE ) . getValueAsString ( ) ) ; String comment = "" ; if ( fields . containsKey ( HiveMetastoreUtil . COMMENT ) ) { comment = fields . get ( HiveMetastoreUtil . COMMENT ) . getValueAsString ( ) ; } return generateHiveTypeInfoFromMetadataField ( hiveType , comment , fields . get ( HiveMetastoreUtil . EXTRA_INFO ) ) ; } else { throw new StageException ( Errors . HIVE_17 , HiveMetastoreUtil . TYPE_INFO ) ; }


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

implementation: return String . format ( HiveMetastoreUtil . COLUMN_TYPE , columnName , hiveTypeInfo . toString ( ) , hiveTypeInfo . getComment ( ) ) ;


signature:
public class Producer { /** * Queues the batch for the consumer and waits until the consumer
 * successfully commits the batch . While waiting processes any
 * control messages from the consumer . Throws an exception
 * when the consumer has indicated it encountered an error . */
public Object put ( OffsetAndResult < Map . Entry > batch ) { } }

implementation:
if ( consumerError != null ) { throw new RuntimeException ( Utils . format ( "Consumer encountered error: {}" , consumerError ) , consumerError ) ; } if ( producerError != null ) { throw new RuntimeException ( Utils . format ( "Producer encountered error: {}" , producerError ) , producerError ) ; } try { Object expectedOffset = "EMPTY_BATCH" ; if ( ! batch . getResult ( ) . isEmpty ( ) ) { expectedOffset = batch . getResult ( ) . get ( batch . getResult ( ) . size ( ) - 1 ) . getKey ( ) ; // get the last one
} while ( ! dataChannel . offer ( batch , 10 , TimeUnit . MILLISECONDS ) ) { for ( ControlChannel . Message controlMessage : controlChannel . getProducerMessages ( ) ) { switch ( controlMessage . getType ( ) ) { case CONSUMER_ERROR : Throwable throwable = ( Throwable ) controlMessage . getPayload ( ) ; consumerError = throwable ; throw new ConsumerRuntimeException ( Utils . format ( "Consumer encountered error: {}" , throwable ) , throwable ) ; default : String msg = Utils . format ( "Illegal control message type: '{}'" , controlMessage . getType ( ) ) ; throw new IllegalStateException ( msg ) ; } } } return expectedOffset ; } catch ( Throwable throwable ) { controlChannel . producerComplete ( ) ; if ( ! ( throwable instanceof ConsumerRuntimeException ) ) { String msg = "Error caught in producer: " + throwable ; LOG . error ( msg , throwable ) ; controlChannel . producerError ( throwable ) ; if ( producerError == null ) { producerError = throwable ; } } throw Throwables . propagate ( throwable ) ; }


signature:
public class GrokDictionary { /** * Compiles the expression into a pattern
 * This uses the internal dictionary of named regular expressions
 * @ param expression
 * @ return The compiled expression */
public Grok compileExpression ( final String expression ) { } }

implementation: throwErrorIfDictionaryIsNotReady ( ) ; final String digestedExpression = digestExpressionAux ( expression ) ; logger . debug ( "Digested [" + expression + "] into [" + digestedExpression + "] before compilation" ) ; return new Grok ( Pattern . compile ( digestedExpression ) ) ;


signature:
public class GrokDictionary { /** * Digests the original expression into a pure named regex
 * @ param originalExpression */
private String digestExpressionAux ( String originalExpression ) { } }

implementation:
final String PATTERN_START = "%{" ; final String PATTERN_STOP = "}" ; final char PATTERN_DELIMITER = ':' ; while ( true ) { int PATTERN_START_INDEX = originalExpression . indexOf ( PATTERN_START ) ; int PATTERN_STOP_INDEX = originalExpression . indexOf ( PATTERN_STOP , PATTERN_START_INDEX + PATTERN_START . length ( ) ) ; // End the loop is % { or } is not in the current line
if ( PATTERN_START_INDEX < 0 || PATTERN_STOP_INDEX < 0 ) { break ; } // Grab what ' s inside % { }
String grokPattern = originalExpression . substring ( PATTERN_START_INDEX + PATTERN_START . length ( ) , PATTERN_STOP_INDEX ) ; // Where is the : character
int PATTERN_DELIMITER_INDEX = grokPattern . indexOf ( PATTERN_DELIMITER ) ; String regexName = grokPattern ; String groupName = null ; if ( PATTERN_DELIMITER_INDEX >= 0 ) { regexName = grokPattern . substring ( 0 , PATTERN_DELIMITER_INDEX ) ; groupName = grokPattern . substring ( PATTERN_DELIMITER_INDEX + 1 , grokPattern . length ( ) ) ; } final String dictionaryValue = regexDictionary . get ( regexName ) ; if ( dictionaryValue == null ) { throw new GrokCompilationException ( "Missing value for regex name : " + regexName ) ; } // Defer till next iteration
if ( dictionaryValue . contains ( PATTERN_START ) ) { break ; } String replacement = dictionaryValue ; // Named capture group
if ( null != groupName ) { replacement = "(?<" + groupName + ">" + dictionaryValue + ")" ; } originalExpression = new StringBuilder ( originalExpression ) . replace ( PATTERN_START_INDEX , PATTERN_STOP_INDEX + PATTERN_STOP . length ( ) , replacement ) . toString ( ) ; } return originalExpression ;


signature:
public class GrokDictionary { /** * Loads dictionary from an input stream
 * This can be used to load dictionaries available in the class path < p >
 * @ param inputStream */
public void addDictionary ( final InputStream inputStream ) { } }

implementation: try { addDictionaryAux ( new InputStreamReader ( inputStream , "UTF-8" ) ) ; } catch ( IOException e ) { throw new GrokCompilationException ( e ) ; }


signature:
public class GrokDictionary { /** * Adds a dictionary entry via a Reader object
 * @ param reader */
public void addDictionary ( Reader reader ) { } }

implementation: try { addDictionaryAux ( reader ) ; } catch ( IOException e ) { throw new GrokCompilationException ( e ) ; } finally { IOUtils . closeQuietly ( reader ) ; }


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

implementation: Utils . checkState ( started , "Not started" ) ; Utils . checkState ( ! stopped , "Already stopped" ) ; stopped = true ; long currentTimeMillis = System . currentTimeMillis ( ) ; for ( Map . Entry < Aggregator , AggregatorData > e : data . entrySet ( ) ) { e . getValue ( ) . setTime ( currentTimeMillis ) ; } Map < Aggregator , AggregatorData > result = data ; result = aggregateDataWindows ( result ) ; return result ;


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

implementation:
Utils . checkState ( started , "Not started" ) ; Utils . checkState ( ! stopped , "Already stopped" ) ; Map < Aggregator , AggregatorData > result = data ; Map < Aggregator , AggregatorData > newData = new ConcurrentHashMap < > ( ) ; for ( Aggregator aggregator : aggregators ) { newData . put ( aggregator , aggregator . createAggregatorData ( newDataWindowEndTimeMillis ) ) ; } data = newData ; Map < Aggregator , AggregatorData > oldData = result ; // In case of sliding window , aggregate the data windows to get the result
result = aggregateDataWindows ( result ) ; if ( currentDataWindow != null ) { currentDataWindow . setDataAndClose ( oldData ) ; } DataWindow newDataWindow = createDataWindow ( newDataWindowEndTimeMillis ) ; synchronized ( dataWindowQueue ) { dataWindowQueue . add ( newDataWindow ) ; dataWindowList = new ArrayList < > ( dataWindowQueue ) ; } currentDataWindow = newDataWindow ; return result ;


signature:
public class AggregatorDataProvider { /** * Returns the current AggregatorData for an Aggregator .
 * This method is also used by the group - by element Aggregators .
 * @ param aggregator Aggregator to get the AggregatorData for .
 * @ return the current AggregatorData for an Aggregator . */
public AggregatorData getData ( Aggregator aggregator ) { } }

implementation: Utils . checkState ( started , "Not started" ) ; Utils . checkState ( ! stopped , "Already stopped" ) ; Utils . checkNotNull ( aggregator , "aggregator" ) ; Utils . checkArgument ( aggregators . contains ( aggregator ) , Utils . formatL ( "Aggregator {} is not registered to provider" , aggregator ) ) ; return data . get ( aggregator ) ;


signature:
public class JSON { /** * Serialize the given Java object into JSON string . */
public String serialize ( Object obj ) throws ApiException { } }

implementation: try { if ( obj != null ) return mapper . writeValueAsString ( obj ) ; else return null ; } catch ( Exception e ) { throw new ApiException ( 400 , e . getMessage ( ) ) ; }


signature:
public class JSON { /** * Deserialize the given JSON string to Java object .
 * @ param body The JSON string
 * @ param returnType The type to deserialize inot
 * @ return The deserialized Java object */
public < T > T deserialize ( String body , TypeRef returnType ) throws ApiException { } }

implementation: JavaType javaType = mapper . constructType ( returnType . getType ( ) ) ; try { return mapper . readValue ( body , javaType ) ; } catch ( IOException e ) { if ( returnType . getType ( ) . equals ( String . class ) ) return ( T ) body ; else throw new ApiException ( 500 , e . getMessage ( ) , null , body ) ; }


signature:
public class JSON { /** * Deserialize the given File to Java object .
 * @ param file The File path
 * @ param returnType The type to deserialize inot
 * @ return The deserialized Java object */
public < T > T deserialize ( File file , TypeRef returnType ) throws ApiException { } }

implementation: JavaType javaType = mapper . constructType ( returnType . getType ( ) ) ; try { return mapper . readValue ( file , javaType ) ; } catch ( IOException e ) { throw new ApiException ( 500 , e . getMessage ( ) , null , "File to read file" ) ; }


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

implementation:
// Validate configuration values and open any required resources .
List < ConfigIssue > issues = super . init ( ) ; if ( getConfig ( ) . equals ( "invalidValue" ) ) { issues . add ( getContext ( ) . createConfigIssue ( Groups . SAMPLE . name ( ) , "config" , Errors . SAMPLE_00 , "Here's what's wrong..." ) ) ; } // If issues is not empty , the UI will inform the user of each configuration issue in the list .
return issues ;


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

implementation: Iterator < Record > batchIterator = batch . getRecords ( ) ; while ( batchIterator . hasNext ( ) ) { Record record = batchIterator . next ( ) ; try { write ( record ) ; } catch ( Exception e ) { switch ( getContext ( ) . getOnErrorRecord ( ) ) { case DISCARD : break ; case TO_ERROR : getContext ( ) . toError ( record , Errors . SAMPLE_01 , e . toString ( ) ) ; break ; case STOP_PIPELINE : throw new StageException ( Errors . SAMPLE_01 , e . toString ( ) ) ; default : throw new IllegalStateException ( Utils . format ( "Unknown OnError value '{}'" , getContext ( ) . getOnErrorRecord ( ) , e ) ) ; } } }


signature:
public class SampleTarget { /** * Writes a single record to the destination .
 * @ param record the record to write to the destination .
 * @ throws OnRecordErrorException when a record cannot be written . */
private void write ( Record record ) throws OnRecordErrorException { } }

implementation:
// This is a contrived example , normally you may be performing an operation that could throw
// an exception or produce an error condition . In that case you can throw an OnRecordErrorException
// to send this record to the error pipeline with some details .
if ( ! record . has ( "/someField" ) ) { throw new OnRecordErrorException ( Errors . SAMPLE_01 , record , "exception detail message." ) ; } // TODO : write the records to your final destination


signature:
public class HiveMetastoreUtil { /** * Resolve expression from record */
public static String resolveEL ( ELEval elEval , ELVars variables , String val ) throws ELEvalException { } }

implementation: return elEval . eval ( variables , val , String . class ) ;


signature:
public class HiveMetastoreUtil { /** * Extract information from the list fields of form :
 * Column Type Information : [ { name : " column1 " , typeInfo : { " type " : " string " , " extraInfo " : " " } , { name : " column2 " , typeInfo : { " type " : " int " , " extraInfo " : " " } ]
 * Partition Type Information : [ { name : " partition _ column1 " , typeInfo : { " type " : " date " , " extraInfo " : " " } , { name : " partition _ column2 " , typeInfo : { " type " : " string " , " extraInfo " : " " } ]
 * Partition Value Information : [ { name : " column1 " , value : " 07-05-2016 " } , { name : " column2 " , value : " production " } ]
 * If any other List field path is given which does not conform to above
 * form a stage exception with the mentioned error is throws .
 * Returns a linked hash map so as to maintain the order . */
@ SuppressWarnings ( "unchecked" ) private static < T > void extractInnerMapFromTheList ( Record metadataRecord , String listFieldName , String innerPairFirstFieldName , String innerPairSecondFieldName , boolean isSecondFieldHiveType , LinkedHashMap < String , T > returnValMap , HiveStageCheckedException exception ) throws HiveStageCheckedException { } }

implementation:
boolean throwException = false ; try { if ( metadataRecord . has ( SEP + listFieldName ) ) { Field columnField = metadataRecord . get ( SEP + listFieldName ) ; List < Field > columnList = columnField . getValueAsList ( ) ; if ( columnList != null ) { for ( Field listElementField : columnList ) { if ( listElementField . getType ( ) != Field . Type . MAP && listElementField . getType ( ) != Field . Type . LIST_MAP ) { throwException = true ; break ; } LinkedHashMap < String , Field > innerPair = listElementField . getValueAsListMap ( ) ; String innerPairFirstField = innerPair . get ( innerPairFirstFieldName ) . getValueAsString ( ) ; T retVal ; if ( isSecondFieldHiveType ) { Field hiveTypeInfoField = innerPair . get ( innerPairSecondFieldName ) ; HiveType hiveType = HiveType . getHiveTypeFromString ( hiveTypeInfoField . getValueAsMap ( ) . get ( HiveMetastoreUtil . TYPE ) . getValueAsString ( ) ) ; retVal = ( T ) ( hiveType . getSupport ( ) . generateHiveTypeInfoFromMetadataField ( hiveTypeInfoField ) ) ; } else { retVal = ( T ) innerPair . get ( innerPairSecondFieldName ) . getValueAsString ( ) ; } returnValMap . put ( innerPairFirstField , retVal ) ; } } } else { // we allow partition to be empty for non - partitioned table
if ( ! listFieldName . equals ( PARTITION_FIELD ) ) throwException = true ; } } catch ( Exception e ) { LOG . error ( "Can't parse metadata record" , e ) ; throwException = true ; exception . initCause ( e ) ; } if ( throwException ) { throw exception ; }


signature:
public class HiveMetastoreUtil { /** * Opposite operation of extractInnerMapFromTheList .
 * It takes LinkedHashMap and generate a Field that contains the list .
 * This is to send metadata record to HMS target .
 * This function is called to for partition type list and partition value list . */
private static < T > Field generateInnerFieldFromTheList ( LinkedHashMap < String , T > original , String innerPairFirstFieldName , String innerPairSecondFieldName , boolean isSecondFieldHiveType ) throws HiveStageCheckedException { } }

implementation:
List < Field > columnList = new LinkedList < > ( ) ; for ( Map . Entry < String , T > pair : original . entrySet ( ) ) { LinkedHashMap < String , Field > entry = new LinkedHashMap < > ( ) ; entry . put ( innerPairFirstFieldName , Field . create ( pair . getKey ( ) ) ) ; if ( isSecondFieldHiveType ) { HiveTypeInfo hiveTypeInfo = ( HiveTypeInfo ) pair . getValue ( ) ; entry . put ( innerPairSecondFieldName , hiveTypeInfo . getHiveType ( ) . getSupport ( ) . generateHiveTypeInfoFieldForMetadataRecord ( hiveTypeInfo ) ) ; } else { entry . put ( innerPairSecondFieldName , Field . create ( pair . getValue ( ) . toString ( ) ) ) ; // stored value is " INT " . need to fix this
} columnList . add ( Field . createListMap ( entry ) ) ; } return ! columnList . isEmpty ( ) ? Field . create ( columnList ) : null ;


signature:
public class HiveMetastoreUtil { /** * Get qualified table name ( defined as dbName . tableName )
 * @ param dbName Database name
 * @ param tableName Table Name
 * @ return the qualified table name . */
public static String getQualifiedTableName ( String dbName , String tableName ) { } }

implementation: return ( dbName == null || dbName . isEmpty ( ) ) ? escapeHiveObjectName ( tableName ) : JOINER . join ( escapeHiveObjectName ( dbName ) , escapeHiveObjectName ( tableName ) ) ;


signature:
public class HiveMetastoreUtil { /** * Extract column information from the column list in " / columns " field . < br >
 * Column Type Information should exist in this form : < br >
 * [ { name : " column1 " , typeInfo : { " type " : " string " , " extraInfo " : " " } ,
 * { name : " column2 " , typeInfo : { " type " : " int " , " extraInfo " : " " } ]
 * @ param metadataRecord record which contains the { @ link # COLUMNS _ FIELD } and conform to the above structure .
 * @ return Map of column name to column type
 * @ throws HiveStageCheckedException if no column information exists or the record has invalid fields . */
public static LinkedHashMap < String , HiveTypeInfo > getColumnNameType ( Record metadataRecord ) throws HiveStageCheckedException { } }

implementation: LinkedHashMap < String , HiveTypeInfo > columnNameType = new LinkedHashMap < > ( ) ; extractInnerMapFromTheList ( metadataRecord , COLUMNS_FIELD , COLUMN_NAME , TYPE_INFO , true , columnNameType , new HiveStageCheckedException ( Errors . HIVE_17 , COLUMNS_FIELD , metadataRecord ) ) ; return columnNameType ;


signature:
public class HiveMetastoreUtil { /** * Extract column information from the Partition list in " / partitions " field . < br >
 * Partition Information should exist in this form : < br >
 * [ { name : " partition _ column1 " , typeInfo : { " type " : " date " , " extraInfo " : " " } ,
 * { name : " partition _ column2 " , typeInfo : { " type " : " string " , " extraInfo " : " " } ]
 * @ param metadataRecord record which contains the { @ link # PARTITION _ FIELD } and conform to the above structure .
 * @ return Map of partition name to partition type
 * @ throws HiveStageCheckedException if no partition information exists or the record has invalid fields . */
public static LinkedHashMap < String , HiveTypeInfo > getPartitionNameType ( Record metadataRecord ) throws HiveStageCheckedException { } }

implementation: LinkedHashMap < String , HiveTypeInfo > partitionNameType = new LinkedHashMap < > ( ) ; extractInnerMapFromTheList ( metadataRecord , PARTITION_FIELD , PARTITION_NAME , TYPE_INFO , true , partitionNameType , new HiveStageCheckedException ( Errors . HIVE_17 , PARTITION_FIELD , metadataRecord ) ) ; return partitionNameType ;


signature:
public class HiveMetastoreUtil { /** * Extract column information from the Partition list in " / partitions " field . < br >
 * Partition Value Information should exist in this form : < br >
 * [ { name : " column1 " , value : " 07-05-2016 " } , { name : " column2 " , value : " production " } ]
 * @ param metadataRecord record which contains the { @ link # PARTITION _ FIELD } and conform to the above structure .
 * @ return Map of partition name to partition value
 * @ throws HiveStageCheckedException if no partition value information exists or the record has invalid fields . */
public static LinkedHashMap < String , String > getPartitionNameValue ( Record metadataRecord ) throws HiveStageCheckedException { } }

implementation: LinkedHashMap < String , String > partitionNameValue = new LinkedHashMap < > ( ) ; extractInnerMapFromTheList ( metadataRecord , PARTITION_FIELD , PARTITION_NAME , PARTITION_VALUE , false , partitionNameValue , new HiveStageCheckedException ( Errors . HIVE_17 , PARTITION_FIELD , metadataRecord ) ) ; return partitionNameValue ;


signature:
public class HiveMetastoreUtil { /** * Returns true if this is a TABLE metadata request ( new or changed table ) .
 * @ param metadataRecord the metadata record
 * @ return boolean true or false indicating whether this metadata record is schema change / partition roll record . */
public static boolean isSchemaChangeRecord ( Record metadataRecord ) { } }

implementation: return MetadataRecordType . TABLE . name ( ) . equals ( metadataRecord . get ( SEP + METADATA_RECORD_TYPE ) . getValueAsString ( ) ) ;


signature:
public class HiveMetastoreUtil { /** * Returns true if this is a TABLE metadata request ( new or changed table ) .
 * @ param hmpDataFormat the target data format \ in { avro , parquet }
 * @ param tblPropertiesInfo the table data format
 * @ param qualifiedTableName the qualified table name
 * @ throws HiveStageCheckedException HiveStageCheckedException if the target data format does not match with the table data format . */
public static void validateTblPropertiesInfo ( HMPDataFormat hmpDataFormat , TBLPropertiesInfoCacheSupport . TBLPropertiesInfo tblPropertiesInfo , String qualifiedTableName ) throws HiveStageCheckedException { } }

implementation: if ( hmpDataFormat == HMPDataFormat . AVRO && ! tblPropertiesInfo . getSerdeLibrary ( ) . equals ( HiveMetastoreUtil . AVRO_SERDE ) ) { throw new HiveStageCheckedException ( Errors . HIVE_32 , qualifiedTableName , tblPropertiesInfo . getSerdeLibrary ( ) , hmpDataFormat . getLabel ( ) ) ; } else if ( hmpDataFormat == HMPDataFormat . PARQUET && ! tblPropertiesInfo . getSerdeLibrary ( ) . equals ( HiveMetastoreUtil . PARQUET_SERDE ) ) { throw new HiveStageCheckedException ( Errors . HIVE_32 , qualifiedTableName , tblPropertiesInfo . getSerdeLibrary ( ) , hmpDataFormat . getLabel ( ) ) ; }


signature:
public class HiveMetastoreUtil { /** * Get Table Name from the metadata record .
 * @ param metadataRecord the metadata record
 * @ return Table Name
 * @ throws HiveStageCheckedException if the table field does not exist in the metadata record . */
public static String getTableName ( Record metadataRecord ) throws HiveStageCheckedException { } }

implementation: if ( metadataRecord . has ( SEP + TABLE_FIELD ) ) { return metadataRecord . get ( SEP + TABLE_FIELD ) . getValueAsString ( ) ; } throw new HiveStageCheckedException ( Errors . HIVE_17 , TABLE_FIELD , metadataRecord ) ;


signature:
public class HiveMetastoreUtil { /** * Get Database Name from the metadata record .
 * @ param metadataRecord the metadata record
 * @ return Database Name
 * @ throws HiveStageCheckedException if the database name does not exist in the metadata record . */
public static String getDatabaseName ( Record metadataRecord ) throws HiveStageCheckedException { } }

implementation: if ( metadataRecord . has ( SEP + DATABASE_FIELD ) ) { String dbName = metadataRecord . get ( SEP + DATABASE_FIELD ) . getValueAsString ( ) ; return dbName . isEmpty ( ) ? DEFAULT_DBNAME : dbName ; } throw new HiveStageCheckedException ( Errors . HIVE_17 , DATABASE_FIELD , metadataRecord ) ;


signature:
public class HiveMetastoreUtil { /** * Get internal field from the metadata record .
 * @ param metadataRecord the metadata record
 * @ return internal field value
 * @ throws HiveStageCheckedException if the internal field does not exist in the metadata record . */
public static boolean getInternalField ( Record metadataRecord ) throws HiveStageCheckedException { } }

implementation: if ( metadataRecord . has ( SEP + INTERNAL_FIELD ) ) { return metadataRecord . get ( SEP + INTERNAL_FIELD ) . getValueAsBoolean ( ) ; } throw new HiveStageCheckedException ( Errors . HIVE_17 , INTERNAL_FIELD , metadataRecord ) ;


signature:
public class HiveMetastoreUtil { /** * Get Location from the metadata record .
 * @ param metadataRecord the metadata record
 * @ return location
 * @ throws HiveStageCheckedException if the location field does not exist in the metadata record . */
public static String getLocation ( Record metadataRecord ) throws HiveStageCheckedException { } }

implementation: if ( metadataRecord . has ( SEP + LOCATION_FIELD ) ) { return metadataRecord . get ( SEP + LOCATION_FIELD ) . getValueAsString ( ) ; } throw new HiveStageCheckedException ( Errors . HIVE_17 , LOCATION_FIELD , metadataRecord ) ;


signature:
public class HiveMetastoreUtil { /** * Get the customLocation flag from the metadata record . This flag marks whether or not the Hive database object
 * is stored into a custom path on the Hadoop filesystem . In both cases , the path is stored in the ' location '
 * field of the metadata record .
 * The customLocation flag is included in partition metadata records with version > 2 . For previous versions this
 * function returns { @ code DEFAULT _ CUSTOM _ LOCATION } to keep the previous behavior .
 * @ param metadataRecord the metadata record .
 * @ return true if the location has been customized , false otherwise .
 * @ throws HiveStageCheckedException */
public static boolean getCustomLocation ( Record metadataRecord ) throws HiveStageCheckedException { } }

implementation: if ( metadataRecord . get ( SEP + VERSION ) . getValueAsInteger ( ) < 3 ) { return DEFAULT_CUSTOM_LOCATION ; } if ( metadataRecord . has ( SEP + CUSTOM_LOCATION ) ) { return metadataRecord . get ( SEP + CUSTOM_LOCATION ) . getValueAsBoolean ( ) ; } throw new HiveStageCheckedException ( Errors . HIVE_17 , CUSTOM_LOCATION , metadataRecord ) ;


signature:
public class HiveMetastoreUtil { /** * Get Avro Schema from Metadata Record .
 * @ param metadataRecord the metadata record .
 * @ return Avro Schema
 * @ throws HiveStageCheckedException if the avro schema field does not exist in the metadata record . */
public static String getAvroSchema ( Record metadataRecord ) throws HiveStageCheckedException { } }

implementation: if ( metadataRecord . has ( SEP + AVRO_SCHEMA ) ) { return metadataRecord . get ( SEP + AVRO_SCHEMA ) . getValueAsString ( ) ; } throw new HiveStageCheckedException ( Errors . HIVE_17 , AVRO_SCHEMA , metadataRecord ) ;


signature:
public class HiveMetastoreUtil { /** * Get DataFormat from Metadata Record .
 * @ param metadataRecord the metadata record
 * @ return the label of dataFormat */
public static String getDataFormat ( Record metadataRecord ) throws HiveStageCheckedException { } }

implementation: if ( metadataRecord . get ( SEP + VERSION ) . getValueAsInteger ( ) == 1 ) { return DEFAULT_DATA_FORMAT ; } if ( metadataRecord . has ( SEP + DATA_FORMAT ) ) { return metadataRecord . get ( SEP + DATA_FORMAT ) . getValueAsString ( ) ; } throw new HiveStageCheckedException ( Errors . HIVE_17 , DATA_FORMAT , metadataRecord ) ;


signature:
public class HiveMetastoreUtil { /** * Fill in metadata to Record . This is for new partition creation .
 * Use the { @ code customLocation } flag to mark whether the { @ code location } is a custom one or the default location
 * used by Hive . */
public static Field newPartitionMetadataFieldBuilder ( String database , String tableName , LinkedHashMap < String , String > partitionList , String location , boolean customLocation , HMPDataFormat dataFormat ) throws HiveStageCheckedException { } }

implementation:
LinkedHashMap < String , Field > metadata = new LinkedHashMap < > ( ) ; metadata . put ( VERSION , Field . create ( PARTITION_ADDITION_METADATA_RECORD_VERSION ) ) ; metadata . put ( METADATA_RECORD_TYPE , Field . create ( MetadataRecordType . PARTITION . name ( ) ) ) ; metadata . put ( DATABASE_FIELD , Field . create ( database ) ) ; metadata . put ( TABLE_FIELD , Field . create ( tableName ) ) ; metadata . put ( LOCATION_FIELD , Field . create ( location ) ) ; metadata . put ( CUSTOM_LOCATION , Field . create ( customLocation ) ) ; metadata . put ( DATA_FORMAT , Field . create ( dataFormat . name ( ) ) ) ; // fill in the partition list here
metadata . put ( PARTITION_FIELD , generateInnerFieldFromTheList ( partitionList , PARTITION_NAME , PARTITION_VALUE , false ) ) ; return Field . createListMap ( metadata ) ;


signature:
public class HiveMetastoreUtil { /** * Fill in metadata to Record . This is for new schema creation . */
public static Field newSchemaMetadataFieldBuilder ( String database , String tableName , LinkedHashMap < String , HiveTypeInfo > columnList , LinkedHashMap < String , HiveTypeInfo > partitionTypeList , boolean internal , String location , String avroSchema , HMPDataFormat dataFormat ) throws HiveStageCheckedException { } }

implementation:
LinkedHashMap < String , Field > metadata = new LinkedHashMap < > ( ) ; metadata . put ( VERSION , Field . create ( SCHEMA_CHANGE_METADATA_RECORD_VERSION ) ) ; metadata . put ( METADATA_RECORD_TYPE , Field . create ( MetadataRecordType . TABLE . name ( ) ) ) ; metadata . put ( DATABASE_FIELD , Field . create ( database ) ) ; metadata . put ( TABLE_FIELD , Field . create ( tableName ) ) ; metadata . put ( LOCATION_FIELD , Field . create ( location ) ) ; metadata . put ( DATA_FORMAT , Field . create ( dataFormat . name ( ) ) ) ; // fill in column type list here
metadata . put ( COLUMNS_FIELD , generateInnerFieldFromTheList ( columnList , COLUMN_NAME , TYPE_INFO , true ) ) ; // fill in partition type list here
if ( partitionTypeList != null && ! partitionTypeList . isEmpty ( ) ) { metadata . put ( PARTITION_FIELD , generateInnerFieldFromTheList ( partitionTypeList , PARTITION_NAME , TYPE_INFO , true ) ) ; } metadata . put ( INTERNAL_FIELD , Field . create ( internal ) ) ; metadata . put ( AVRO_SCHEMA , Field . create ( avroSchema ) ) ; return Field . createListMap ( metadata ) ;


signature:
public class HiveMetastoreUtil { /** * Evaluate precision or scale in context of record and given field path . */
private static int resolveScaleOrPrecisionExpression ( String type , ELEval elEval , ELVars variables , String defaultScaleEL , String fieldPath ) throws ELEvalException , HiveStageCheckedException { } }

implementation:
// By default we take the constant given to this method
String value = defaultScaleEL ; // And if so evaluate it
if ( elEval != null ) { value = elEval . eval ( variables , defaultScaleEL , String . class ) ; } // Finally try to parse output as an integer . Failure means that we are unable to calculate proper scale / precision .
try { return Integer . parseInt ( value ) ; } catch ( NumberFormatException e ) { throw new HiveStageCheckedException ( Errors . HIVE_29 , type , fieldPath , defaultScaleEL , value , e ) ; }


signature:
public class HiveMetastoreUtil { /** * Convert a Record to LinkedHashMap . This is for comparing the structure of incoming Record with cache .
 * Since Avro does not support char , short , and date types , it needs to convert the type to corresponding
 * supported types and change the value in record .
 * @ param record incoming Record
 * @ return LinkedHashMap version of record . Key is the column name , and value is column type in HiveType
 * @ throws HiveStageCheckedException
 * @ throws ELEvalException */
public static LinkedHashMap < String , HiveTypeInfo > convertRecordToHMSType ( Record record , ELEval scaleEL , ELEval precisionEL , ELEval commentEL , String scaleExpression , String precisionExpression , String commentExpression , ELVars variables ) throws HiveStageCheckedException , ELEvalException { } }

implementation:
if ( ! record . get ( ) . getType ( ) . isOneOf ( Field . Type . MAP , Field . Type . LIST_MAP ) ) { throw new HiveStageCheckedException ( Errors . HIVE_33 , record . getHeader ( ) . getSourceId ( ) , record . get ( ) . getType ( ) . toString ( ) ) ; } LinkedHashMap < String , HiveTypeInfo > columns = new LinkedHashMap < > ( ) ; Map < String , Field > list = record . get ( ) . getValueAsMap ( ) ; for ( Map . Entry < String , Field > pair : list . entrySet ( ) ) { if ( StringUtils . isEmpty ( pair . getKey ( ) ) ) { throw new HiveStageCheckedException ( Errors . HIVE_01 , "Field name is empty" ) ; } Field currField = pair . getValue ( ) ; switch ( currField . getType ( ) ) { case SHORT : currField = Field . create ( Field . Type . INTEGER , currField . getValue ( ) ) ; break ; case CHAR : currField = Field . create ( currField . getValueAsString ( ) ) ; break ; case DATETIME : currField = Field . create ( Field . Type . STRING , currField . getValue ( ) == null ? null : datetimeFormat . get ( ) . format ( currField . getValueAsDate ( ) ) ) ; break ; case TIME : currField = Field . create ( Field . Type . STRING , currField . getValue ( ) == null ? null : timeFormat . get ( ) . format ( currField . getValueAsTime ( ) ) ) ; break ; default : break ; } // Set current field in the context - used by subsequent ELs ( decimal resolution , comments , . . . )
FieldPathEL . setFieldInContext ( variables , pair . getKey ( ) ) ; String comment = commentEL . eval ( variables , commentExpression , String . class ) ; if ( ! COMMENT_PATTERN . matcher ( comment ) . matches ( ) ) { throw new HiveStageCheckedException ( com . streamsets . pipeline . stage . processor . hive . Errors . HIVE_METADATA_11 , pair . getKey ( ) , comment ) ; } // Update the Field type and value in Record
pair . setValue ( currField ) ; HiveType hiveType = HiveType . getHiveTypeforFieldType ( currField . getType ( ) ) ; HiveTypeInfo hiveTypeInfo ; // Some types requires special checks or alterations
if ( hiveType == HiveType . DECIMAL ) { int precision = resolveScaleOrPrecisionExpression ( "precision" , precisionEL , variables , precisionExpression , pair . getKey ( ) ) ; int scale = resolveScaleOrPrecisionExpression ( "scale" , scaleEL , variables , scaleExpression , pair . getKey ( ) ) ; validateScaleAndPrecision ( pair . getKey ( ) , currField , precision , scale ) ; hiveTypeInfo = hiveType . getSupport ( ) . generateHiveTypeInfoFromRecordField ( currField , comment , precision , scale ) ; // We need to make sure that all java objects have the same scale
if ( currField . getValue ( ) != null ) { pair . setValue ( Field . create ( currField . getValueAsDecimal ( ) . setScale ( scale ) ) ) ; } } else { hiveTypeInfo = hiveType . getSupport ( ) . generateHiveTypeInfoFromRecordField ( currField , comment ) ; } columns . put ( pair . getKey ( ) . toLowerCase ( ) , hiveTypeInfo ) ; } return columns ;


signature:
public class HiveMetastoreUtil { /** * Generate avro schema from column name and type information . typeInfo in 1st parameter
 * needs to contain precision and scale information in the value ( HiveTypeInfo ) .
 * The 2nd parameter qualifiedName will be the name of Avro Schema .
 * @ param typeInfo Record structure
 * @ param qualifiedName qualified name that will be the name of generated avro schema
 * @ return String avro schema
 * @ throws HiveStageCheckedException */
public static String generateAvroSchema ( Map < String , HiveTypeInfo > typeInfo , String qualifiedName ) throws HiveStageCheckedException { } }

implementation:
Utils . checkNotNull ( typeInfo , "Error TypeInfo cannot be null" ) ; // Avro doesn ' t allow " ` " in names , so we ' re dropping those from qualified name
AvroHiveSchemaGenerator gen = new AvroHiveSchemaGenerator ( qualifiedName . replace ( "`" , "" ) ) ; try { return gen . inferSchema ( typeInfo ) ; } catch ( StageException e ) { // So that any error to generate avro schema will result in onRecordErrorException and routed to error lane .
throw new HiveStageCheckedException ( e . getErrorCode ( ) , e . getParams ( ) ) ; }


signature:
public class HiveMetastoreUtil { /** * Checks whether the number of partition columns and names match w . r . t hive .
 * @ param typeInfo { @ link com . streamsets . pipeline . stage . lib . hive . cache . TypeInfoCacheSupport . TypeInfo }
 * @ param partitionValMap Map of partition name to values
 * @ param qualifiedTableName Qualified table name .
 * @ throws HiveStageCheckedException if there is a mismatch w . r . t hive */
public static void validatePartitionInformation ( TypeInfoCacheSupport . TypeInfo typeInfo , LinkedHashMap < String , String > partitionValMap , String qualifiedTableName ) throws HiveStageCheckedException { } }

implementation: Set < String > partitionNamesInHive = typeInfo . getPartitionTypeInfo ( ) . keySet ( ) ; Set < String > partitionNames = partitionValMap . keySet ( ) ; if ( ! ( partitionNamesInHive . size ( ) == partitionNames . size ( ) && partitionNamesInHive . containsAll ( partitionNames ) ) ) { LOG . error ( Utils . format ( "Partition mismatch. In Hive: {}, In Record : {}" , partitionNamesInHive . size ( ) , partitionNames . size ( ) ) ) ; throw new HiveStageCheckedException ( Errors . HIVE_27 , qualifiedTableName ) ; }


signature:
public class HiveMetastoreUtil { /** * Build a partition path for the external table .
 * @ param partitions A list of key - value pair to build a partition path
 * @ return String that represents partition path */
public static String generatePartitionPath ( LinkedHashMap < String , String > partitions ) { } }

implementation: StringBuilder builder = new StringBuilder ( ) ; for ( Map . Entry < String , String > pair : partitions . entrySet ( ) ) { builder . append ( String . format ( PARTITION_PATH , pair . getKey ( ) , pair . getValue ( ) ) ) ; } return builder . toString ( ) ;


signature:
public class HiveMetastoreUtil { /** * Returns the hdfs paths where the avro schema is stored after serializing .
 * Path is appended with current time so as to have an ordering .
 * @ param schemaFolder Schema Folder ( If this starts with ' / ' it is considered absolute )
 * @ return Hdfs Path String . */
public static String serializeSchemaToHDFS ( UserGroupInformation loginUGI , final FileSystem fs , final String location , final String schemaFolder , final String databaseName , final String tableName , final String schemaJson ) throws StageException { } }

implementation:
String folderLocation ; if ( schemaFolder . startsWith ( SEP ) ) { folderLocation = schemaFolder ; } else { folderLocation = location + SEP + schemaFolder ; } final Path schemasFolderPath = new Path ( folderLocation ) ; final String path = folderLocation + SEP + String . format ( AVRO_SCHEMA_FILE_FORMAT , databaseName , tableName , UUID . randomUUID ( ) . toString ( ) ) ; try { loginUGI . doAs ( new PrivilegedExceptionAction < Void > ( ) { @ Override public Void run ( ) throws Exception { if ( ! fs . exists ( schemasFolderPath ) ) { fs . mkdirs ( schemasFolderPath ) ; } Path schemaFilePath = new Path ( path ) ; // This will never happen unless two HMS targets are writing , we will error out for this
// and let user handle this via error record handling .
if ( ! fs . exists ( schemaFilePath ) ) { try ( FSDataOutputStream os = fs . create ( schemaFilePath ) ) { byte [ ] schemaBytes = schemaJson . getBytes ( "UTF-8" ) ; os . write ( schemaBytes , 0 , schemaBytes . length ) ; } } else { LOG . error ( Utils . format ( "Already schema file {} exists in HDFS" , path ) ) ; throw new IOException ( "Already schema file exists" ) ; } return null ; } } ) ; } catch ( Exception e ) { LOG . error ( "Error in Writing Schema to HDFS: " + e . toString ( ) , e ) ; throw new StageException ( Errors . HIVE_18 , path , e . getMessage ( ) ) ; } return path ;


signature:
public class HiveMetastoreUtil { /** * Gets cached { @ link com . streamsets . pipeline . stage . lib . hive . cache . HMSCacheSupport . HMSCacheInfo }
 * from cache . < br >
 * First call getIfPresent to obtain data from local cache . If not exists , load from HMS
 * @ param < T > { @ link HMSCacheSupport . HMSCacheInfo }
 * @ param hmsCache { @ link HMSCache }
 * @ param cacheType Type of cache to load .
 * @ param qualifiedName qualified table name .
 * @ return Cache object if successfully loaded . Null if no data is found in cache .
 * @ throws StageException */
@ SuppressWarnings ( "unchecked" ) public static < T extends HMSCacheSupport . HMSCacheInfo > T getCacheInfo ( HMSCache hmsCache , HMSCacheType cacheType , String qualifiedName , HiveQueryExecutor queryExecutor ) throws StageException { } }

implementation:
HMSCacheSupport . HMSCacheInfo cacheInfo = hmsCache . getIfPresent ( cacheType , qualifiedName ) ; if ( cacheType != HMSCacheType . AVRO_SCHEMA_INFO && cacheInfo == null ) { // Try loading by executing HMS query
cacheInfo = hmsCache . getOrLoad ( cacheType , qualifiedName , queryExecutor ) ; } return ( T ) cacheInfo ;


signature:
public class JdbcGenericRecordWriter { /** * Set parameters and primary keys in query .
 * @ param opCode
 * @ param columnsToParameters
 * @ param record
 * @ param connection
 * @ param statement
 * @ return */
@ VisibleForTesting @ SuppressWarnings ( "unchecked" ) int setParameters ( int opCode , SortedMap < String , String > columnsToParameters , final Record record , final Connection connection , PreparedStatement statement ) throws OnRecordErrorException { } }

implementation:
int paramIdx = 1 ; // Set columns and their value in query . No need to perform this for delete operation .
if ( opCode != OperationType . DELETE_CODE ) { paramIdx = setParamsToStatement ( paramIdx , statement , columnsToParameters , record , connection , opCode ) ; } // Set primary keys in WHERE clause for update and delete operations
if ( opCode != OperationType . INSERT_CODE ) { paramIdx = setPrimaryKeys ( paramIdx , record , statement , opCode ) ; } return paramIdx ;


signature:
public class JdbcGenericRecordWriter { /** * Some databases drivers allow us to figure out which record in a particular batch failed .
 * In the case that we have a list of update counts , we can mark just the record as erroneous .
 * Otherwise we must send the entire batch to error .
 * @ param failedRecords List of Failed Records
 * @ param e BatchUpdateException
 * @ param errorRecords List of error records for this batch */
private void handleBatchUpdateException ( Collection < Record > failedRecords , SQLException e , List < OnRecordErrorException > errorRecords ) throws StageException { } }

implementation: if ( jdbcUtil . isDataError ( getCustomDataSqlStateCodes ( ) , getConnectionString ( ) , e ) ) { String formattedError = JdbcErrors . JDBC_79 . getMessage ( ) ; LOG . error ( formattedError ) ; LOG . debug ( formattedError , e ) ; if ( ! getRollbackOnError ( ) && e instanceof BatchUpdateException && ( ( BatchUpdateException ) e ) . getUpdateCounts ( ) . length > 0 ) { BatchUpdateException bue = ( BatchUpdateException ) e ; int i = 0 ; for ( Record record : failedRecords ) { if ( i >= bue . getUpdateCounts ( ) . length || bue . getUpdateCounts ( ) [ i ] == PreparedStatement . EXECUTE_FAILED ) { errorRecords . add ( new OnRecordErrorException ( record , JDBC_14 , e . getSQLState ( ) , e . getErrorCode ( ) , e . getMessage ( ) , jdbcUtil . formatSqlException ( e ) , e ) ) ; } i ++ ; } } else { for ( Record record : failedRecords ) { errorRecords . add ( new OnRecordErrorException ( record , JDBC_14 , e . getSQLState ( ) , e . getErrorCode ( ) , e . getMessage ( ) , jdbcUtil . formatSqlException ( e ) , e ) ) ; } } } else { handleSqlException ( e ) ; }


signature:
public class SystemApi { /** * Returns ALL SDC Configuration
 * @ return Map < String , Object > */
public Map < String , Object > getConfiguration ( ) throws ApiException { } }

implementation:
Object postBody = null ; byte [ ] postBinaryBody = null ; // create path and map variables
String path = "/v1/system/configuration" . 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 > ( ) ; 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 < Map < String , Object > > ( ) { } ; return apiClient . invokeAPI ( path , "GET" , queryParams , postBody , postBinaryBody , headerParams , formParams , accept , contentType , authNames , returnType ) ;


signature:
public class JdbcBaseRunnable { /** * Builds the Read Context Cache { @ link # tableReadContextCache } */
@ SuppressWarnings ( "unchecked" ) private LoadingCache < TableRuntimeContext , TableReadContext > buildReadContextCache ( CacheLoader < TableRuntimeContext , TableReadContext > tableCacheLoader ) { } }

implementation: CacheBuilder resultSetCacheBuilder = CacheBuilder . newBuilder ( ) . removalListener ( new JdbcTableReadContextInvalidationListener ( ) ) ; if ( tableJdbcConfigBean . batchTableStrategy == BatchTableStrategy . SWITCH_TABLES ) { if ( tableJdbcConfigBean . resultCacheSize > 0 ) { resultSetCacheBuilder = resultSetCacheBuilder . maximumSize ( tableJdbcConfigBean . resultCacheSize ) ; } } else { resultSetCacheBuilder = resultSetCacheBuilder . maximumSize ( 1 ) ; } if ( tableCacheLoader != null ) { return resultSetCacheBuilder . build ( tableCacheLoader ) ; } else { return resultSetCacheBuilder . build ( new JdbcTableReadContextLoader ( connectionManager , offsets , tableJdbcConfigBean . fetchSize , tableJdbcConfigBean . quoteChar . getQuoteCharacter ( ) , tableJdbcELEvalContext , isReconnect ) ) ; }


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

implementation: gaugeMap . put ( THREAD_NAME , Thread . currentThread ( ) . getName ( ) ) ; gaugeMap . put ( STATUS , "" ) ; gaugeMap . put ( TABLES_OWNED_COUNT , tableReadContextCache . size ( ) ) ; gaugeMap . put ( CURRENT_TABLE , "" ) ;


signature:
public class JdbcBaseRunnable { /** * Generate a batch ( looping through as many tables as needed ) until a batch can be generated
 * and then commit offset . */
private void generateBatchAndCommitOffset ( BatchContext batchContext ) { } }

implementation:
int recordCount = 0 ; int eventCount = 0 ; try { while ( tableRuntimeContext == null ) { tableRuntimeContext = tableProvider . nextTable ( threadNumber ) ; if ( tableRuntimeContext == null ) { // small sleep before trying to acquire a table again , to potentially allow a new partition to be
// returned to shared queue or created
final boolean uninterrupted = ThreadUtil . sleep ( ACQUIRE_TABLE_SLEEP_INTERVAL ) ; if ( ! uninterrupted ) { LOG . trace ( "Interrupted which trying to acquire table" ) ; } if ( ! uninterrupted || tableRuntimeContext == null ) { return ; } } } updateGauge ( JdbcBaseRunnable . Status . QUERYING_TABLE ) ; tableReadContext = getOrLoadTableReadContext ( ) ; ResultSet rs = tableReadContext . getResultSet ( ) ; boolean resultSetEndReached = false ; try { updateGauge ( JdbcBaseRunnable . Status . GENERATING_BATCH ) ; while ( recordCount < batchSize ) { if ( rs . isClosed ( ) || ! rs . next ( ) ) { if ( rs . isClosed ( ) ) { LOG . trace ( "ResultSet is closed" ) ; } resultSetEndReached = true ; break ; } createAndAddRecord ( rs , tableRuntimeContext , batchContext ) ; recordCount ++ ; } LOG . trace ( "{} records generated" , recordCount ) ; if ( commonSourceConfigBean . enableSchemaChanges ) { generateSchemaChanges ( batchContext ) ; } tableRuntimeContext . setResultSetProduced ( true ) ; // Reset numSqlErrors if we are able to read result set and add records to the batch context .
numSQLErrors = 0 ; firstSqlException = null ; // If exception happened we do not report anything about no more data event
// We report noMoreData if either evictTableReadContext is true ( result set no more rows ) / record count is 0.
final AtomicBoolean tableFinished = new AtomicBoolean ( false ) ; final AtomicBoolean schemaFinished = new AtomicBoolean ( false ) ; final List < String > schemaFinishedTables = new LinkedList < > ( ) ; tableProvider . reportDataOrNoMoreData ( tableRuntimeContext , recordCount , batchSize , resultSetEndReached , tableFinished , schemaFinished , schemaFinishedTables ) ; if ( tableFinished . get ( ) ) { TableJdbcEvents . createTableFinishedEvent ( context , batchContext , tableRuntimeContext ) ; eventCount ++ ; } if ( schemaFinished . get ( ) ) { TableJdbcEvents . createSchemaFinishedEvent ( context , batchContext , tableRuntimeContext , schemaFinishedTables ) ; eventCount ++ ; } } finally { if ( resultSetEndReached ) { tableReadContext . closeResultSet ( ) ; } handlePostBatchAsNeeded ( resultSetEndReached , recordCount , eventCount , batchContext ) ; } } catch ( SQLException | ExecutionException | StageException | InterruptedException e ) { LOG . error ( "Error happened" , e ) ; // invalidate if the connection is closed
tableReadContextCache . invalidateAll ( ) ; connectionManager . closeConnection ( ) ; // If we have executed post batch that had errored out
if ( tableRuntimeContext != null ) { final TableContext table = tableRuntimeContext . getSourceTableContext ( ) ; // if the currently acquired tableContext is no longer a valid candidate , try re - fetch the table from table provider
if ( commonSourceConfigBean . allowLateTable && ! tableProvider . getActiveRuntimeContexts ( ) . containsKey ( table ) ) { tableRuntimeContext = null ; } } Throwable th = ( e instanceof ExecutionException ) ? e . getCause ( ) : e ; if ( th instanceof SQLException ) { handleSqlException ( ( SQLException ) th ) ; } else if ( e instanceof InterruptedException ) { LOG . error ( "Thread {} interrupted" , gaugeMap . get ( THREAD_NAME ) ) ; } else { handleStageError ( JdbcErrors . JDBC_67 , e ) ; } }


signature:
public class JdbcBaseRunnable { /** * After a batch is generate perform needed operations , generate and commit batch
 * Evict entries from { @ link # tableReadContextCache } , reset { @ link # tableRuntimeContext } */
protected void handlePostBatchAsNeeded ( boolean resultSetEndReached , int recordCount , int eventCount , BatchContext batchContext ) { } }

implementation:
AtomicBoolean shouldEvict = new AtomicBoolean ( resultSetEndReached ) ; // If we read at least one record , it ' s safe to drop the starting offsets , otherwise they have to re - used
if ( recordCount > 0 ) { tableRuntimeContext . getSourceTableContext ( ) . clearStartOffset ( ) ; } // Only process batch if there are records or events
if ( recordCount > 0 || eventCount > 0 ) { TableReadContext tableReadContext = tableReadContextCache . getIfPresent ( tableRuntimeContext ) ; Optional . ofNullable ( tableReadContext ) . ifPresent ( readContext -> { readContext . addProcessingMetrics ( 1 , recordCount ) ; LOG . debug ( "Table {} read batches={} and records={}" , tableRuntimeContext . getQualifiedName ( ) , readContext . getNumberOfBatches ( ) , readContext . getNumberOfRecords ( ) ) ; calculateEvictTableFlag ( shouldEvict , tableReadContext ) ; } ) ; updateGauge ( JdbcBaseRunnable . Status . BATCH_GENERATED ) ; // Process And Commit offsets
if ( tableRuntimeContext . isUsingNonIncrementalLoad ( ) ) { // process the batch now , will handle the offset commit outside this block
context . processBatch ( batchContext ) ; } else { // for incremental ( normal ) mode , the offset was already stored in this map
// by the specific subclass ' s createAndAddRecord method
final String offsetValue = offsets . get ( tableRuntimeContext . getOffsetKey ( ) ) ; context . processBatch ( batchContext , tableRuntimeContext . getOffsetKey ( ) , offsetValue ) ; } } if ( tableRuntimeContext . isUsingNonIncrementalLoad ( ) ) { // for non - incremental mode , the offset is simply a singleton map indicating whether it ' s finished
final String offsetValue = createNonIncrementalLoadOffsetValue ( resultSetEndReached ) ; context . commitOffset ( tableRuntimeContext . getOffsetKey ( ) , offsetValue ) ; } // Make sure we close the result set only when there are no more rows in the result set
if ( shouldEvict . get ( ) ) { // Invalidate so as to fetch a new result set
// We close the result set / statement in Removal Listener
tableReadContextCache . invalidate ( tableRuntimeContext ) ; tableProvider . releaseOwnedTable ( tableRuntimeContext , threadNumber ) ; tableRuntimeContext = null ; } else if ( tableJdbcConfigBean . batchTableStrategy == BatchTableStrategy . SWITCH_TABLES && ! tableRuntimeContext . isUsingNonIncrementalLoad ( ) ) { tableRuntimeContext = null ; } final List < TableRuntimeContext > removedPartitions = tableProvider . getAndClearRemovedPartitions ( ) ; if ( removedPartitions != null && removedPartitions . size ( ) > 0 ) { for ( TableRuntimeContext partition : removedPartitions ) { LOG . debug ( "Removing offset entry for partition {} since it has been removed from the table provider" , partition . getDescription ( ) ) ; context . commitOffset ( partition . getOffsetKey ( ) , null ) ; } }


signature:
public class JdbcBaseRunnable { /** * Initialize the { @ link TableJdbcELEvalContext } before generating a batch */
private static void initTableEvalContextForProduce ( TableJdbcELEvalContext tableJdbcELEvalContext , TableRuntimeContext tableContext , Calendar calendar ) { } }

implementation: tableJdbcELEvalContext . setCalendar ( calendar ) ; tableJdbcELEvalContext . setTime ( calendar . getTime ( ) ) ; tableJdbcELEvalContext . setTableContext ( tableContext ) ;


signature:
public class JdbcBaseRunnable { /** * Wait If needed before a new query is issued . ( Does not wait if the result set is already cached ) */
private void waitIfNeeded ( ) throws InterruptedException { } }

implementation: if ( queryRateLimiter != null ) { updateGauge ( Status . WAITING_FOR_RATE_LIMIT_PERMIT ) ; double waitTime = queryRateLimiter . acquire ( ) ; updateGauge ( Status . ACQUIRED_RATE_LIMIT_PERMIT ) ; gaugeMap . put ( LAST_RATE_LIMIT_WAIT_TIME , waitTime ) ; } else { LOG . debug ( "No query rate limiter in effect; not waiting" ) ; }


signature:
public class JdbcBaseRunnable { /** * Get Or Load { @ link TableReadContext } for { @ link # tableRuntimeContext } */
private TableReadContext getOrLoadTableReadContext ( ) throws ExecutionException , InterruptedException { } }

implementation:
initTableEvalContextForProduce ( tableJdbcELEvalContext , tableRuntimeContext , Calendar . getInstance ( TimeZone . getTimeZone ( ZoneId . of ( tableJdbcConfigBean . timeZoneID ) ) ) ) ; // Check and then if we want to wait for query being issued do that
TableReadContext tableReadContext = tableReadContextCache . getIfPresent ( tableRuntimeContext ) ; LOG . trace ( "Selected table : '{}' for generating records" , tableRuntimeContext . getDescription ( ) ) ; if ( tableReadContext == null ) { // Wait before issuing query ( Optimization instead of waiting during each batch )
waitIfNeeded ( ) ; // Set time before query
initTableEvalContextForProduce ( tableJdbcELEvalContext , tableRuntimeContext , Calendar . getInstance ( TimeZone . getTimeZone ( ZoneId . of ( tableJdbcConfigBean . timeZoneID ) ) ) ) ; tableReadContext = tableReadContextCache . get ( tableRuntimeContext ) ; // Record query time
lastQueryIntervalTime = System . currentTimeMillis ( ) ; } return tableReadContext ;


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

implementation:
Utils . checkNotNull ( offsets , "offsets" ) ; // retrieve file : offset for each directory
for ( FileContext fileContext : fileContexts ) { String offset = offsets . get ( fileContext . getMultiFileInfo ( ) . getFileKey ( ) ) ; LiveFile file = null ; long fileOffset = 0 ; if ( offset != null && ! offset . isEmpty ( ) ) { file = FileContextProviderUtil . getRefreshedLiveFileFromFileOffset ( offset ) ; fileOffset = FileContextProviderUtil . getLongOffsetFromFileOffset ( offset ) ; } fileContext . setStartingCurrentFileName ( file ) ; fileContext . setStartingOffset ( fileOffset ) ; if ( LOG . isTraceEnabled ( ) ) { LOG . trace ( "Setting offset: directory '{}', file '{}', offset '{}'" , fileContext . getMultiFileInfo ( ) . getFileFullPath ( ) , file , fileOffset ) ; } } currentIdx = startingIdx ;


signature:
public class SyslogDecoder { /** * Parse date in RFC 5424 format . Uses an LRU cache to speed up parsing for
 * multiple messages that occur in the same second .
 * @ param tsStr
 * @ return Typical ( for Java ) milliseconds since UNIX epoch */
public static long parseRfc5424Date ( LoadingCache < String , Long > cache , String tsStr ) throws OnRecordErrorException { } }

implementation:
boolean includesTimezone = true ; long ts ; int curPos = 0 ; int msgLen = tsStr . length ( ) ; if ( msgLen <= RFC5424_PREFIX_LEN ) { throw new OnRecordErrorException ( Errors . SYSLOG_09 , tsStr ) ; } String timestampPrefix = tsStr . substring ( curPos , RFC5424_PREFIX_LEN ) ; try { ts = cache . get ( timestampPrefix ) ; } catch ( ExecutionException ex ) { Throwable cause = Throwables . getRootCause ( ex ) ; if ( cause instanceof IllegalArgumentException ) { throw new OnRecordErrorException ( Errors . SYSLOG_05 , cause , timestampPrefix , cause ) ; } else { // I don ' t believe this will ever occur
throw new IllegalStateException ( Utils . format ( Errors . SYSLOG_05 . getMessage ( ) , cause , timestampPrefix ) , cause ) ; } } curPos += RFC5424_PREFIX_LEN ; // look for the optional fractional seconds
if ( tsStr . charAt ( curPos ) == '.' ) { // figure out how many numeric digits
boolean foundEnd = false ; int endMillisPos = curPos + 1 ; if ( msgLen <= endMillisPos ) { throw new OnRecordErrorException ( Errors . SYSLOG_06 , tsStr ) ; } // FIXME : TODO : ensure we handle all bad formatting cases
while ( ! foundEnd && endMillisPos < msgLen ) { char curDigit = tsStr . charAt ( endMillisPos ) ; if ( curDigit >= '0' && curDigit <= '9' ) { endMillisPos ++ ; } else { foundEnd = true ; } } includesTimezone = foundEnd ; if ( ! includesTimezone ) { endMillisPos -- ; } // if they had a valid fractional second , append it rounded to millis
final int fractionalPositions = endMillisPos - ( curPos + 1 ) ; if ( fractionalPositions > 0 ) { long milliseconds = Long . parseLong ( tsStr . substring ( curPos + 1 , endMillisPos ) ) ; if ( fractionalPositions > 3 ) { milliseconds /= Math . pow ( 10 , ( fractionalPositions - 3 ) ) ; } else if ( fractionalPositions < 3 ) { milliseconds *= Math . pow ( 10 , ( 3 - fractionalPositions ) ) ; } ts += milliseconds ; } else { throw new OnRecordErrorException ( Errors . SYSLOG_07 , tsStr ) ; } curPos = endMillisPos ; } // look for timezone
if ( includesTimezone ) { char tzFirst = tsStr . charAt ( curPos ) ; // UTC
if ( tzFirst == 'Z' ) { // no - op
} else if ( tzFirst == '+' || tzFirst == '-' ) { if ( msgLen <= curPos + 5 ) { throw new OnRecordErrorException ( Errors . SYSLOG_08 , tsStr ) ; } int polarity ; if ( tzFirst == '+' ) { polarity = + 1 ; } else { polarity = - 1 ; } char [ ] h = new char [ 5 ] ; for ( int i = 0 ; i < 5 ; i ++ ) { h [ i ] = tsStr . charAt ( curPos + 1 + i ) ; } if ( h [ 0 ] >= '0' && h [ 0 ] <= '9' && h [ 1 ] >= '0' && h [ 1 ] <= '9' && h [ 2 ] == ':' && h [ 3 ] >= '0' && h [ 3 ] <= '9' && h [ 4 ] >= '0' && h [ 4 ] <= '9' ) { try { int hourOffset = Integer . parseInt ( tsStr . substring ( curPos + 1 , curPos + 3 ) ) ; int minOffset = Integer . parseInt ( tsStr . substring ( curPos + 4 , curPos + 6 ) ) ; ts -= polarity * ( ( hourOffset * 60L ) + minOffset ) * 60000L ; } catch ( NumberFormatException nfe ) { throw new OnRecordErrorException ( Errors . SYSLOG_08 , tsStr , nfe ) ; } } else { throw new OnRecordErrorException ( Errors . SYSLOG_08 , tsStr ) ; } } } return ts ;


signature:
public class SyslogDecoder { /** * Parse the RFC3164 date format . This is trickier than it sounds because this
 * format does not specify a year so we get weird edge cases at year
 * boundaries . This implementation tries to " do what I mean " .
 * @ param ts RFC3164 - compatible timestamp to be parsed
 * @ return Typical ( for Java ) milliseconds since the UNIX epoch */
public static long parseRfc3164Time ( String ts ) throws OnRecordErrorException { } }

implementation:
LocalDateTime now = LocalDateTime . now ( ) ; int year = now . getYear ( ) ; ts = TWO_SPACES . matcher ( ts ) . replaceFirst ( " " ) ; LocalDateTime date ; try { MonthDay monthDay = MonthDay . parse ( ts , rfc3164Format ) ; LocalTime time = LocalTime . parse ( ts , rfc3164Format ) ; // this is overly complicated because of the way Java 8 Time API works , as compared to Joda
// essentially , we just want to pull year out of " now " and set all other fields based on
// what was parsed
date = now ; // zero out millis since we aren ' t actually parsing those
date = date . with ( ChronoField . MILLI_OF_SECOND , 0 ) ; // set month and day of month from parsed
date = date . withMonth ( monthDay . getMonthValue ( ) ) . withDayOfMonth ( monthDay . getDayOfMonth ( ) ) ; // set time fields from parsed
date = date . withHour ( time . getHour ( ) ) . withMinute ( time . getMinute ( ) ) . withSecond ( time . getSecond ( ) ) ; } catch ( DateTimeParseException e ) { throw new OnRecordErrorException ( Errors . SYSLOG_10 , ts , e ) ; } // The RFC3164 is a bit weird date format - it contains day and month , but no year . So we have to somehow guess
// the year . The current logic is to provide a sliding window - going 11 months to the past and 1 month to the
// future . If the message is outside of this window , it will have incorrectly guessed year . We go 11 months to the
// past as we ' re expecting that more messages will be from the past ( syslog usually contains historical data ) .
LocalDateTime fixed = date ; if ( fixed . isAfter ( now ) && fixed . minusMonths ( 1 ) . isAfter ( now ) ) { fixed = date . withYear ( year - 1 ) ; } else if ( fixed . isBefore ( now ) && fixed . plusMonths ( 11 ) . isBefore ( now ) ) { fixed = date . withYear ( year + 1 ) ; } date = fixed ; return date . toInstant ( ZoneOffset . UTC ) . toEpochMilli ( ) ;


signature:
public class ThreadUtil { /** * Puts the current thread to sleep for the specified number of milliseconds .
 * If the thread was interrupted before the sleep method is called , the sleep method wont sleep .
 * @ param milliseconds number of milliseconds to sleep .
 * @ return < code > true < / code > if the thread slept uninterrupted for the specified milliseconds , < code > false < / code > if
 * it was interrupted . */
public static boolean sleep ( long milliseconds ) { } }

implementation:
// checking if we got pre - interrupted .
boolean interrupted = Thread . interrupted ( ) ; if ( ! interrupted ) { try { Thread . sleep ( milliseconds ) ; } catch ( InterruptedException ex ) { interrupted = true ; // clearing the interrupt flag
Thread . interrupted ( ) ; } } return ! interrupted ;


signature:
public class Utils { /** * no escaping is supported , no array content printing either . */
public static String format ( String template , Object ... args ) { } }

implementation:
String [ ] templateArr = TEMPLATES . get ( template ) ; if ( templateArr == null ) { // we may have a race condition here but the end result is idempotent
templateArr = prepareTemplate ( template ) ; TEMPLATES . put ( template , templateArr ) ; } StringBuilder sb = new StringBuilder ( template . length ( ) * 2 ) ; for ( int i = 0 ; i < templateArr . length ; i ++ ) { sb . append ( templateArr [ i ] ) ; if ( args != null && ( i < templateArr . length - 1 ) ) { sb . append ( ( i < args . length ) ? args [ i ] : TOKEN ) ; } } return sb . toString ( ) ;


signature:
public class LiveFile { /** * Serializes the < code > LiveFile < / code > as a string .
 * @ return the serialized string representation of the < code > LiveFile < / code > . */
@ SuppressWarnings ( "unchecked" ) public String serialize ( ) { } }

implementation: Map map = new LinkedHashMap ( ) ; map . put ( "path" , path . toString ( ) ) ; map . put ( "headHash" , headHash ) ; map . put ( "headLen" , headLen ) ; map . put ( "inode" , iNode ) ; try { JsonMapper objectMapper = DataCollectorServices . instance ( ) . get ( JsonMapper . SERVICE_KEY ) ; return objectMapper . writeValueAsString ( map ) ; } catch ( Exception ex ) { throw new RuntimeException ( Utils . format ( "Unexpected exception: {}" , ex . toString ( ) ) , ex ) ; }


signature:
public class LiveFile { /** * Deserializes a string representation of a < code > LiveFile < / code > .
 * @ param str the string representation of a < code > LiveFile < / code > .
 * @ return the deserialized < code > LiveFile < / code >
 * @ throws IOException thrown if the string con not be deserialized into a < code > LiveFile < / code > . */
public static LiveFile deserialize ( String str ) throws IOException { } }

implementation: Utils . checkNotNull ( str , "str" ) ; try { JsonMapper objectMapper = DataCollectorServices . instance ( ) . get ( JsonMapper . SERVICE_KEY ) ; Map map = objectMapper . readValue ( str , Map . class ) ; Path path = Paths . get ( ( String ) map . get ( "path" ) ) ; String headHash = ( map . containsKey ( "headHash" ) ) ? ( String ) map . get ( "headHash" ) : "" ; int headLen = ( map . containsKey ( "headLen" ) ) ? ( int ) map . get ( "headLen" ) : 0 ; String inode = ( String ) map . get ( "inode" ) ; return new LiveFile ( path , inode , headHash , headLen ) ; } catch ( RuntimeException | IOException ex ) { throw new IllegalArgumentException ( Utils . format ( "Invalid LiveFile serialized string '{}': {}" , str , ex . toString ( ) ) , ex ) ; }


signature:
public class LiveFile { /** * Refreshes the < code > LiveFile < / code > , if the file was renamed , the path will have the new name .
 * @ return the refreshed file if the file has been renamed , or itself if the file has not been rename or the file
 * does not exist in the directory anymore .
 * @ throws IOException thrown if the LiveFile could not be refreshed */
public LiveFile refresh ( ) throws IOException { } }

implementation: LiveFile refresh = this ; boolean changed ; try { BasicFileAttributes attrs = Files . readAttributes ( path , BasicFileAttributes . class ) ; String iNodeCurrent = attrs . fileKey ( ) . toString ( ) ; int headLenCurrent = ( int ) Math . min ( headLen , attrs . size ( ) ) ; String headHashCurrent = computeHash ( path , headLenCurrent ) ; changed = ! this . iNode . equals ( iNodeCurrent ) || ! this . headHash . equals ( headHashCurrent ) ; } catch ( NoSuchFileException ex ) { changed = true ; } if ( changed ) { try ( DirectoryStream < Path > directoryStream = Files . newDirectoryStream ( path . getParent ( ) ) ) { for ( Path path : directoryStream ) { BasicFileAttributes attrs = Files . readAttributes ( path , BasicFileAttributes . class ) ; String iNode = attrs . fileKey ( ) . toString ( ) ; int headLen = ( int ) Math . min ( this . headLen , attrs . size ( ) ) ; String headHash = computeHash ( path , headLen ) ; if ( iNode . equals ( this . iNode ) && headHash . equals ( this . headHash ) ) { if ( headLen == 0 ) { headLen = ( int ) Math . min ( HEAD_LEN , attrs . size ( ) ) ; headHash = computeHash ( path , headLen ) ; } refresh = new LiveFile ( path , iNode , headHash , headLen ) ; break ; } } } } return refresh ;


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

implementation:
// Validate configuration values and open any required resources .
List < ConfigIssue > issues = super . init ( ) ; errorRecordHandler = new DefaultErrorRecordHandler ( getContext ( ) ) ; Processor . Context context = getContext ( ) ; issues = hikariConfigBean . validateConfigs ( context , issues ) ; if ( hikariConfigBean . getConnectionString ( ) . toLowerCase ( ) . startsWith ( "jdbc:sqlserver" ) && useMultiRowOp ) { issues . add ( getContext ( ) . createConfigIssue ( Groups . JDBC . name ( ) , MULTI_ROW_OP , JdbcErrors . JDBC_57 ) ) ; } if ( dynamicTableName ) { tableNameVars = getContext ( ) . createELVars ( ) ; tableNameEval = context . createELEval ( JdbcUtil . TABLE_NAME ) ; ELUtils . validateExpression ( tableNameTemplate , getContext ( ) , Groups . JDBC . getLabel ( ) , JdbcUtil . TABLE_NAME , JdbcErrors . JDBC_26 , issues ) ; } if ( issues . isEmpty ( ) && null == dataSource ) { try { dataSource = jdbcUtil . createDataSourceForWrite ( hikariConfigBean , schema , tableNameTemplate , caseSensitive , issues , customMappings , getContext ( ) ) ; } catch ( RuntimeException | SQLException | StageException e ) { LOG . debug ( "Could not connect to data source" , e ) ; issues . add ( getContext ( ) . createConfigIssue ( Groups . JDBC . name ( ) , CONNECTION_STRING , JdbcErrors . JDBC_00 , e . toString ( ) ) ) ; } } // If issues is not empty , the UI will inform the user of each configuration issue in the list .
return issues ;


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

implementation:
if ( ! batch . getRecords ( ) . hasNext ( ) ) { // No records - take the opportunity to clean up the cache so that we don ' t hold on to memory indefinitely
cacheCleaner . periodicCleanUp ( ) ; } boolean perRecord = false ; // MS SQL Server does not support returning generateKey after executeBatch
// Instead of executeBatch , do executeUpdate per record
if ( hikariConfigBean . getConnectionString ( ) . toLowerCase ( ) . startsWith ( "jdbc:sqlserver" ) ) { perRecord = true ; } if ( dynamicTableName ) { jdbcUtil . write ( batch , tableNameEval , tableNameVars , tableNameTemplate , recordWriters , errorRecordHandler , perRecord ) ; } else { jdbcUtil . write ( batch . getRecords ( ) , tableNameTemplate , recordWriters , errorRecordHandler , perRecord ) ; } Iterator < Record > it = batch . getRecords ( ) ; while ( it . hasNext ( ) ) { batchMaker . addRecord ( it . next ( ) ) ; }


signature:
public class MongoDBOpLogFieldConverter { /** * This is used by destination which performs CRUD operations .
 * Records sent from MongoOpLog origin have / o field that contains a map of
 * key - value for INSERT and DELETE operation , but
 * UPDATE record has / o2 field which contains _ id and / o / $ set field
 * which contains a map of key - value to update .
 * This method looks into the right field path for UPDATE so that
 * users don ' t need a separate field - column mapping for UPDATE .
 * @ param fieldPath fieldPath in incoming record
 * @ param operation Crud operation code
 * @ return Actual field path in record */
@ Override public String getFieldPath ( String fieldPath , int operation ) { } }

implementation:
if ( operation == OperationType . UPDATE_CODE ) { if ( fieldPath . contains ( ID_FIELD ) ) { // _ id is stored in " / o2 / column _ name for update records . Need to change the fieldpath "
return fieldPath . replace ( OP_FIELD , OP2_FIELD ) ; } else { // column and values are stored in " / o / $ set / column _ name " . Need to change the fieldpath
return fieldPath . replaceFirst ( OP_FIELD , String . format ( "%s/%s" , OP_FIELD , SET_FIELD ) ) ; } } return fieldPath ;