public static Multimap < String , QueueName > configureQueue ( Program program , FlowSpecification flowSpec , QueueAdmin queueAdmin ) { // Generate all queues specifications Table < QueueSpecificationGenerator . Node , String , Set < QueueSpecification > > queueSpecs = new SimpleQueueSpecificationGenerator ( ) . create ( flowSpec ) ; // For each queue in the flow, gather a map of consumer groupId to number of instances Table < QueueName , Long , Integer > queueConfigs = HashBasedTable . create ( ) ; // For storing result from flowletId to queue. ImmutableSetMultimap . Builder < String , QueueName > resultBuilder = ImmutableSetMultimap . builder ( ) ; // Loop through each flowlet for ( Map . Entry < String , FlowletDefinition > entry : flowSpec . getFlowlets ( ) . entrySet ( ) ) { String flowletId = entry . getKey ( ) ; long groupId = FlowUtils . generateConsumerGroupId ( program , flowletId ) ; int instances = entry . getValue ( ) . getInstances ( ) ; // For each queue that the flowlet is a consumer, store the number of instances for this flowlet for ( QueueSpecification queueSpec : Iterables . concat ( queueSpecs . column ( flowletId ) . values ( ) ) ) { queueConfigs . put ( queueSpec . getQueueName ( ) , groupId , instances ) ; resultBuilder . put ( flowletId , queueSpec . getQueueName ( ) ) ; } } try { // For each queue in the flow, configure it through QueueAdmin for ( Map . Entry < QueueName , Map < Long , Integer > > row : queueConfigs . rowMap ( ) . entrySet ( ) ) { LOG . info ( "Queue config for {} : {}" , row . getKey ( ) , row . getValue ( ) ) ; queueAdmin . configureGroups ( row . getKey ( ) , row . getValue ( ) ) ; } return resultBuilder . build ( ) ; } catch ( Exception e ) { LOG . error ( "Failed to configure queues" , e ) ; throw Throwables . propagate ( e ) ; } }
Configures all queues being used in a flow .