private SegmentIdWithShardSpec getSegment ( final InputRow row , final String sequenceName , final boolean skipSegmentLineageCheck ) throws IOException { synchronized ( segments ) { final DateTime timestamp = row . getTimestamp ( ) ; final SegmentIdWithShardSpec existing = getAppendableSegment ( timestamp , sequenceName ) ; if ( existing != null ) { return existing ; } else { // Allocate new segment. final SegmentsForSequence segmentsForSequence = segments . get ( sequenceName ) ; final SegmentIdWithShardSpec newSegment = segmentAllocator . allocate ( row , sequenceName , segmentsForSequence == null ? null : segmentsForSequence . lastSegmentId , // send lastSegmentId irrespective of skipSegmentLineageCheck so that // unique constraint for sequence_name_prev_id_sha1 does not fail for // allocatePendingSegment in IndexerSQLMetadataStorageCoordinator skipSegmentLineageCheck ) ; if ( newSegment != null ) { for ( SegmentIdWithShardSpec identifier : appenderator . getSegments ( ) ) { if ( identifier . equals ( newSegment ) ) { throw new ISE ( "WTF?! Allocated segment[%s] which conflicts with existing segment[%s]." , newSegment , identifier ) ; } } log . info ( "New segment[%s] for row[%s] sequenceName[%s]." , newSegment , row , sequenceName ) ; addSegment ( sequenceName , newSegment ) ; } else { // Well, we tried. log . warn ( "Cannot allocate segment for timestamp[%s], sequenceName[%s]. " , timestamp , sequenceName ) ; } return newSegment ; } } }
Return a segment usable for timestamp . May return null if no segment can be allocated .