signature:
public class Pattern { /** * Gets the names of all capture groups
 * @ return the list of names */
public List < String > groupNames ( ) { } }

implementation: if ( groupNames == null ) { groupNames = new ArrayList < String > ( groupInfo . keySet ( ) ) ; } return groupNames ;


signature:
public class Pattern { /** * Determines if the character at the specified position
 * of a string is escaped
 * @ param s string to evaluate
 * @ param pos the position of the character to evaluate
 * @ return true if the character is escaped ; otherwise false */
static private boolean isEscapedChar ( String s , int pos ) { } }

implementation: return isSlashEscapedChar ( s , pos ) || isQuoteEscapedChar ( s , pos ) ;


signature:
public class Pattern { /** * Determines if the character at the specified position
 * of a string is escaped with a backslash
 * @ param s string to evaluate
 * @ param pos the position of the character to evaluate
 * @ return true if the character is escaped ; otherwise false */
static private boolean isSlashEscapedChar ( String s , int pos ) { } }

implementation:
// Count the backslashes preceding this position . If it ' s
// even , there is no escape and the slashes are just literals .
// If it ' s odd , one of the slashes ( the last one ) is escaping
// the character at the given position .
int numSlashes = 0 ; while ( pos > 0 && ( s . charAt ( pos - 1 ) == '\\' ) ) { pos -- ; numSlashes ++ ; } return numSlashes % 2 != 0 ;


signature:
public class Pattern { /** * Determines if the character at the specified position
 * of a string is quote - escaped ( between \ \ Q and \ \ E )
 * @ param s string to evaluate
 * @ param pos the position of the character to evaluate
 * @ return true if the character is quote - escaped ; otherwise false */
static private boolean isQuoteEscapedChar ( String s , int pos ) { } }

implementation:
boolean openQuoteFound = false ; boolean closeQuoteFound = false ; // find last non - escaped open - quote
String s2 = s . substring ( 0 , pos ) ; int posOpen = pos ; while ( ( posOpen = s2 . lastIndexOf ( "\\Q" , posOpen - 1 ) ) != - 1 ) { if ( ! isSlashEscapedChar ( s2 , posOpen ) ) { openQuoteFound = true ; break ; } } if ( openQuoteFound ) { // search remainder of string ( after open - quote ) for a close - quote ;
// no need to check that it ' s slash - escaped because it can ' t be
// ( the escape character itself is part of the literal when quoted )
if ( s2 . indexOf ( "\\E" , posOpen ) != - 1 ) { closeQuoteFound = true ; } } return openQuoteFound && ! closeQuoteFound ;


signature:
public class Pattern { /** * Determines if a string ' s character is within a regex character class
 * @ param s string to evaluate
 * @ param pos the position of the character to evaluate
 * @ return true if the character is inside a character class ; otherwise false */
static private boolean isInsideCharClass ( String s , int pos ) { } }

implementation:
boolean openBracketFound = false ; boolean closeBracketFound = false ; // find last non - escaped open - bracket
String s2 = s . substring ( 0 , pos ) ; int posOpen = pos ; while ( ( posOpen = s2 . lastIndexOf ( '[' , posOpen - 1 ) ) != - 1 ) { if ( ! isEscapedChar ( s2 , posOpen ) ) { openBracketFound = true ; break ; } } if ( openBracketFound ) { // search remainder of string ( after open - bracket ) for a close - bracket
String s3 = s . substring ( posOpen , pos ) ; int posClose = - 1 ; while ( ( posClose = s3 . indexOf ( ']' , posClose + 1 ) ) != - 1 ) { if ( ! isEscapedChar ( s3 , posClose ) ) { closeBracketFound = true ; break ; } } } return openBracketFound && ! closeBracketFound ;


signature:
public class Pattern { /** * Determines if the parenthesis at the specified position
 * of a string is for a non - capturing group , which is one of
 * the flag specifiers ( e . g . , ( ? s ) or ( ? m ) or ( ? : pattern ) .
 * If the parenthesis is followed by " ? " , it must be a non -
 * capturing group unless it ' s a named group ( which begins
 * with " ? < " ) . Make sure not to confuse it with the lookbehind
 * construct ( " ? < = " or " ? < ! " ) .
 * @ param s string to evaluate
 * @ param pos the position of the parenthesis to evaluate
 * @ return true if the parenthesis is non - capturing ; otherwise false */
static private boolean isNoncapturingParen ( String s , int pos ) { } }

implementation:
// int len = s . length ( ) ;
boolean isLookbehind = false ; // code - coverage reports show that pos and the text to
// check never exceed len in this class , so it ' s safe
// to not test for it , which resolves uncovered branches
// in Cobertura
/* if ( pos > = 0 & & pos + 4 < len ) */
{ String pre = s . substring ( pos , pos + 4 ) ; isLookbehind = pre . equals ( "(?<=" ) || pre . equals ( "(?<!" ) ; } return /* ( pos > = 0 & & pos + 2 < len ) & & */
s . charAt ( pos + 1 ) == '?' && ( isLookbehind || s . charAt ( pos + 2 ) != '<' ) ;


signature:
public class Pattern { /** * Counts the open - parentheses to the left of a string position ,
 * excluding escaped parentheses
 * @ param s string to evaluate
 * @ param pos ending position of string ; characters to the left
 * of this position are evaluated
 * @ return number of open parentheses */
static private int countOpenParens ( String s , int pos ) { } }

implementation:
java . util . regex . Pattern p = java . util . regex . Pattern . compile ( "\\(" ) ; java . util . regex . Matcher m = p . matcher ( s . subSequence ( 0 , pos ) ) ; int numParens = 0 ; while ( m . find ( ) ) { // ignore parentheses inside character classes : [ 0-9 ( ) a - f ]
// which are just literals
if ( isInsideCharClass ( s , m . start ( ) ) ) { continue ; } // ignore escaped parens
if ( isEscapedChar ( s , m . start ( ) ) ) continue ; if ( ! isNoncapturingParen ( s , m . start ( ) ) ) { numParens ++ ; } } return numParens ;


signature:
public class Pattern { /** * Parses info on named capture groups from a pattern
 * @ param namedPattern regex the regular expression pattern to parse
 * @ return list of group info for all named groups */
static public Map < String , List < GroupInfo > > extractGroupInfo ( String namedPattern ) { } }

implementation:
Map < String , List < GroupInfo > > groupInfo = new LinkedHashMap < String , List < GroupInfo > > ( ) ; java . util . regex . Matcher matcher = NAMED_GROUP_PATTERN . matcher ( namedPattern ) ; while ( matcher . find ( ) ) { int pos = matcher . start ( ) ; // ignore escaped paren
if ( isEscapedChar ( namedPattern , pos ) ) continue ; String name = matcher . group ( INDEX_GROUP_NAME ) ; int groupIndex = countOpenParens ( namedPattern , pos ) ; List < GroupInfo > list ; if ( groupInfo . containsKey ( name ) ) { list = groupInfo . get ( name ) ; } else { list = new ArrayList < GroupInfo > ( ) ; } list . add ( new GroupInfo ( groupIndex , pos ) ) ; groupInfo . put ( name , list ) ; } return groupInfo ;


signature:
public class Pattern { /** * Replaces strings matching a pattern with another string . If the string
 * to be replaced is escaped with a slash , it is skipped .
 * @ param input the string to evaluate
 * @ param pattern the pattern that matches the string to be replaced
 * @ param replacement the string to replace the target
 * @ return the modified string ( original instance of { @ code input } ) */
static private StringBuilder replace ( StringBuilder input , java . util . regex . Pattern pattern , String replacement ) { } }

implementation:
java . util . regex . Matcher m = pattern . matcher ( input ) ; while ( m . find ( ) ) { if ( isEscapedChar ( input . toString ( ) , m . start ( ) ) ) { continue ; } // since we ' re replacing the original string being matched ,
// we have to reset the matcher so that it searches the new
// string
input . replace ( m . start ( ) , m . end ( ) , replacement ) ; m . reset ( input ) ; } return input ;


signature:
public class Pattern { /** * Replaces referenced group names with the reference to the corresponding group
 * index ( e . g . , < b > < code > \ k & lt ; named > < / code > < / b > } to < b > < code > \ k2 < / code > < / b > } ;
 * < b > < code > $ { named } < / code > < / b > to < b > < code > $ 2 < / code > < / b > } ) .
 * This assumes the group names have already been parsed from the pattern .
 * @ param input the string to evaluate
 * @ param pattern the pattern that matches the string to be replaced
 * @ param prefix string to prefix to the replacement ( e . g . , " $ " or " \ \ " )
 * @ return the modified string ( original instance of { @ code input } )
 * @ throws PatternSyntaxException group name was not found */
private StringBuilder replaceGroupNameWithIndex ( StringBuilder input , java . util . regex . Pattern pattern , String prefix ) { } }

implementation:
java . util . regex . Matcher m = pattern . matcher ( input ) ; while ( m . find ( ) ) { if ( isEscapedChar ( input . toString ( ) , m . start ( ) ) ) { continue ; } int index = indexOf ( m . group ( INDEX_GROUP_NAME ) ) ; if ( index >= 0 ) { index ++ ; } else { throw new PatternSyntaxException ( "unknown group name" , input . toString ( ) , m . start ( INDEX_GROUP_NAME ) ) ; } // since we ' re replacing the original string being matched ,
// we have to reset the matcher so that it searches the new
// string
input . replace ( m . start ( ) , m . end ( ) , prefix + index ) ; m . reset ( input ) ; } return input ;


signature:
public class Pattern { /** * Builds a { @ code java . util . regex . Pattern } from a given regular expression
 * pattern ( which may contain named groups ) and flags
 * @ param namedPattern the expression to be compiled
 * @ param flags Match flags , a bit mask that may include :
 * < ul >
 * < li > { @ link java . util . regex . Pattern # CASE _ INSENSITIVE } < / li >
 * < li > { @ link java . util . regex . Pattern # MULTILINE } < / li >
 * < li > { @ link java . util . regex . Pattern # DOTALL } < / li >
 * < li > { @ link java . util . regex . Pattern # UNICODE _ CASE } < / li >
 * < li > { @ link java . util . regex . Pattern # CANON _ EQ } < / li >
 * < li > { @ link java . util . regex . Pattern # UNIX _ LINES } < / li >
 * < li > { @ link java . util . regex . Pattern # LITERAL } < / li >
 * < li > { @ link java . util . regex . Pattern # COMMENTS } < / li >
 * < / ul >
 * @ return the standard { @ code java . util . regex . Pattern } */
private java . util . regex . Pattern buildStandardPattern ( String namedPattern , Integer flags ) { } }

implementation:
// replace the named - group construct with left - paren but
// make sure we ' re actually looking at the construct ( ignore escapes )
StringBuilder s = new StringBuilder ( namedPattern ) ; s = replace ( s , NAMED_GROUP_PATTERN , "(" ) ; s = replaceGroupNameWithIndex ( s , BACKREF_NAMED_GROUP_PATTERN , "\\" ) ; return java . util . regex . Pattern . compile ( s . toString ( ) , flags ) ;


signature:
public class JMXJsonServlet { /** * Process a GET request for the specified resource .
 * @ param request
 * The servlet request we are processing
 * @ param response
 * The servlet response we are creating */
@ Override public void doGet ( HttpServletRequest request , HttpServletResponse response ) { } }

implementation:
try { JsonGenerator jg = null ; String jsonpcb = null ; PrintWriter writer = null ; try { writer = response . getWriter ( ) ; // " callback " parameter implies JSONP outpout
jsonpcb = request . getParameter ( CALLBACK_PARAM ) ; if ( jsonpcb != null ) { response . setContentType ( "application/javascript; charset=utf8" ) ; writer . write ( jsonpcb + "(" ) ; } else { response . setContentType ( "application/json; charset=utf8" ) ; } jg = jsonFactory . createGenerator ( writer ) ; jg . disable ( JsonGenerator . Feature . AUTO_CLOSE_TARGET ) ; jg . useDefaultPrettyPrinter ( ) ; jg . writeStartObject ( ) ; // query per mbean attribute
String getmethod = request . getParameter ( "get" ) ; if ( getmethod != null ) { String [ ] splitStrings = getmethod . split ( "\\:\\:" ) ; if ( splitStrings . length != 2 ) { jg . writeStringField ( "result" , "ERROR" ) ; jg . writeStringField ( "message" , "query format is not as expected." ) ; jg . flush ( ) ; response . setStatus ( HttpServletResponse . SC_BAD_REQUEST ) ; return ; } listBeans ( jg , new ObjectName ( splitStrings [ 0 ] ) , splitStrings [ 1 ] , response ) ; return ; } // query per mbean
String qry = request . getParameter ( "qry" ) ; if ( qry == null ) { qry = "*:*" ; } listBeans ( jg , new ObjectName ( qry ) , null , response ) ; } finally { if ( jg != null ) { jg . close ( ) ; } if ( jsonpcb != null ) { writer . write ( ");" ) ; } if ( writer != null ) { writer . close ( ) ; } } } catch ( IOException e ) { response . setStatus ( HttpServletResponse . SC_INTERNAL_SERVER_ERROR ) ; } catch ( MalformedObjectNameException e ) { response . setStatus ( HttpServletResponse . SC_BAD_REQUEST ) ; }


signature:
public class JMXJsonServlet { /** * - - - - - Private Methods */
private void listBeans ( JsonGenerator jg , ObjectName qry , String attribute , HttpServletResponse response ) throws IOException { } }

implementation:
Set < ObjectName > names = null ; names = mBeanServer . queryNames ( qry , null ) ; jg . writeArrayFieldStart ( "beans" ) ; Iterator < ObjectName > it = names . iterator ( ) ; while ( it . hasNext ( ) ) { ObjectName oname = it . next ( ) ; MBeanInfo minfo ; String code = "" ; Object attributeinfo = null ; try { minfo = mBeanServer . getMBeanInfo ( oname ) ; code = minfo . getClassName ( ) ; String prs = "" ; try { if ( "org.apache.commons.modeler.BaseModelMBean" . equals ( code ) ) { prs = "modelerType" ; code = ( String ) mBeanServer . getAttribute ( oname , prs ) ; } if ( attribute != null ) { prs = attribute ; attributeinfo = mBeanServer . getAttribute ( oname , prs ) ; } } catch ( AttributeNotFoundException e ) { // If the modelerType attribute was not found , the class name is used
// instead .
} catch ( MBeanException e ) { // The code inside the attribute getter threw an exception so
// and fall back on the class name
} catch ( RuntimeException e ) { // For some reason even with an MBeanException available to them
// Runtime exceptionscan still find their way through , so treat them
// the same as MBeanException
} catch ( ReflectionException e ) { // This happens when the code inside the JMX bean ( setter ? ? from the
// java docs ) threw an exception , so
// class name
} } catch ( InstanceNotFoundException e ) { // Ignored for some reason the bean was not found so don ' t output it
continue ; } catch ( IntrospectionException e ) { // This is an internal error , something odd happened with reflection so
continue ; } catch ( ReflectionException e ) { // This happens when the code inside the JMX bean threw an exception , so
continue ; } jg . writeStartObject ( ) ; jg . writeStringField ( "name" , oname . toString ( ) ) ; jg . writeStringField ( "modelerType" , code ) ; if ( ( attribute != null ) && ( attributeinfo == null ) ) { jg . writeStringField ( "result" , "ERROR" ) ; jg . writeStringField ( "message" , "No attribute with name " + attribute + " was found." ) ; jg . writeEndObject ( ) ; jg . writeEndArray ( ) ; jg . close ( ) ; response . setStatus ( HttpServletResponse . SC_NOT_FOUND ) ; return ; } if ( attribute != null ) { writeAttribute ( jg , attribute , attributeinfo ) ; } else { MBeanAttributeInfo attrs [ ] = minfo . getAttributes ( ) ; for ( int i = 0 ; i < attrs . length ; i ++ ) { writeAttribute ( jg , oname , attrs [ i ] ) ; } } jg . writeEndObject ( ) ; } jg . writeEndArray ( ) ;


signature:
public class OracleCDCSource { /** * Refresh the schema for the table if the last update of this table was before the given SCN .
 * Returns true if it was updated , else returns false . */
private boolean refreshSchema ( BigDecimal scnDecimal , SchemaAndTable schemaAndTable ) throws SQLException { } }

implementation: try { if ( ! tableSchemaLastUpdate . containsKey ( schemaAndTable ) || scnDecimal . compareTo ( tableSchemaLastUpdate . get ( schemaAndTable ) ) > 0 ) { if ( containerized ) { try ( Statement switchToPdb = connection . createStatement ( ) ) { switchToPdb . execute ( "ALTER SESSION SET CONTAINER = " + configBean . pdb ) ; } } tableSchemas . put ( schemaAndTable , getTableSchema ( schemaAndTable ) ) ; tableSchemaLastUpdate . put ( schemaAndTable , scnDecimal ) ; return true ; } return false ; } finally { alterSession ( ) ; }


signature:
public class OracleCDCSource { /** * This method needs to get SQL like string with all required schemas and tables .
 * @ param schemaAndTables List of SchemaAndTable objects
 * @ return SQL string of schemas and tables */
@ VisibleForTesting String getListOfSchemasAndTables ( List < SchemaAndTable > schemaAndTables ) { } }

implementation: Map < String , List < String > > schemas = new HashMap < > ( ) ; for ( SchemaAndTable schemaAndTable : schemaAndTables ) { if ( schemas . containsKey ( schemaAndTable . getSchema ( ) ) ) { schemas . get ( schemaAndTable . getSchema ( ) ) . add ( schemaAndTable . getTable ( ) ) ; } else { List < String > tbls = new ArrayList < > ( ) ; tbls . add ( schemaAndTable . getTable ( ) ) ; schemas . put ( schemaAndTable . getSchema ( ) , tbls ) ; } } List < String > queries = new ArrayList < > ( ) ; for ( Map . Entry < String , List < String > > entry : schemas . entrySet ( ) ) { List < String > tables = new ArrayList < > ( ) ; int fromIndex = 0 ; int range = 1000 ; int maxIndex = entry . getValue ( ) . size ( ) ; int toIndex = range < maxIndex ? range : maxIndex ; while ( fromIndex < toIndex ) { tables . add ( Utils . format ( "TABLE_NAME IN ({})" , formatTableList ( entry . getValue ( ) . subList ( fromIndex , toIndex ) ) ) ) ; fromIndex = toIndex ; toIndex = ( toIndex + range ) < maxIndex ? ( toIndex + range ) : maxIndex ; } queries . add ( Utils . format ( "(SEG_OWNER='{}' AND ({}))" , entry . getKey ( ) , String . join ( " OR " , tables ) ) ) ; } return "( " + String . join ( " OR " , queries ) + " )" ;


signature:
public class OracleCDCSource { /** * An element is " expired " if the transaction started before the current window being processed
 * and if no records have actually been sent to the pipeline . If a record has been sent , then a commit was seen ,
 * so it is not expired .
 * @ param entry
 * @ return */
private boolean expired ( Map . Entry < TransactionIdKey , HashQueue < RecordSequence > > entry , LocalDateTime startTime ) { } }

implementation:
return startTime != null && // Can be null if starting from SCN and first batch is not complete yet .
entry . getKey ( ) . txnStartTime . isBefore ( startTime . minusSeconds ( configBean . txnWindow ) ) && entry . getValue ( ) . peek ( ) . seq == 1 ;


signature:
public class SingleLineLiveFileReader { /** * offset will be negative if we are in truncate mode . */
@ Override public long getOffset ( ) { } }

implementation: Utils . checkState ( open , Utils . formatL ( "LiveFileReder for '{}' is not open" , currentFile ) ) ; return ( truncateMode ) ? - offset : offset ;


signature:
public class SingleLineLiveFileReader { /** * returns true if still in truncate mode , false otherwise */
private boolean fastForward ( ) throws IOException { } }

implementation:
try { boolean stillTruncate ; buffer . clear ( ) ; if ( channel . read ( buffer ) > - 1 || isEof ( ) ) { // set the buffer into read from mode
buffer . flip ( ) ; // we have data , lets look for the first EOL in it .
int firstEolIdx = findEndOfFirstLine ( buffer ) ; if ( firstEolIdx > - 1 ) { // set position to position after first EOL
buffer . position ( firstEolIdx + 1 ) ; // set the buffer back into write into mode keeping data after first EOL
buffer . compact ( ) ; stillTruncate = false ; offset = channel . position ( ) - buffer . position ( ) ; } else { // no EOL yet
// whatever was read will be discarded on next next ( ) call
stillTruncate = true ; offset = channel . position ( ) ; } } else { // no data read
// whatever was read will be discarded on next next ( ) call
stillTruncate = true ; offset = channel . position ( ) ; } return stillTruncate ; } catch ( IOException ex ) { closeChannel ( ) ; throw ex ; }


signature:
public class RabbitUtil { /** * - - - - - Private Members - - - - - */
private static void initRabbitConf ( Channel channel , BaseRabbitConfigBean conf ) throws IOException { } }

implementation:
// Channel is always bound to the default exchange . When specified , we must declare the exchange .
for ( RabbitExchangeConfigBean exchange : conf . exchanges ) { channel . exchangeDeclare ( exchange . name , exchange . type . getValue ( ) , exchange . durable , exchange . autoDelete , exchange . declarationProperties ) ; } channel . queueDeclare ( conf . queue . name , conf . queue . durable , conf . queue . exclusive , conf . queue . autoDelete , conf . queue . properties ) ; for ( RabbitExchangeConfigBean exchange : conf . exchanges ) { bindQueue ( channel , conf , exchange ) ; }


signature:
public class PipelineStoreResource { /** * Detached stage APIs */
@ Path ( "/detachedstage" ) @ GET @ ApiOperation ( value = "Returns empty envelope for detached stage." , response = DetachedStageConfigurationJson . class , authorizations = @ Authorization ( value = "basic" ) ) @ Produces ( MediaType . APPLICATION_JSON ) @ RolesAllowed ( { } }

implementation: AuthzRole . CREATOR , AuthzRole . ADMIN , AuthzRole . CREATOR_REMOTE , AuthzRole . ADMIN_REMOTE } ) public Response createDetachedStageEnvelope ( ) throws PipelineException { DetachedStageConfigurationJson detachedStage = new DetachedStageConfigurationJson ( new DetachedStageConfiguration ( ) ) ; return Response . ok ( ) . entity ( detachedStage ) . build ( ) ;


signature:
public class SdcIpcRequestFragmenter { /** * max array size will be limit + 1 ( because the magic byte is being added ) */
static byte [ ] extract ( InputStream is , ByteArrayOutputStream overflowBuffer , int limit ) throws IOException { } }

implementation:
// the inputstream we get has been already stripped of the magic byte if first call
byte [ ] message ; if ( copy ( is , overflowBuffer , limit - overflowBuffer . size ( ) ) ) { // got rest of payload without exceeding the max message size
if ( overflowBuffer . size ( ) == 0 ) { // there is no more payload
message = null ; } else { // extract the rest payload and prefix it with the magic byte
byte [ ] data = overflowBuffer . toByteArray ( ) ; message = new byte [ data . length + 1 ] ; message [ 0 ] = JSON1_MAGIC_NUMBER ; System . arraycopy ( data , 0 , message , 1 , data . length ) ; overflowBuffer . reset ( ) ; } } else { // got partial payload , exceeded the max message size
byte [ ] data = overflowBuffer . toByteArray ( ) ; // find last full record in partial payload
int lastEOL = findEndOfLastLineBeforeLimit ( data , limit ) ; if ( lastEOL == - 1 ) { throw new IOException ( Utils . format ( "Maximum message size '{}' exceeded" , limit ) ) ; } // extract payload up to last EOL and prefix with the magic byte
message = new byte [ lastEOL + 1 ] ; message [ 0 ] = JSON1_MAGIC_NUMBER ; System . arraycopy ( data , 0 , message , 1 , lastEOL ) ; // put back in the stream buffer the portion of the payload that did not make it to the message
overflowBuffer . reset ( ) ; overflowBuffer . write ( data , lastEOL , data . length - lastEOL ) ; } return message ;


signature:
public class ProtoContext { /** * ContextExtensions */
@ Override public RecordReader createRecordReader ( InputStream inputStream , long initialPosition , int maxObjectLen ) throws IOException { } }

implementation: return RecordWriterReaderFactory . createRecordReader ( inputStream , initialPosition , maxObjectLen ) ;


signature:
public class MathEL { /** * We need to support Strings as some information that user might need to deal with is inherently stored in String
 * variables - for example header values or CSV files .
 * The way we ' re processing strings is - if it contains dot , then it ' s assumed to be a double otherwise long . If we for
 * whatever reason fail to parse the number and error is thrown . */
private static Object convertStringToAppropriateNumber ( String value ) { } }

implementation: if ( value . contains ( "." ) ) { return Double . valueOf ( value ) ; } else { return Long . valueOf ( value ) ; }


signature:
public class CollectdParser { /** * Parses a collectd packet " part " .
 * @ param startOffset beginning offset for this part
 * @ param buf buffered packet
 * @ param fields field map for the output record
 * @ return offset after consuming part */
private int parsePart ( int startOffset , ByteBuf buf , Map < String , Field > fields ) throws OnRecordErrorException { } }

implementation:
int offset = startOffset ; int type = buf . getUnsignedShort ( offset ) ; // 0-1
offset += 2 ; final int length = buf . getUnsignedShort ( offset ) ; // 2-3
offset += 2 ; switch ( type ) { case HOST : case PLUGIN : case PLUGIN_INSTANCE : case TYPE : case TYPE_INSTANCE : case MESSAGE : pruneFields ( type ) ; fields . put ( PART_TYPES . get ( type ) , Field . create ( parseString ( offset , length , buf ) ) ) ; offset += length - 4 ; break ; case TIME_HIRES : case INTERVAL_HIRES : if ( type != INTERVAL_HIRES || ! excludeInterval ) { long value = parseNumeric ( offset , buf ) ; if ( convertTime ) { value *= ( Math . pow ( 2 , - 30 ) * 1000 ) ; type = type == TIME_HIRES ? TIME : INTERVAL ; } fields . put ( PART_TYPES . get ( type ) , Field . create ( value ) ) ; } offset += 8 ; break ; case TIME : case INTERVAL : case SEVERITY : if ( type != INTERVAL || ! excludeInterval ) { fields . put ( PART_TYPES . get ( type ) , Field . create ( parseNumeric ( offset , buf ) ) ) ; } offset += 8 ; break ; case VALUES : offset = parseValues ( offset , buf ) ; startNewRecord ( ) ; break ; case SIGNATURE : if ( ! verifySignature ( offset , length , buf ) ) { throw new OnRecordErrorException ( Errors . COLLECTD_02 ) ; } offset += length - 4 ; break ; case ENCRYPTION : String user = parseUser ( offset , buf ) ; offset += ( 2 + user . length ( ) ) ; byte [ ] iv = parseIv ( offset , buf ) ; offset += 16 ; decrypt ( offset , length , buf , user , iv ) ; // Skip the checksum and continue processing .
offset += 20 ; break ; default : // Don ' t recognize this part type , so skip it
LOG . warn ( "Unrecognized part type: {}" , type ) ; offset += length - 4 ; break ; } return offset ;


signature:
public class CollectdParser { /** * Parses the value part of the packet where metrics are located
 * @ param startOffset beginning offset for this part
 * @ param buf buffered packet
 * @ return offset after consuming part */
private int parseValues ( int startOffset , ByteBuf buf ) throws OnRecordErrorException { } }

implementation:
int offset = startOffset ; // N Values
// For each Value :
// 1 byte data type code
int numValues = buf . getUnsignedShort ( offset ) ; // 4-5
offset += 2 ; List < Byte > types = new ArrayList < > ( numValues ) ; while ( numValues -- > 0 ) { types . add ( buf . getByte ( offset ) ) ; offset += 1 ; } for ( int i = 0 ; i < types . size ( ) ; i ++ ) { Byte type = types . get ( i ) ; String label = getValueLabel ( i , type ) ; switch ( type ) { case COUNTER : fields . put ( label , Field . create ( buf . getUnsignedInt ( offset ) ) ) ; offset += 8 ; break ; case GAUGE : fields . put ( label , Field . create ( buf . order ( ByteOrder . LITTLE_ENDIAN ) . getDouble ( offset ) ) ) ; offset += 8 ; break ; case DERIVE : fields . put ( label , Field . create ( buf . getLong ( offset ) ) ) ; offset += 8 ; break ; case ABSOLUTE : fields . put ( label , Field . create ( buf . getUnsignedInt ( offset ) ) ) ; offset += 8 ; break ; default : // error
throw new OnRecordErrorException ( Errors . COLLECTD_01 , type ) ; } } return offset ;


signature:
public class NetflowV9Decoder { /** * Adapted from https : / / stackoverflow . com / a / 7932774/375670
 * @ param bytes the raw bytes to interpret as an unsigned short
 * @ return an INTEGER field with the unsigned short value */
public static Field getUnsignedShortField ( byte [ ] bytes ) { } }

implementation: Utils . checkState ( bytes . length == 2 , "2 bytes required to parse an unsigned short" ) ; final short shortVal = Shorts . fromByteArray ( bytes ) ; int intVal = shortVal >= 0 ? shortVal : 0x10000 + shortVal ; return Field . create ( intVal ) ;


signature:
public class AvroHiveSchemaGenerator { /** * It takes a record structure in < String , HiveTypeInfo > format .
 * Generate a schema and return in String .
 * @ param record : record structure
 * @ return String representation of Avro schema .
 * @ throws StageException : If record contains unsupported type */
@ Override public String inferSchema ( Map < String , HiveTypeInfo > record ) throws StageException { } }

implementation:
Map < String , Schema > fields = new LinkedHashMap < > ( ) ; for ( Map . Entry < String , HiveTypeInfo > pair : record . entrySet ( ) ) { if ( ! HiveMetastoreUtil . validateObjectName ( pair . getKey ( ) ) ) { throw new HiveStageCheckedException ( Errors . HIVE_30 , pair . getKey ( ) ) ; } Schema columnSchema = Schema . createUnion ( ImmutableList . of ( Schema . create ( Schema . Type . NULL ) , traverse ( pair ) ) ) ; // We always set default value to null
columnSchema . addProp ( "default" , NullNode . getInstance ( ) ) ; fields . put ( pair . getKey ( ) , columnSchema ) ; } Schema schema = buildSchema ( fields ) ; return schema . toString ( ) ;


signature:
public class TableContextUtil { /** * Our own implementation of JDBCType . valueOf ( ) that won ' t throw an exception in case of unknown type .
 * @ param jdbcType
 * @ return */
public static String nameForType ( DatabaseVendor vendor , int jdbcType ) { } }

implementation: for ( JDBCType sqlType : JDBCType . class . getEnumConstants ( ) ) { if ( jdbcType == sqlType . getVendorTypeNumber ( ) ) return sqlType . name ( ) ; } switch ( vendor ) { case ORACLE : switch ( jdbcType ) { case - 101 : return "TIMESTAMP WITH TIME ZONE" ; case - 102 : return "TIMESTAMP WITH LOCAL TIME ZONE" ; } break ; } return "Unknown" ;


signature:
public class TableContextUtil { /** * Returns qualified table name ( schema . table name )
 * @ param schema schema name , can be null
 * @ param tableName table name
 * @ return qualified table name if schema is not null and tableName alone if schema is null . */
public static String getQualifiedTableName ( String schema , String tableName ) { } }

implementation: return StringUtils . isEmpty ( schema ) ? tableName : schema + "." + tableName ;


signature:
public class TableContextUtil { /** * Returns quoted qualified table name ( schema . table name ) based on quoteChar
 * @ param schema schema name , can be null
 * @ param tableName table name
 * @ param qC quote character to be attached before and after table and schema name .
 * @ return qualified table name if schema is not null and tableName alone if schema is null . */
public static String getQuotedQualifiedTableName ( String schema , String tableName , String qC ) { } }

implementation: String quotedTableName = String . format ( OffsetQueryUtil . QUOTED_NAME , qC , tableName , qC ) ; return StringUtils . isEmpty ( schema ) ? quotedTableName : String . format ( OffsetQueryUtil . QUOTED_NAME , qC , schema , qC ) + "." + quotedTableName ;


signature:
public class TableContextUtil { /** * Evaluate ELs in Initial offsets as needed and populate the final String representation of initial offsets
 * in { @ param offsetColumnToStartOffset } */
private void populateInitialOffset ( PushSource . Context context , List < Stage . ConfigIssue > issues , Map < String , String > configuredColumnToInitialOffset , TableJdbcELEvalContext tableJdbcELEvalContext , Map < String , String > offsetColumnToStartOffset ) throws StageException { } }

implementation: for ( Map . Entry < String , String > partitionColumnInitialOffsetEntry : configuredColumnToInitialOffset . entrySet ( ) ) { String value ; try { value = tableJdbcELEvalContext . evaluateAsString ( "offsetColumnToInitialOffsetValue" , partitionColumnInitialOffsetEntry . getValue ( ) ) ; if ( value == null ) { issues . add ( context . createConfigIssue ( Groups . TABLE . name ( ) , TableJdbcConfigBean . TABLE_CONFIG , JdbcErrors . JDBC_73 , partitionColumnInitialOffsetEntry . getValue ( ) , Utils . format ( "Expression returned date as null. Check Expression" ) ) ) ; return ; } } catch ( ELEvalException e ) { issues . add ( context . createConfigIssue ( Groups . TABLE . name ( ) , TableJdbcConfigBean . TABLE_CONFIG , JdbcErrors . JDBC_73 , partitionColumnInitialOffsetEntry . getValue ( ) , e ) ) ; return ; } offsetColumnToStartOffset . put ( partitionColumnInitialOffsetEntry . getKey ( ) , value ) ; }


signature:
public class TableContextUtil { /** * Lists all tables matching the { @ link TableConfigBean } and creates a table context for each table .
 * @ param connection JDBC connection
 * @ param tableConfigBean { @ link TableConfigBean }
 * @ return Map of qualified table name to Table Context
 * @ throws SQLException If list tables call fails
 * @ throws StageException if partition configuration is not correct . */
public Map < String , TableContext > listTablesForConfig ( DatabaseVendor vendor , PushSource . Context context , List < Stage . ConfigIssue > issues , Connection connection , TableConfigBean tableConfigBean , TableJdbcELEvalContext tableJdbcELEvalContext , QuoteChar quoteChar ) throws SQLException , StageException { } }

implementation: Map < String , TableContext > tableContextMap = new LinkedHashMap < > ( ) ; Pattern tableExclusion = StringUtils . isEmpty ( tableConfigBean . tableExclusionPattern ) ? null : Pattern . compile ( tableConfigBean . tableExclusionPattern ) ; Pattern schemaExclusion = StringUtils . isEmpty ( tableConfigBean . schemaExclusionPattern ) ? null : Pattern . compile ( tableConfigBean . schemaExclusionPattern ) ; try ( ResultSet rs = jdbcUtil . getTableAndViewMetadata ( connection , tableConfigBean . schema , tableConfigBean . tablePattern ) ) { while ( rs . next ( ) ) { String schemaName = rs . getString ( TABLE_METADATA_TABLE_SCHEMA_CONSTANT ) ; String tableName = rs . getString ( TABLE_METADATA_TABLE_NAME_CONSTANT ) ; if ( ( tableExclusion == null || ! tableExclusion . matcher ( tableName ) . matches ( ) ) && ( schemaExclusion == null || ! schemaExclusion . matcher ( schemaName ) . matches ( ) ) ) { TableContext tableContext = createTableContext ( vendor , context , issues , connection , schemaName , tableName , tableConfigBean , tableJdbcELEvalContext , quoteChar ) ; if ( tableContext != null ) { tableContextMap . put ( getQualifiedTableName ( schemaName , tableName ) , tableContext ) ; } } } } return tableContextMap ;


signature:
public class JdbcBaseRecordWriter { /** * Access the database , obtain a list of primary key columns , and store them in primaryKeyColumns .
 * If table has no primary keys , primaryKeyColumns stays empty .
 * @ throws StageException */
void lookupPrimaryKeys ( ) throws StageException { } }

implementation: Connection connection = null ; try { connection = dataSource . getConnection ( ) ; primaryKeyColumns = jdbcUtil . getPrimaryKeys ( connection , schema , tableName ) ; } catch ( SQLException e ) { String formattedError = jdbcUtil . formatSqlException ( e ) ; LOG . error ( formattedError , e ) ; throw new StageException ( JdbcErrors . JDBC_17 , tableName , formattedError ) ; } finally { if ( connection != null ) { try { connection . close ( ) ; } catch ( SQLException e ) { String formattedError = jdbcUtil . formatSqlException ( e ) ; LOG . error ( formattedError , e ) ; } } }


signature:
public class JdbcBaseRecordWriter { /** * Access database and obtain the metadata for the table .
 * Store columnName and " / columnName " to the columnsToFields map as a default column - to - field mapping .
 * Store columnName and " ? " to columnsToParameters map as a default column - to - value mapping .
 * They will be updated later in createCustomFieldMappings ( ) .
 * @ throws StageException */
private void createDefaultFieldMappings ( ) throws StageException { } }

implementation:
try ( Connection connection = dataSource . getConnection ( ) ) { try ( ResultSet res = jdbcUtil . getTableMetadata ( connection , schema , tableName ) ) { if ( ! res . next ( ) ) { throw new StageException ( JdbcErrors . JDBC_16 , getTableName ( ) ) ; } } try ( ResultSet columns = jdbcUtil . getColumnMetadata ( connection , schema , tableName ) ) { while ( columns . next ( ) ) { String columnName = columns . getString ( COLUMN_NAME ) ; columnsToFields . put ( columnName , "/" + columnName ) ; // Default implicit field mappings
columnsToParameters . put ( columnName , "?" ) ; columnType . put ( columnName , columns . getInt ( DATA_TYPE ) ) ; } } } catch ( SQLException e ) { String errorMessage = jdbcUtil . formatSqlException ( e ) ; LOG . error ( errorMessage ) ; LOG . debug ( errorMessage , e ) ; throw new StageException ( JdbcErrors . JDBC_09 , tableName ) ; }


signature:
public class JdbcBaseRecordWriter { /** * Use " Field to Column Mapping " option obtained from configuration and update
 * columnsToFields and columnsToParameters . */
private void createCustomFieldMappings ( ) { } }

implementation: for ( JdbcFieldColumnParamMapping mapping : customMappings ) { LOG . debug ( "Custom mapping field {} to column {}" , mapping . field , mapping . columnName ) ; if ( columnsToFields . containsKey ( mapping . columnName ) ) { LOG . debug ( "Mapping field {} to column {}" , mapping . field , mapping . columnName ) ; columnsToFields . put ( mapping . columnName , mapping . field ) ; columnsToParameters . put ( mapping . columnName , mapping . paramValue ) ; } }


signature:
public class JdbcBaseRecordWriter { /** * spec requires a string name for a data type , rather than just an enum . */
static String getSQLTypeName ( Field . Type type ) throws OnRecordErrorException { } }

implementation: switch ( type ) { case BOOLEAN : return "BOOLEAN" ; case CHAR : return "CHAR" ; case BYTE : return "BINARY" ; case SHORT : return "SMALLINT" ; case INTEGER : return "INTEGER" ; case LONG : return "BIGINT" ; case FLOAT : return "FLOAT" ; case DOUBLE : return "DOUBLE" ; case DATE : return "DATE" ; case TIME : return "TIME" ; case DATETIME : return "TIMESTAMP" ; case DECIMAL : return "DECIMAL" ; case STRING : return "VARCHAR" ; case BYTE_ARRAY : return "VARBINARY" ; case LIST_MAP : case MAP : throw new OnRecordErrorException ( JdbcErrors . JDBC_05 , "Unsupported list or map type: MAP" ) ; case LIST : return "ARRAY" ; default : throw new OnRecordErrorException ( JdbcErrors . JDBC_05 , "Unsupported type: " + type . name ( ) ) ; }


signature:
public class JdbcBaseRecordWriter { /** * Table this writer will write to .
 * @ return table name */
protected String getTableName ( ) { } }

implementation: if ( ! Strings . isNullOrEmpty ( schema ) ) { if ( caseSensitive ) { return "\"" + schema + "\"." + "\"" + tableName + "\"" ; } else { return schema + "." + tableName ; } } if ( caseSensitive ) { return "\"" + tableName + "\"" ; } return tableName ;


signature:
public class JdbcBaseRecordWriter { /** * Set primary key values to query . This is called only for UPDATE and DELETE operations .
 * If primary key value is missing in record , it throws OnRecordErrorException .
 * @ param index
 * @ param record
 * @ param statement
 * @ param opCode
 * @ return
 * @ throws OnRecordErrorException */
int setPrimaryKeys ( int index , final Record record , PreparedStatement statement , int opCode ) throws OnRecordErrorException { } }

implementation: for ( String key : getPrimaryKeyColumns ( ) ) { Field field = record . get ( recordReader . getFieldPath ( key , getColumnsToFields ( ) , opCode ) ) ; if ( field == null ) { LOG . error ( "Primary key {} is missing in record" , key ) ; throw new OnRecordErrorException ( record , JdbcErrors . JDBC_19 , key ) ; } Object value = field . getValue ( ) ; try { statement . setObject ( index , value , getColumnType ( key ) ) ; } catch ( SQLException ex ) { LOG . error ( "SQLException thrown: {}" , ex . getMessage ( ) ) ; throw new OnRecordErrorException ( record , JdbcErrors . JDBC_19 , key , ex ) ; } ++ index ; } return index ;


signature:
public class JdbcBaseRecordWriter { /** * This is an error that is not due to bad input record and should throw a StageException
 * once we format the error .
 * @ param e SQLException
 * @ throws StageException */
void handleSqlException ( SQLException e ) throws StageException { } }

implementation: String formattedError = jdbcUtil . formatSqlException ( e ) ; LOG . error ( formattedError , e ) ; throw new StageException ( JdbcErrors . JDBC_14 , e . getSQLState ( ) , e . getErrorCode ( ) , e . getMessage ( ) , formattedError , e ) ;


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

implementation: return recordReader . getOperationFromRecord ( record , defaultOpCode , unsupportedAction , errorRecords ) ;


signature:
public class SchemaTableClassifier { /** * Split records according to tuples ( schemaName , tableName )
 * @ param batch batch of SDC records
 * @ return batch partitions grouped by the tuples ( schemaName , tableName )
 * @ throws OnRecordErrorException */
public Multimap < SchemaAndTable , Record > classify ( Batch batch ) throws OnRecordErrorException { } }

implementation: Multimap < SchemaAndTable , Record > partitions = ArrayListMultimap . create ( ) ; Iterator < Record > batchIterator = batch . getRecords ( ) ; while ( batchIterator . hasNext ( ) ) { Record record = batchIterator . next ( ) ; String schemaName = schemaNameExpr ; String tableName = tableNameExpr ; if ( dynamicSchemaName ) { try { RecordEL . setRecordInContext ( schemaNameVars , record ) ; schemaName = schemaNameEval . eval ( schemaNameVars , schemaNameExpr , String . class ) ; LOG . debug ( "Expression '{}' is evaluated to '{}' : " , schemaNameExpr , schemaName ) ; } catch ( ELEvalException e ) { LOG . error ( "Failed to evaluate expression '{}' : " , schemaNameExpr , e . toString ( ) , e ) ; throw new OnRecordErrorException ( record , e . getErrorCode ( ) , e . getParams ( ) ) ; } } if ( dynamicTableName ) { try { RecordEL . setRecordInContext ( tableNameVars , record ) ; tableName = tableNameEval . eval ( tableNameVars , tableNameExpr , String . class ) ; LOG . debug ( "Expression '{}' is evaluated to '{}' : " , tableNameExpr , tableName ) ; } catch ( ELEvalException e ) { LOG . error ( "Failed to evaluate expression '{}' : " , tableNameExpr , e . toString ( ) , e ) ; throw new OnRecordErrorException ( record , e . getErrorCode ( ) , e . getParams ( ) ) ; } } SchemaAndTable partitionKey = new SchemaAndTable ( schemaName , tableName ) ; partitions . put ( partitionKey , record ) ; } return partitions ;


signature:
public class SourcePipe { /** * Process method for Push source that will give control of the execution to the origin .
 * @ param offsets Offsets from last execution
 * @ param batchSize Maximal configured batch size */
public void process ( Map < String , String > offsets , int batchSize , ReportErrorDelegate reportErrorDelegate ) throws StageException , PipelineRuntimeException { } }

implementation: this . reportErrorDelegate = reportErrorDelegate ; getStage ( ) . setReportErrorDelegate ( this ) ; try { MDC . put ( LogConstants . STAGE , getStage ( ) . getInfo ( ) . getInstanceName ( ) ) ; getStage ( ) . execute ( offsets , batchSize ) ; } finally { MDC . put ( LogConstants . STAGE , "" ) ; }


signature:
public class SourcePipe { /** * Called by PipelineRunner when push origin started a new batch to prepare context for it .
 * @ param batchContext BatchContext associated with the starting batch . */
public void prepareBatchContext ( BatchContextImpl batchContext ) { } }

implementation:
PipeBatch pipeBatch = batchContext . getPipeBatch ( ) ; // Start stage in the pipe batch and persist reference to batch maker in the batch context
BatchMakerImpl batchMaker = pipeBatch . startStage ( this ) ; batchContext . setBatchMaker ( batchMaker ) ; batchContext . setOriginStageName ( getStage ( ) . getInfo ( ) . getInstanceName ( ) , getStage ( ) . getInfo ( ) . getLabel ( ) ) ;


signature:
public class SourcePipe { /** * Finish batch from the origin ' s perspective .
 * @ param batchContext Batch context enriched by prepareBatchContext
 * @ return Map with statistics that are usually stored inside the Pipe object itself . */
public Map < String , Object > finishBatchContext ( BatchContextImpl batchContext ) throws StageException { } }

implementation: return finishBatchAndCalculateMetrics ( batchContext . getStartTime ( ) , batchContext . getPipeBatch ( ) , ( BatchMakerImpl ) batchContext . getBatchMaker ( ) , batchContext . getPipeBatch ( ) . getBatch ( this ) , batchContext . getPipeBatch ( ) . getErrorSink ( ) , batchContext . getPipeBatch ( ) . getEventSink ( ) , null ) ;


signature:
public class FieldFlattenerProcessor { /** * Flatten the entire record to one giant map */
private Map < String , Field > flattenEntireRecord ( Field rootField ) { } }

implementation: Map < String , Field > ret = new LinkedHashMap < > ( ) ; switch ( rootField . getType ( ) ) { case MAP : case LIST_MAP : flattenMap ( "" , rootField . getValueAsMap ( ) , ret ) ; break ; case LIST : flattenList ( "" , rootField . getValueAsList ( ) , ret ) ; break ; default : break ; } return ret ;


signature:
public class DetachedStageRuntime { /** * Create new instance of detached stage runtime . */
public static < T > DetachedStageRuntime < ? extends T > create ( StageBean bean , Stage . Info info , Stage . Context context , Class < T > klass ) { } }

implementation: switch ( bean . getDefinition ( ) . getType ( ) ) { case PROCESSOR : return new DetachedStageRuntime . DetachedProcessor ( bean , info , context ) ; case TARGET : case EXECUTOR : return new DetachedStageRuntime . DetachedTarget ( bean , info , context ) ; default : throw new RuntimeException ( "Unsupported stage type: " + bean . getDefinition ( ) . getType ( ) ) ; }


signature:
public class SampleExecutor { /** * Executes an action for given record .
 * @ param record the record that will parametrize the action
 * @ throws OnRecordErrorException when action can ' t be executed */
private void execute ( 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 : execute action


signature:
public class JdbcMySqlBinLogRecordReader { /** * Records from MySQL BinLog origin have a bit unique structure .
 * Records for Insert and update operations have a field path " / Data " , which is a map
 * of all column names as key and values stored in DB as value .
 * Record for Delete operation don ' t have / Data field . Instead it has a filed path " / OldData " ,
 * which store the original data in table . So need to look into / OldData when operation is delete .
 * We expect user to configure / Data as a filed path in column - filed mapping configuration .
 * @ param record
 * @ param op
 * @ param parameters
 * @ param columnsToFields
 * @ return */
@ Override public SortedMap < String , String > getColumnsToParameters ( final Record record , int op , Map < String , String > parameters , Map < String , String > columnsToFields ) { } }

implementation: SortedMap < String , String > columnsToParameters = new TreeMap < > ( ) ; for ( Map . Entry < String , String > entry : columnsToFields . entrySet ( ) ) { String columnName = entry . getKey ( ) ; String fieldPath = entry . getValue ( ) ; if ( op == OperationType . DELETE_CODE ) { fieldPath = fieldPath . replace ( DATA_FIELD , OLD_DATA_FIELD ) ; } if ( record . has ( fieldPath ) ) { columnsToParameters . put ( columnName , parameters . get ( columnName ) ) ; } } return columnsToParameters ;


signature:
public class JdbcMySqlBinLogRecordReader { /** * Return a fieldpath for the corresponding column name . The columnToField Map
 * contains column - to - field mapping , so simply get fieldpath for the column name
 * if operation is INSERT or UPDATE .
 * We replace / Data to / OldData for delete operation because that ' s where records
 * store primary key info .
 * @ param columnName column name
 * @ param columnsToField mapping of column name to field path
 * @ param op Not used here but is used in subclass .
 * @ return */
@ Override String getFieldPath ( String columnName , Map < String , String > columnsToField , int op ) { } }

implementation:
if ( op == OperationType . DELETE_CODE ) { String fieldPath = columnsToField . get ( columnName ) ; if ( fieldPath == null ) { LOG . error ( "Column name {} is not defined in column-filed mapping" , columnName ) ; return null ; } return fieldPath . replace ( "/Data" , "/OldData" ) ; } // For insert and update , ok to use the field name set by column - field mapping
return columnsToField . get ( columnName ) ;


signature:
public class OffsetQueryUtil { /** * Build query using the lastOffset which is of the form ( < column1 > = < value1 > : : < column2 > = < value2 > : : < column3 > = < value3 > )
 * @ param tableRuntimeContext Context for the current table .
 * @ param lastOffset the last offset for this particular table
 * @ return A query to execute for the current batch . */
public static Pair < String , List < Pair < Integer , String > > > buildAndReturnQueryAndParamValToSet ( TableRuntimeContext tableRuntimeContext , String lastOffset , String quoteChar , TableJdbcELEvalContext tableJdbcELEvalContext ) throws ELEvalException { } }

implementation:
final TableContext tableContext = tableRuntimeContext . getSourceTableContext ( ) ; StringBuilder queryBuilder = new StringBuilder ( ) ; List < Pair < Integer , String > > paramValueToSet = new ArrayList < > ( ) ; queryBuilder . append ( buildBaseTableQuery ( tableRuntimeContext , quoteChar ) ) ; Map < String , String > storedTableToOffset = getColumnsToOffsetMapFromOffsetFormat ( lastOffset ) ; final boolean noStoredOffsets = storedTableToOffset . isEmpty ( ) ; // Determines whether an initial offset is specified in the config and there is no stored offset .
boolean isOffsetOverriden = tableContext . isOffsetOverriden ( ) && noStoredOffsets ; OffsetComparison minComparison = OffsetComparison . GREATER_THAN ; Map < String , String > offset = null ; if ( isOffsetOverriden ) { // Use the offset in the configuration
offset = tableContext . getOffsetColumnToStartOffset ( ) ; } else if ( tableRuntimeContext . isPartitioned ( ) && noStoredOffsets ) { // use partitioned starting offsets
offset = tableRuntimeContext . getStartingPartitionOffsets ( ) ; minComparison = OffsetComparison . GREATER_THAN_OR_EQUALS ; } else { // if offset is available
// get the stored offset ( which is of the form partitionName = value ) and strip off ' offsetColumns = ' prefix
// else null
// offset = storedOffsets ;
offset = storedTableToOffset ; } Map < String , String > maxOffsets = new HashMap < > ( ) ; if ( tableRuntimeContext . isPartitioned ( ) ) { maxOffsets . putAll ( tableRuntimeContext . getMaxPartitionOffsets ( ) ) ; } List < String > finalAndConditions = new ArrayList < > ( ) ; // Apply last offset conditions
if ( offset != null && ! offset . isEmpty ( ) ) { List < String > finalOrConditions = new ArrayList < > ( ) ; List < String > preconditions = new ArrayList < > ( ) ; List < Pair < Integer , String > > preconditionParamVals = new ArrayList < > ( ) ; // For partition columns p1 , p2 and p3 with offsets o1 , o2 and o3 respectively , the query will look something like
// select * from tableName where ( p1 > o1 ) or ( p1 = o1 and p2 > o2 ) or ( p1 = o1 and p2 = o2 and p3 > o3 ) order by p1 , p2 , p3.
for ( String partitionColumn : tableContext . getOffsetColumns ( ) ) { int partitionSqlType = tableContext . getOffsetColumnType ( partitionColumn ) ; String partitionOffset = offset . get ( partitionColumn ) ; String thisPartitionColumnMax = null ; boolean hasMaxOffset = false ; // add max value for partition column ( if applicable )
if ( maxOffsets . containsKey ( partitionColumn ) ) { final String maxOffset = maxOffsets . get ( partitionColumn ) ; if ( ! Strings . isNullOrEmpty ( maxOffset ) ) { Pair < Integer , String > paramValForCurrentOffsetColumnMax = Pair . of ( partitionSqlType , maxOffset ) ; // add for current partition column max value
paramValueToSet . add ( paramValForCurrentOffsetColumnMax ) ; thisPartitionColumnMax = getConditionForPartitionColumn ( partitionColumn , OffsetComparison . LESS_THAN , preconditions , quoteChar ) ; hasMaxOffset = true ; } } final String thisPartitionColumnMin = getConditionForPartitionColumn ( partitionColumn , minComparison , preconditions , quoteChar ) ; String conditionForThisPartitionColumn ; if ( hasMaxOffset ) { conditionForThisPartitionColumn = String . format ( AND_CONDITION_FORMAT , // add max condition first , since its param to set was already added
thisPartitionColumnMax , thisPartitionColumnMin ) ; } else { conditionForThisPartitionColumn = String . format ( CONDITION_FORMAT , thisPartitionColumnMin ) ; } // Add for preconditions ( EX : composite keys )
paramValueToSet . addAll ( new ArrayList < > ( preconditionParamVals ) ) ; Pair < Integer , String > paramValForCurrentOffsetColumn = Pair . of ( partitionSqlType , partitionOffset ) ; // Add for current partition column
paramValueToSet . add ( paramValForCurrentOffsetColumn ) ; finalOrConditions . add ( conditionForThisPartitionColumn ) ; preconditions . add ( getConditionForPartitionColumn ( partitionColumn , OffsetComparison . EQUALS , Collections . emptyList ( ) , quoteChar ) ) ; preconditionParamVals . add ( paramValForCurrentOffsetColumn ) ; } finalAndConditions . add ( String . format ( CONDITION_FORMAT , OR_JOINER . join ( finalOrConditions ) ) ) ; } if ( ! StringUtils . isEmpty ( tableContext . getExtraOffsetColumnConditions ( ) ) ) { // Apply extra offset column conditions configured which will be appended as AND on the query
String condition = tableJdbcELEvalContext . evaluateAsString ( "extraOffsetColumnConditions" , tableContext . getExtraOffsetColumnConditions ( ) ) ; finalAndConditions . add ( String . format ( CONDITION_FORMAT , condition ) ) ; } if ( ! finalAndConditions . isEmpty ( ) ) { queryBuilder . append ( String . format ( WHERE_CLAUSE , AND_JOINER . join ( finalAndConditions ) ) ) ; } Collection < String > quotedOffsetColumns = tableContext . getOffsetColumns ( ) . stream ( ) . map ( offsetCol -> String . format ( QUOTED_NAME , quoteChar , offsetCol , quoteChar ) ) . collect ( Collectors . toList ( ) ) ; queryBuilder . append ( String . format ( ORDER_BY_CLAUSE , COMMA_SPACE_JOINER . join ( quotedOffsetColumns ) ) ) ; return Pair . of ( queryBuilder . toString ( ) , paramValueToSet ) ;


signature:
public class OffsetQueryUtil { /** * Builds parts of the query in the where clause for the the partitition column .
 * @ param partitionColumn Partition Column
 * @ param comparison The comparison to use with respect to the partition column
 * @ param preconditions Any other precondition in the specific condition that needs to be combined with partition column .
 * ( For EX : if there are multiple order by we may need to say equals for columns
 * before this partition column and apply the current partition column conditions )
 * @ return the constructed condition for the partition column */
private static String getConditionForPartitionColumn ( String partitionColumn , OffsetComparison comparison , List < String > preconditions , String quoteChar ) { } }

implementation: String conditionTemplate = comparison . getQueryCondition ( ) ; List < String > finalConditions = new ArrayList < > ( preconditions ) ; finalConditions . add ( String . format ( conditionTemplate , String . format ( QUOTED_NAME , quoteChar , partitionColumn , quoteChar ) , PREPARED_STATEMENT_POSITIONAL_PARAMETER ) ) ; return AND_JOINER . join ( finalConditions ) ;


signature:
public class OffsetQueryUtil { /** * Splits the offset in the form of ( < column1 > = < value1 > : : < column2 > = < value2 > : : < column3 > = < value3 > ) into a map of columns and values
 * @ param lastOffset the last offset for the current table .
 * @ return Map of columns to values */
public static Map < String , String > getColumnsToOffsetMapFromOffsetFormat ( String lastOffset ) { } }

implementation: Map < String , String > offsetColumnsToOffsetMap = new HashMap < > ( ) ; if ( StringUtils . isNotBlank ( lastOffset ) ) { Iterator < String > offsetColumnsAndOffsetIterator = OFFSET_COLUMN_SPLITTER . split ( lastOffset ) . iterator ( ) ; while ( offsetColumnsAndOffsetIterator . hasNext ( ) ) { String offsetColumnAndOffset = offsetColumnsAndOffsetIterator . next ( ) ; String [ ] offsetColumnOffsetSplit = offsetColumnAndOffset . split ( "=" ) ; String offsetColumn = offsetColumnOffsetSplit [ 0 ] ; String offset = offsetColumnOffsetSplit [ 1 ] ; offsetColumnsToOffsetMap . put ( offsetColumn , offset ) ; } } return offsetColumnsToOffsetMap ;


signature:
public class OffsetQueryUtil { /** * Joins the map of column to values to a string offset in the form of ( < column1 > = < value1 > : : < column2 > = < value2 > : : < column3 > = < value3 > )
 * @ param tableContext Context for the current table .
 * @ param fields The current record fields
 * @ return Offset in the form of ( < column1 > = < value1 > : : < column2 > = < value2 > : : < column3 > = < value3 > ) */
public static String getOffsetFormatFromColumns ( TableRuntimeContext tableContext , Map < String , Field > fields ) throws StageException { } }

implementation: return getOffsetFormat ( getOffsetsFromColumns ( tableContext , fields ) ) ;


signature:
public class OffsetQueryUtil { /** * This is the inverse of { @ link # getSourceKeyOffsetsRepresentation ( Map ) } .
 * @ param offsets the String representation of source key offsets
 * @ return the { @ link Map } of the offsets reconstructed from the supplied representation */
public static Map < String , String > getOffsetsFromSourceKeyRepresentation ( String offsets ) { } }

implementation: final Map < String , String > offsetMap = new HashMap < > ( ) ; if ( StringUtils . isNotBlank ( offsets ) ) { for ( String col : StringUtils . splitByWholeSeparator ( offsets , OFFSET_KEY_COLUMN_SEPARATOR ) ) { final String [ ] parts = StringUtils . splitByWholeSeparator ( col , OFFSET_KEY_COLUMN_NAME_VALUE_SEPARATOR , 2 ) ; if ( parts . length != 2 ) { throw new IllegalArgumentException ( String . format ( "Invalid column offset of \"%s\" seen.  Expected colName%svalue.  Full offsets representation: %s" , col , OFFSET_KEY_COLUMN_NAME_VALUE_SEPARATOR , offsets ) ) ; } offsetMap . put ( parts [ 0 ] , parts [ 1 ] ) ; } } return offsetMap ;


signature:
public class OffsetQueryUtil { /** * Validates whether offset names match in the stored offset with respect to table configuration
 * @ param tableContext { @ link TableContext } for table
 * @ param offset Stored offset from the previous stage run
 * @ return the actual offset map as deserialized from the offset param , if valid
 * @ throws StageException if columns mismatch */
public static Map < String , String > validateStoredAndSpecifiedOffset ( TableContext tableContext , String offset ) throws StageException { } }

implementation:
Set < String > expectedColumns = Sets . newHashSet ( tableContext . getOffsetColumns ( ) ) ; final Map < String , String > actualOffsets = getColumnsToOffsetMapFromOffsetFormat ( offset ) ; // only perform the actual validation below if there ARE stored offsets
if ( actualOffsets . size ( ) == 0 ) { return actualOffsets ; } Set < String > actualColumns = actualOffsets . keySet ( ) ; Set < String > expectedSetDifference = Sets . difference ( expectedColumns , actualColumns ) ; Set < String > actualSetDifference = Sets . difference ( actualColumns , expectedColumns ) ; if ( expectedSetDifference . size ( ) > 0 || actualSetDifference . size ( ) > 0 ) { throw new StageException ( JdbcErrors . JDBC_71 , tableContext . getQualifiedName ( ) , COMMA_SPACE_JOINER . join ( actualColumns ) , COMMA_SPACE_JOINER . join ( expectedColumns ) ) ; } return actualOffsets ;


signature:
public class SupportBundleManager { /** * Return InputStream from which a new generated resource bundle can be retrieved . */
public SupportBundle generateNewBundle ( List < String > generators , BundleType bundleType ) throws IOException { } }

implementation: List < BundleContentGeneratorDefinition > defs = getRequestedDefinitions ( generators ) ; return generateNewBundleFromInstances ( defs . stream ( ) . map ( BundleContentGeneratorDefinition :: createInstance ) . collect ( Collectors . toList ( ) ) , bundleType ) ;


signature:
public class SupportBundleManager { /** * Return InputStream from which a new generated resource bundle can be retrieved . */
public SupportBundle generateNewBundleFromInstances ( List < BundleContentGenerator > generators , BundleType bundleType ) throws IOException { } }

implementation: PipedInputStream inputStream = new PipedInputStream ( ) ; PipedOutputStream outputStream = new PipedOutputStream ( ) ; inputStream . connect ( outputStream ) ; ZipOutputStream zipOutputStream = new ZipOutputStream ( outputStream ) ; executor . submit ( ( ) -> generateNewBundleInternal ( generators , bundleType , zipOutputStream ) ) ; String bundleName = generateBundleName ( bundleType ) ; String bundleKey = generateBundleDate ( bundleType ) + "/" + bundleName ; return new SupportBundle ( bundleKey , bundleName , inputStream ) ;


signature:
public class SupportBundleManager { /** * Instead of providing support bundle directly to user , upload it to StreamSets backend services . */
public void uploadNewBundle ( List < String > generators , BundleType bundleType ) throws IOException { } }

implementation: List < BundleContentGeneratorDefinition > defs = getRequestedDefinitions ( generators ) ; uploadNewBundleFromInstances ( defs . stream ( ) . map ( BundleContentGeneratorDefinition :: createInstance ) . collect ( Collectors . toList ( ) ) , bundleType ) ;


signature:
public class SupportBundleManager { /** * Instead of providing support bundle directly to user , upload it to StreamSets backend services . */
public void uploadNewBundleFromInstances ( List < BundleContentGenerator > generators , BundleType bundleType ) throws IOException { } }

implementation:
// Generate bundle
SupportBundle bundle = generateNewBundleFromInstances ( generators , bundleType ) ; boolean enabled = configuration . get ( Constants . UPLOAD_ENABLED , Constants . DEFAULT_UPLOAD_ENABLED ) ; String accessKey = configuration . get ( Constants . UPLOAD_ACCESS , Constants . DEFAULT_UPLOAD_ACCESS ) ; String secretKey = configuration . get ( Constants . UPLOAD_SECRET , Constants . DEFAULT_UPLOAD_SECRET ) ; String bucket = configuration . get ( Constants . UPLOAD_BUCKET , Constants . DEFAULT_UPLOAD_BUCKET ) ; int bufferSize = configuration . get ( Constants . UPLOAD_BUFFER_SIZE , Constants . DEFAULT_UPLOAD_BUFFER_SIZE ) ; if ( ! enabled ) { throw new IOException ( "Uploading support bundles was disabled by administrator." ) ; } AWSCredentialsProvider credentialsProvider = new StaticCredentialsProvider ( new BasicAWSCredentials ( accessKey , secretKey ) ) ; AmazonS3Client s3Client = new AmazonS3Client ( credentialsProvider , new ClientConfiguration ( ) ) ; s3Client . setS3ClientOptions ( new S3ClientOptions ( ) . withPathStyleAccess ( true ) ) ; s3Client . setRegion ( Region . getRegion ( Regions . US_WEST_2 ) ) ; // Object Metadata
ObjectMetadata s3Metadata = new ObjectMetadata ( ) ; for ( Map . Entry < Object , Object > entry : getMetadata ( bundleType ) . entrySet ( ) ) { s3Metadata . addUserMetadata ( ( String ) entry . getKey ( ) , ( String ) entry . getValue ( ) ) ; } List < PartETag > partETags ; InitiateMultipartUploadResult initResponse = null ; try { // Uploading part by part
LOG . info ( "Initiating multi-part support bundle upload" ) ; partETags = new ArrayList < > ( ) ; InitiateMultipartUploadRequest initRequest = new InitiateMultipartUploadRequest ( bucket , bundle . getBundleKey ( ) ) ; initRequest . setObjectMetadata ( s3Metadata ) ; initResponse = s3Client . initiateMultipartUpload ( initRequest ) ; } catch ( AmazonClientException e ) { LOG . error ( "Support bundle upload failed: " , e ) ; throw new IOException ( "Support bundle upload failed" , e ) ; } try { byte [ ] buffer = new byte [ bufferSize ] ; int partId = 1 ; int size = - 1 ; while ( ( size = readFully ( bundle . getInputStream ( ) , buffer ) ) != - 1 ) { LOG . debug ( "Uploading part {} of size {}" , partId , size ) ; UploadPartRequest uploadRequest = new UploadPartRequest ( ) . withBucketName ( bucket ) . withKey ( bundle . getBundleKey ( ) ) . withUploadId ( initResponse . getUploadId ( ) ) . withPartNumber ( partId ++ ) . withInputStream ( new ByteArrayInputStream ( buffer ) ) . withPartSize ( size ) ; partETags . add ( s3Client . uploadPart ( uploadRequest ) . getPartETag ( ) ) ; } CompleteMultipartUploadRequest compRequest = new CompleteMultipartUploadRequest ( bucket , bundle . getBundleKey ( ) , initResponse . getUploadId ( ) , partETags ) ; s3Client . completeMultipartUpload ( compRequest ) ; LOG . info ( "Support bundle upload finished" ) ; } catch ( Exception e ) { LOG . error ( "Support bundle upload failed" , e ) ; s3Client . abortMultipartUpload ( new AbortMultipartUploadRequest ( bucket , bundle . getBundleKey ( ) , initResponse . getUploadId ( ) ) ) ; throw new IOException ( "Can't upload support bundle" , e ) ; } finally { // Close the client
s3Client . shutdown ( ) ; }


signature:
public class SupportBundleManager { /** * Try to upload bundle as part of internal SDC error ( for example failing pipeline ) . */
public void uploadNewBundleOnError ( ) { } }

implementation:
boolean enabled = configuration . get ( Constants . UPLOAD_ON_ERROR , Constants . DEFAULT_UPLOAD_ON_ERROR ) ; LOG . info ( "Upload bundle on error: {}" , enabled ) ; // We won ' t upload the bundle unless it ' s explicitly allowed
if ( ! enabled ) { return ; } try { uploadNewBundle ( Collections . emptyList ( ) , BundleType . SUPPORT ) ; } catch ( IOException e ) { LOG . error ( "Failed to upload error bundle" , e ) ; }


signature:
public class SupportBundleManager { /** * This method will read from the input stream until the whole buffer is loaded up with actual bytes or end of stream
 * has been reached . Hence it will return buffer . length of all executions except the last two - one to the last call
 * will return less then buffer . length ( reminder of the data ) and returns - 1 on any subsequent calls . */
private int readFully ( InputStream inputStream , byte [ ] buffer ) throws IOException { } }

implementation: int readBytes = 0 ; while ( readBytes < buffer . length ) { int loaded = inputStream . read ( buffer , readBytes , buffer . length - readBytes ) ; if ( loaded == - 1 ) { return readBytes == 0 ? - 1 : readBytes ; } readBytes += loaded ; } return readBytes ;


signature:
public class SupportBundleManager { /** * Orchestrate what definitions should be used for this bundle .
 * Either get all definitions that should be used by default or only those specified in the generators argument . */
private List < BundleContentGeneratorDefinition > getRequestedDefinitions ( List < String > generators ) { } }

implementation:
Stream < BundleContentGeneratorDefinition > stream = definitions . stream ( ) ; if ( generators == null || generators . isEmpty ( ) ) { // Filter out default generators
stream = stream . filter ( BundleContentGeneratorDefinition :: isEnabledByDefault ) ; } else { stream = stream . filter ( def -> generators . contains ( def . getId ( ) ) ) ; } return stream . sorted ( Comparator . comparingInt ( BundleContentGeneratorDefinition :: getOrder ) ) . collect ( Collectors . toList ( ) ) ;


signature:
public class HttpReceiverServlet { /** * From https : / / stackoverflow . com / a / 31928740/33905 */
@ VisibleForTesting protected static Map < String , String [ ] > getQueryParameters ( HttpServletRequest request ) { } }

implementation:
Map < String , String [ ] > queryParameters = new HashMap < > ( ) ; String queryString = request . getQueryString ( ) ; if ( StringUtils . isEmpty ( queryString ) ) { return queryParameters ; } String [ ] parameters = queryString . split ( "&" ) ; for ( String parameter : parameters ) { String [ ] keyValuePair = parameter . split ( "=" ) ; String [ ] values = queryParameters . get ( keyValuePair [ 0 ] ) ; values = ArrayUtils . add ( values , keyValuePair . length == 1 ? "" : keyValuePair [ 1 ] ) ; // length is one if no value is available .
queryParameters . put ( keyValuePair [ 0 ] , values ) ; } return queryParameters ;


signature:
public class MySQLBinLogFieldConverter { /** * This is used by destination which performs CRUD operations .
 * Records sent from MySQLBinLog origin have / Data field that contains a map of
 * key - value for INSERT and UPDATE operation , but
 * / OldData field for DELETE operation .
 * This method looks into / OldData for DELETE operation so that users don ' t
 * need a separate field - column mapping for DELETE .
 * @ param fieldPath fieldPath specified in field - column mapping
 * @ param operation Crud operation code
 * @ return Actual field path in record */
@ Override public String getFieldPath ( String fieldPath , int operation ) { } }

implementation: if ( operation == OperationType . DELETE_CODE ) { return fieldPath . replace ( DATA_FIELD , OLDDATA_FIELD ) ; } return fieldPath ;


signature:
public class YarnAppLauncher { /** * If there is a RecordEL , then an arg could eval to empty string . This method returns
 * args that are not null or empty . */
private String [ ] getNonEmptyArgs ( List < String > appArgs ) { } }

implementation: List < String > nonEmpty = new ArrayList < > ( ) ; appArgs . forEach ( ( String val ) -> { if ( ! StringUtils . isEmpty ( val ) ) { nonEmpty . add ( val ) ; } } ) ; return nonEmpty . toArray ( new String [ nonEmpty . size ( ) ] ) ;


signature:
public class KuduTarget { /** * Return Operation based on the operation code . If the code has a number
 * that Kudu destination doesn ' t support , it throws UnsupportedOperationException .
 * @ param table
 * @ param op
 * @ return
 * @ throws UnsupportedOperationException */
protected Operation getOperation ( KuduTable table , int op ) throws UnsupportedOperationException { } }

implementation: Operation operation = null ; switch ( op ) { case OperationType . INSERT_CODE : operation = table . newInsert ( ) ; break ; case OperationType . UPSERT_CODE : operation = table . newUpsert ( ) ; break ; case OperationType . UPDATE_CODE : operation = table . newUpdate ( ) ; break ; case OperationType . DELETE_CODE : operation = table . newDelete ( ) ; break ; default : LOG . error ( "Operation {} not supported" , op ) ; throw new UnsupportedOperationException ( String . format ( "Unsupported Operation: %s" , op ) ) ; } return operation ;


signature:
public class MetricEnabledWrapperStream { /** * Convert the bytes to a human readable format upto 2 decimal places
 * The maximum unit is TB , so anything exceeding 1024 TB will be shown
 * with TB unit .
 * @ param bytes the number of bytes .
 * @ return human readable format of bytes in units . */
static String convertBytesToDisplayFormat ( double bytes ) { } }

implementation: int unitIdx = 0 ; double unitChangedBytes = bytes ; while ( unitIdx < UNITS . length - 1 && Math . floor ( unitChangedBytes / 1024 ) > 0 ) { unitChangedBytes = unitChangedBytes / 1024 ; unitIdx ++ ; } return df . format ( unitChangedBytes ) + " " + UNITS [ unitIdx ] ;


signature:
public class LiveDirectoryScanner { /** * Scans the directory of for the next file .
 * @ param current the last ' rolled ' file processed . Use < code > null < / code > to look for the first one . The provided
 * file cannot be the ' live ' file .
 * @ return the next ' rolled ' file in order , if there are no more ' rolled ' files it will return the ' live ' file ,
 * if there is not ' live ' file it will return < code > null < / code > .
 * @ throws IOException thrown if the directory could not be scanned . */
public LiveFile scan ( LiveFile current ) throws IOException { } }

implementation:
try { return scanInternal ( current ) ; } catch ( NoSuchFileException ex ) { // this could happen because there has been a file rotation / deletion after the search / filter / sort and before
// the creation of the nen current . Lets sleep for 50ms and try again , if fails again give up .
ThreadUtil . sleep ( 50 ) ; return scanInternal ( current ) ; }


signature:
public class LiveDirectoryScanner { /** * Scans the directory for number of files yet to be processed .
 * @ param current the last ' rolled ' file processed . Use < code > null < / code > to look for the first one .
 * @ return the number of files yet to be processed .
 * @ throws IOException thrown if the directory could not be scanned . */
public long getPendingFiles ( LiveFile current ) throws IOException { } }

implementation:
// Current will not be acceptable for roll files ( if active file is without a counter / date pattern )
// and will be later renamed to a file with counter / date suffix , if that is the case we should
// return 0 as number of pending files
if ( current == null || rollMode . isCurrentAcceptable ( current . getPath ( ) . getFileName ( ) . toString ( ) ) ) { return findToBeProcessedMatchingFiles ( current != null ? current . refresh ( ) : null ) . size ( ) ; } return 0 ;


signature:
public class CollisionWhitelist { /** * Return true if this dependency and given set of versions is whitelisted .
 * This class have several rules for whitelisting - some of them are harcoded ( known whitelist for all libraries ) ,
 * whereas the optional Properties argument allows specific exceptions for this particular classpath .
 * @ param name Name of the dependency
 * @ param specificWhitelist Properties file with exceptions
 * @ param dependencies Version - > List of full dependencies
 * @ return */
public static boolean isWhitelisted ( String name , Properties specificWhitelist , Map < String , List < Dependency > > dependencies ) { } }

implementation:
if ( specificWhitelist != null && specificWhitelist . containsKey ( name ) ) { return versionsMatch ( specificWhitelist . getProperty ( name ) , dependencies . keySet ( ) ) ; } // Otherwise try hardcoded rules :
WhitelistRule rule = WHITELIST_RULES . get ( name ) ; return rule != null && rule . isWhitelisted ( dependencies ) ;


signature:
public class CollisionWhitelist { /** * Compare expected versions with given versions to see if they are the same or not .
 * @ param expectedVersions Versions that are expected ( and thus whitelisted )
 * @ param versions Versions that were detected on the classpath .
 * @ return True if and only if those two " sets " equals */
private static boolean versionsMatch ( String expectedVersions , Set < String > versions ) { } }

implementation: Set < String > expectedSet = Sets . newHashSet ( expectedVersions . split ( "," ) ) ; return Sets . symmetricDifference ( expectedSet , versions ) . isEmpty ( ) ;


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

implementation: if ( ! trustStorePath . isEmpty ( ) ) { if ( Files . notExists ( Paths . get ( trustStorePath ) ) ) { issues . add ( context . createConfigIssue ( groupName , prefix + "trustStorePath" , HTTP_04 , trustStorePath ) ) ; } try { if ( trustStorePassword . get ( ) . isEmpty ( ) ) { issues . add ( context . createConfigIssue ( groupName , prefix + "trustStorePassword" , HTTP_05 ) ) ; } } catch ( StageException e ) { issues . add ( context . createConfigIssue ( groupName , prefix + "trustStorePassword" , HTTP_29 , e . toString ( ) ) ) ; } } if ( ! keyStorePath . isEmpty ( ) ) { if ( Files . notExists ( Paths . get ( keyStorePath ) ) ) { issues . add ( context . createConfigIssue ( groupName , prefix + "keyStorePath" , HTTP_04 , keyStorePath ) ) ; } try { if ( keyStorePassword . get ( ) . isEmpty ( ) ) { issues . add ( context . createConfigIssue ( groupName , prefix + "keyStorePassword" , HTTP_05 ) ) ; } } catch ( StageException e ) { issues . add ( context . createConfigIssue ( groupName , prefix + "keyStorePassword" , HTTP_29 , e . toString ( ) ) ) ; } }


signature:
public class ValidationUtil { /** * Returns true if given snapshot output is usable - e . g . if it make sense to persist . */
public static boolean isSnapshotOutputUsable ( List < StageOutput > stagesOutput ) { } }

implementation:
// In case that the snapshot actually does not exists
if ( stagesOutput == null ) { return false ; } // We ' re looking for at least one output lane that is not empty . In most cases the first stage in the list will
// be origin that generated some data and hence the loop will terminate fast . In the worst case scenario we will
// iterate over all stages in attempt to find at least one record in the snapshot .
for ( StageOutput output : stagesOutput ) { if ( CollectionUtils . isNotEmpty ( output . getErrorRecords ( ) ) || CollectionUtils . isNotEmpty ( output . getEventRecords ( ) ) || CollectionUtils . isNotEmpty ( output . getStageErrors ( ) ) ) { return true ; } for ( Map . Entry < String , List < Record > > entry : output . getOutput ( ) . entrySet ( ) ) { if ( ! entry . getValue ( ) . isEmpty ( ) ) { return true ; } } } return false ;


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

implementation:
SparkStreamingBinding binding = null ; try { binding = SparkStreamingBindingFactory . build ( BootstrapCluster . getProperties ( ) ) ; binding . init ( ) ; BootstrapCluster . createTransformers ( binding . getStreamingContext ( ) . sparkContext ( ) , binding . getSparkSession ( ) ) ; binding . startContext ( ) ; binding . awaitTermination ( ) ; } catch ( Throwable error ) { String msg = "Error trying to invoke BootstrapClusterStreaming.main: " + error ; System . err . println ( new Date ( ) + ": " + msg ) ; error . printStackTrace ( System . err ) ; // required as in local mode the following seems to be lost
LOG . error ( msg , error ) ; throw new IllegalStateException ( msg , error ) ; } finally { try { if ( binding != null ) { binding . close ( ) ; } } catch ( Exception ex ) { LOG . warn ( "Error on binding close: " + ex , ex ) ; } }


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

implementation:
// Validate configuration values and open any required resources .
List < ConfigIssue > issues = super . init ( ) ; Optional . ofNullable ( conf . init ( getContext ( ) , CONF_PREFIX ) ) . ifPresent ( issues :: addAll ) ; try { ConnectorConfig partnerConfig = ForceUtils . getPartnerConfig ( conf , new WaveSessionRenewer ( ) ) ; connection = Connector . newConnection ( partnerConfig ) ; LOG . info ( "Successfully authenticated as {}" , conf . username ) ; if ( conf . mutualAuth . useMutualAuth ) { ForceUtils . setupMutualAuth ( partnerConfig , conf . mutualAuth ) ; } String soapEndpoint = connection . getConfig ( ) . getServiceEndpoint ( ) ; restEndpoint = soapEndpoint . substring ( 0 , soapEndpoint . indexOf ( "services/Soap/" ) ) ; httpClient = new HttpClient ( ForceUtils . makeSslContextFactory ( conf ) ) ; if ( conf . useProxy ) { ForceUtils . setProxy ( httpClient , conf ) ; } httpClient . start ( ) ; } catch ( Exception e ) { LOG . error ( "Exception during init()" , e ) ; issues . add ( getContext ( ) . createConfigIssue ( Groups . FORCE . name ( ) , ForceConfigBean . CONF_PREFIX + "authEndpoint" , Errors . WAVE_00 , ForceUtils . getExceptionCode ( e ) + ", " + ForceUtils . getExceptionMessage ( e ) ) ) ; } // If issues is not empty , the UI will inform the user of each configuration issue in the list .
return issues ;


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

implementation: if ( batch . getRecords ( ) . hasNext ( ) ) { if ( datasetID == null ) { LOG . info ( "Opening dataset" ) ; try { openDataset ( ) ; } catch ( ConnectionException ce ) { throw new StageException ( Errors . WAVE_01 , ce ) ; } } LOG . info ( "Writing batch to dataset" ) ; writeToDataset ( batch ) ; try { commitDataset ( ) ; } catch ( ConnectionException | IOException e ) { throw new StageException ( Errors . WAVE_01 , e ) ; } }


signature:
public class PipelineConfigurationValidator { /** * We have special type of a ConfigDef called RUNTIME . This config is never displayed in UI and instead it ' s values
 * are supplied at " runtime " . This method is the " runtime " method that propagates them . */
private void propagateRuntimeConfiguration ( ) { } }

implementation:
// If pipeline wasn ' t loaded or there if there are no stages , there is nothing to propagate
if ( pipelineBean == null || pipelineBean . getPipelineStageBeans ( ) == null ) { return ; } for ( StageBean stageBean : pipelineBean . getPipelineStageBeans ( ) . getStages ( ) ) { for ( ServiceDependencyDefinition serviceDependency : stageBean . getDefinition ( ) . getServices ( ) ) { ServiceBean stageService = stageBean . getService ( serviceDependency . getService ( ) ) ; if ( stageService == null ) { continue ; } ServiceConfiguration serviceConfiguration = stageService . getConf ( ) ; List < Config > configs = serviceConfiguration . getConfiguration ( ) ; // Simply remove all RUNTIME configs
configs . removeAll ( serviceDependency . getConfiguration ( ) . keySet ( ) . stream ( ) . map ( serviceConfiguration :: getConfig ) . collect ( Collectors . toList ( ) ) ) ; // And insert them with the stage - instance - constant values
serviceDependency . getConfiguration ( ) . forEach ( ( key , value ) -> configs . add ( new Config ( key , value ) ) ) ; // And overwrite the new state
serviceConfiguration . setConfig ( configs ) ; } }


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

implementation:
// Validate configuration values and open any required resources .
List < ConfigIssue > issues = super . init ( ) ; if ( issues . isEmpty ( ) ) { jdbcUtil = UtilsProvider . getJdbcUtil ( ) ; } errorRecordHandler = new DefaultErrorRecordHandler ( getContext ( ) ) ; Processor . Context context = getContext ( ) ; queryEval = getContext ( ) . createELEval ( "query" ) ; issues = hikariConfigBean . validateConfigs ( context , issues ) ; if ( context . getRunnerId ( ) == 0 ) { if ( issues . isEmpty ( ) && null == dataSource ) { try { dataSource = jdbcUtil . createDataSourceForRead ( hikariConfigBean ) ; context . getStageRunnerSharedMap ( ) . put ( "jdbcLookupProcessor.dataSource" , dataSource ) ; } catch ( StageException e ) { issues . add ( context . createConfigIssue ( Groups . JDBC . name ( ) , CONNECTION_STRING , JdbcErrors . JDBC_00 , e . toString ( ) ) ) ; } } } else { dataSource = ( HikariDataSource ) context . getStageRunnerSharedMap ( ) . get ( "jdbcLookupProcessor.dataSource" ) ; } if ( issues . isEmpty ( ) ) { this . defaultValue = calculateDefault ( context , issues ) ; } if ( issues . isEmpty ( ) ) { cache = buildCache ( ) ; cacheCleaner = new CacheCleaner ( cache , "JdbcLookupProcessor" , 10 * 60 * 1000 ) ; if ( cacheConfig . enabled ) { preprocessThreads = Math . min ( hikariConfigBean . minIdle , Runtime . getRuntime ( ) . availableProcessors ( ) - 1 ) ; preprocessThreads = Math . max ( preprocessThreads , 1 ) ; } } if ( context . getRunnerId ( ) == 0 ) { if ( issues . isEmpty ( ) && generationExecutor == null ) { generationExecutor = new SafeScheduledExecutorService ( hikariConfigBean . maximumPoolSize , "JDBC Lookup Cache Warmer" ) ; context . getStageRunnerSharedMap ( ) . put ( "jdbcLookupProcessor.generationExecutor" , generationExecutor ) ; } } else { generationExecutor = ( SafeScheduledExecutorService ) context . getStageRunnerSharedMap ( ) . get ( "jdbcLookupProcessor.generationExecutor" ) ; } // If issues is not empty , the UI will inform the user of each configuration issue in the list .
return issues ;


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

implementation:
if ( getContext ( ) . getRunnerId ( ) == 0 ) { if ( generationExecutor != null ) { generationExecutor . shutdown ( ) ; try { if ( ! generationExecutor . awaitTermination ( 5 , TimeUnit . SECONDS ) ) { generationExecutor . shutdownNow ( ) ; } } catch ( InterruptedException ex ) { LOG . error ( "Interrupted while attempting to shutdown Generator Executor: " , ex ) ; Thread . currentThread ( ) . interrupt ( ) ; } } // close dataSource after closing threadpool executor as we could have queries running before closing the executor
if ( jdbcUtil != null ) { jdbcUtil . closeQuietly ( dataSource ) ; } } super . destroy ( ) ;


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

implementation:
try { ELVars elVars = getContext ( ) . createELVars ( ) ; RecordEL . setRecordInContext ( elVars , record ) ; String preparedQuery = queryEval . eval ( elVars , query , String . class ) ; Optional < List < Map < String , Field > > > entry = cache . get ( preparedQuery ) ; if ( ! entry . isPresent ( ) ) { // No results
switch ( missingValuesBehavior ) { case SEND_TO_ERROR : LOG . error ( JdbcErrors . JDBC_04 . getMessage ( ) , preparedQuery ) ; errorRecordHandler . onError ( new OnRecordErrorException ( record , JdbcErrors . JDBC_04 , preparedQuery ) ) ; break ; case PASS_RECORD_ON : batchMaker . addRecord ( record ) ; break ; default : throw new IllegalStateException ( "Unknown missing value behavior: " + missingValuesBehavior ) ; } } else { List < Map < String , Field > > values = entry . get ( ) ; switch ( multipleValuesBehavior ) { case FIRST_ONLY : setFieldsInRecord ( record , values . get ( 0 ) ) ; batchMaker . addRecord ( record ) ; break ; case SPLIT_INTO_MULTIPLE_RECORDS : for ( Map < String , Field > lookupItem : values ) { Record newRecord = getContext ( ) . cloneRecord ( record ) ; setFieldsInRecord ( newRecord , lookupItem ) ; batchMaker . addRecord ( newRecord ) ; } break ; case ALL_AS_LIST : Map < String , List < Field > > valuesMap = new HashMap < > ( ) ; for ( Map < String , Field > lookupItem : values ) { lookupItem . forEach ( ( k , v ) -> { if ( valuesMap . get ( k ) == null ) { List < Field > lookupValue = new ArrayList < > ( ) ; valuesMap . put ( k , lookupValue ) ; } valuesMap . get ( k ) . add ( v ) ; } ) ; } Map < String , Field > valueMap = new HashMap < > ( ) ; valuesMap . forEach ( ( k , v ) -> valueMap . put ( k , Field . create ( v ) ) ) ; setFieldsInRecord ( record , valueMap ) ; batchMaker . addRecord ( record ) ; break ; default : throw new IllegalStateException ( "Unknown multiple value behavior: " + multipleValuesBehavior ) ; } } } catch ( ELEvalException e ) { LOG . error ( JdbcErrors . JDBC_01 . getMessage ( ) , query , e ) ; throw new OnRecordErrorException ( record , JdbcErrors . JDBC_01 , query ) ; } catch ( ExecutionException e ) { Throwables . propagateIfPossible ( e . getCause ( ) , StageException . class ) ; throw new IllegalStateException ( e ) ; // The cache loader shouldn ' t throw anything that isn ' t a StageException .
} catch ( OnRecordErrorException error ) { // NOSONAR
errorRecordHandler . onError ( new OnRecordErrorException ( record , error . getErrorCode ( ) , error . getParams ( ) ) ) ; }


signature:
public class OmnitureSource { /** * Validate Ominture Report Description . */
private void validateReportDescription ( List < ConfigIssue > issues ) { } }

implementation: if ( ! jsonMapper . isValidJson ( this . reportDescription ) ) { issues . add ( getContext ( ) . createConfigIssue ( Groups . REPORT . name ( ) , "reportDescription" , Errors . OMNITURE_03 ) ) ; }


signature:
public class RBGenMojo { /** * Escapes special characters with a preceding slash */
private String escapeValue ( String str , boolean escapeSpace ) { } }

implementation:
int len = str . length ( ) ; int bufLen = len * 2 ; if ( bufLen < 0 ) { bufLen = Integer . MAX_VALUE ; } StringBuilder outBuffer = new StringBuilder ( bufLen ) ; for ( int x = 0 ; x < len ; x ++ ) { char aChar = str . charAt ( x ) ; // Handle common case first , selecting largest block that
// avoids the specials below
if ( ( aChar > 61 ) && ( aChar < 127 ) ) { if ( aChar == '\\' ) { outBuffer . append ( '\\' ) ; outBuffer . append ( '\\' ) ; continue ; } outBuffer . append ( aChar ) ; continue ; } switch ( aChar ) { case ' ' : if ( x == 0 || escapeSpace ) outBuffer . append ( '\\' ) ; outBuffer . append ( ' ' ) ; break ; case '\t' : outBuffer . append ( '\\' ) ; outBuffer . append ( 't' ) ; break ; case '\n' : outBuffer . append ( '\\' ) ; outBuffer . append ( 'n' ) ; break ; case '\r' : outBuffer . append ( '\\' ) ; outBuffer . append ( 'r' ) ; break ; case '\f' : outBuffer . append ( '\\' ) ; outBuffer . append ( 'f' ) ; break ; case '=' : // Fall through
case ':' : // Fall through
case '#' : // Fall through
case '!' : outBuffer . append ( '\\' ) ; outBuffer . append ( aChar ) ; break ; default : outBuffer . append ( aChar ) ; } } return outBuffer . toString ( ) ;


signature:
public class JdbcRecordReaderWriterFactory { /** * Called by JdbcTarget */
public static JdbcRecordWriter createJdbcRecordWriter ( String connectionString , HikariDataSource dataSource , String schema , String tableName , List < JdbcFieldColumnParamMapping > customMappings , boolean rollbackOnError , boolean useMultiRowOp , int maxPrepStmtParameters , int defaultOpCode , UnsupportedOperationAction unsupportedAction , DuplicateKeyAction duplicateKeyAction , JdbcRecordReader recordReader , boolean caseSensitive , List < String > customDataSqlStateCodes ) throws StageException { } }

implementation: if ( defaultOpCode == OperationType . LOAD_CODE ) { return new JdbcLoadRecordWriter ( connectionString , dataSource , schema , tableName , customMappings , duplicateKeyAction , recordReader , caseSensitive , customDataSqlStateCodes ) ; } else { return createJdbcRecordWriter ( connectionString , dataSource , schema , tableName , customMappings , null , rollbackOnError , useMultiRowOp , maxPrepStmtParameters , defaultOpCode , unsupportedAction , recordReader , caseSensitive , customDataSqlStateCodes ) ; }


signature:
public class JdbcRecordReaderWriterFactory { /** * Called by JdbcTeeProcessor */
public static JdbcRecordWriter createJdbcRecordWriter ( String connectionString , HikariDataSource dataSource , String schema , String tableName , List < JdbcFieldColumnParamMapping > customMappings , List < JdbcFieldColumnMapping > generatedColumnMappings , boolean rollbackOnError , boolean useMultiRowOp , int maxPrepStmtParameters , int defaultOpCode , UnsupportedOperationAction unsupportedAction , JdbcRecordReader recordReader , boolean caseSensitive , List < String > customDataSqlStateCodes ) throws StageException { } }

implementation: JdbcRecordWriter recordWriter ; if ( useMultiRowOp ) { recordWriter = new JdbcMultiRowRecordWriter ( connectionString , dataSource , schema , tableName , rollbackOnError , customMappings , maxPrepStmtParameters , defaultOpCode , unsupportedAction , generatedColumnMappings , recordReader , caseSensitive , customDataSqlStateCodes ) ; } else { recordWriter = new JdbcGenericRecordWriter ( connectionString , dataSource , schema , tableName , rollbackOnError , customMappings , defaultOpCode , unsupportedAction , generatedColumnMappings , recordReader , caseSensitive , customDataSqlStateCodes ) ; } return recordWriter ;


signature:
public class BigQueryTargetUpgrader { /** * Remove implicit field mapping */
private void upgradeV1toV2 ( List < Config > configs ) { } }

implementation: configs . removeIf ( config -> ( config . getName ( ) . equals ( IMPLICIT_FIELD_MAPPING_CONFIG ) || config . getName ( ) . equals ( BIG_QUERY_IMPLICIT_FIELD_MAPPING_CONFIG ) ) ) ; configs . add ( new Config ( MAX_CACHE_SIZE , - 1 ) ) ;


signature:
public class Grok { /** * Extracts named groups from the raw data
 * @ param rawData
 * @ return A map of group names mapped to their extracted values or null if there are no matches */
public Map < String , String > extractNamedGroups ( final CharSequence rawData ) { } }

implementation: Matcher matcher = compiledPattern . matcher ( rawData ) ; if ( matcher . find ( ) ) { MatchResult r = matcher . toMatchResult ( ) ; if ( r != null && r . namedGroups ( ) != null ) { return r . namedGroups ( ) ; } } return null ;


signature:
public class ControlChannel { /** * If a null value is passed to this method , it ' s replaced with
 * a dummy due to the fact the payload for each message is wrapped
 * in an Optional . */
public void consumerCommit ( String offset ) { } }

implementation: Object offsetValue = offset ; if ( offsetValue == null ) { offsetValue = new NullOffset ( ) ; } LOG . trace ( "Commit Offset: '{}'" , offsetValue ) ; try { producerQueue . put ( new Message ( MessageType . CONSUMER_COMMIT , offsetValue ) ) ; } catch ( InterruptedException e ) { LOG . info ( "Interrupted while queuing '{}'" , MessageType . CONSUMER_COMMIT . name ( ) , offsetValue ) ; Thread . currentThread ( ) . interrupt ( ) ; }


signature:
public class EmbeddedSDCPool { /** * Creates an instance of SDC and adds to pool
 * @ return EmbeddedSDC
 * @ throws Exception */
@ SuppressWarnings ( "unchecked" ) protected EmbeddedSDC create ( ) throws Exception { } }

implementation:
Utils . checkState ( open , "Not open" ) ; final EmbeddedSDC embeddedSDC = new EmbeddedSDC ( ) ; Object source ; // post - batch runnable
Object pipelineStartResult = BootstrapCluster . startPipeline ( ( ) -> LOG . debug ( "Batch completed" ) ) ; source = pipelineStartResult . getClass ( ) . getDeclaredField ( "source" ) . get ( pipelineStartResult ) ; if ( source instanceof DSource ) { long startTime = System . currentTimeMillis ( ) ; long endTime = startTime ; long diff = 0 ; Source actualSource = ( ( DSource ) source ) . getSource ( ) ; while ( actualSource == null && diff < 60000 ) { Thread . sleep ( 100 ) ; actualSource = ( ( DSource ) source ) . getSource ( ) ; endTime = System . currentTimeMillis ( ) ; diff = endTime - startTime ; } if ( actualSource == null ) { throw new IllegalStateException ( "Actual source is null, pipeline may not have been initialized" ) ; } source = actualSource ; } if ( ! ( source instanceof ClusterSource ) ) { throw new IllegalArgumentException ( "Source is not of type ClusterSource: " + source . getClass ( ) . getName ( ) ) ; } embeddedSDC . setSource ( ( ClusterSource ) source ) ; embeddedSDC . setSparkProcessors ( ( List < Object > ) pipelineStartResult . getClass ( ) . getDeclaredField ( "sparkProcessors" ) . get ( pipelineStartResult ) ) ; return embeddedSDC ;


signature:
public class EmbeddedSDCPool { /** * What is this ? !
 * Spark could end up scheduling transformers to run on machines that may actually not have received the data , due to :
 * - Non - local reads : Spark Locality is time - bound , so if local executor where the data is located is not available ,
 * Spark will tell another one to pick up the task .
 * - Shuffle in the transformer : After a shuffle , it is possible the next task passing the data to the next batch could
 * end up on a different executor .
 * In both these cases , it is possible that there is no SDC at that id . So we create a new SDC and fast - forward to
 * that id . Since a read can never happen from a fast - forwarded id , we read the fake batch we passed till there , and
 * then return the sdc so it is ready for writes / errors . */
private EmbeddedSDC fastForward ( int id ) throws Exception { } }

implementation:
if ( id == sparkProcessorCount ) { return null ; } LOG . info ( "No SDC was found at ID: " + id + ". Fast-forwarding.." ) ; EmbeddedSDC sdc ; // If there are not SDCs that are just idling , create a new one , else return one from the not started pool .
if ( notStarted . isEmpty ( ) ) { sdc = create ( ) ; } else { sdc = getNotStartedSDC ( ) ; } Class < ? > clusterFunctionClass = Class . forName ( "com.streamsets.pipeline.cluster.ClusterFunctionImpl" ) ; Method getBatch = clusterFunctionClass . getMethod ( "getNextBatch" , int . class , EmbeddedSDC . class ) ; Method forward = clusterFunctionClass . getMethod ( "doForward" , Iterator . class , int . class , EmbeddedSDC . class ) ; sdc . getSource ( ) . put ( Collections . emptyList ( ) ) ; getBatch . invoke ( null , 0 , sdc ) ; for ( int i = 0 ; i <= id - 1 ; i ++ ) { forward . invoke ( null , Collections . emptyIterator ( ) , i , sdc ) ; } return sdc ;


signature:
public class MongoDBProcessor { /** * Set the lookup reuslt in the result field
 * @ param record Lookup result
 * @ param fields Field name configured in " result field " */
private void setFieldsInRecord ( Record record , Map < String , Field > fields ) { } }

implementation: record . set ( configBean . resultField , Field . createListMap ( new LinkedHashMap < > ( fields ) ) ) ;