protected int persistHydrant ( FireHydrant indexToPersist , DataSchema schema , Interval interval , Map < String , Object > metadataElems ) { synchronized ( indexToPersist ) { if ( indexToPersist . hasSwapped ( ) ) { log . info ( "DataSource[%s], Interval[%s], Hydrant[%s] already swapped. Ignoring request to persist." , schema . getDataSource ( ) , interval , indexToPersist ) ; return 0 ; } log . info ( "DataSource[%s], Interval[%s], Metadata [%s] persisting Hydrant[%s]" , schema . getDataSource ( ) , interval , metadataElems , indexToPersist ) ; try { int numRows = indexToPersist . getIndex ( ) . size ( ) ; final IndexSpec indexSpec = config . getIndexSpec ( ) ; indexToPersist . getIndex ( ) . getMetadata ( ) . putAll ( metadataElems ) ; final File persistedFile = indexMerger . persist ( indexToPersist . getIndex ( ) , interval , new File ( computePersistDir ( schema , interval ) , String . valueOf ( indexToPersist . getCount ( ) ) ) , indexSpec , config . getSegmentWriteOutMediumFactory ( ) ) ; indexToPersist . swapSegment ( new QueryableIndexSegment ( indexIO . loadIndex ( persistedFile ) , indexToPersist . getSegmentId ( ) ) ) ; return numRows ; } catch ( IOException e ) { log . makeAlert ( "dataSource[%s] -- incremental persist failed" , schema . getDataSource ( ) ) . addData ( "interval" , interval ) . addData ( "count" , indexToPersist . getCount ( ) ) . emit ( ) ; throw new RuntimeException ( e ) ; } } }
Persists the given hydrant and returns the number of rows persisted