private Histogram getHistogram ( String entity , String watermarkColumn , SourceState state , Partition partition ) { SalesforceConnector connector = getConnector ( state ) ; try { if ( ! connector . connect ( ) ) { throw new RuntimeException ( "Failed to connect." ) ; } } catch ( RestApiConnectionException e ) { throw new RuntimeException ( "Failed to connect." , e ) ; } Histogram histogram = getHistogramByDayBucketing ( connector , entity , watermarkColumn , partition ) ; // exchange the first histogram group key with the global low watermark to ensure that the low watermark is captured // in the range of generated partitions HistogramGroup firstGroup = histogram . get ( 0 ) ; Date lwmDate = Utils . toDate ( partition . getLowWatermark ( ) , Partitioner . WATERMARKTIMEFORMAT ) ; histogram . getGroups ( ) . set ( 0 , new HistogramGroup ( Utils . epochToDate ( lwmDate . getTime ( ) , SECONDS_FORMAT ) , firstGroup . getCount ( ) ) ) ; // refine the histogram if ( state . getPropAsBoolean ( ENABLE_DYNAMIC_PROBING ) ) { histogram = getRefinedHistogram ( connector , entity , watermarkColumn , state , partition , histogram ) ; } return histogram ; }
Generate the histogram