fn create_partition_queue(&mut self, partition_id: i32, initial_offset: Offset) {
        info!(
            "Activating Kafka queue for {} [{}] (source {}) on worker {}",
            self.topic_name, partition_id, self.id, self.worker_id
        );

        // Collect old partition assignments
        let tpl = self.consumer.assignment().unwrap();
        // Create list from assignments
        let mut partition_list = TopicPartitionList::new();
        for partition in tpl.elements_for_topic(&self.topic_name) {
            partition_list
                .add_partition_offset(partition.topic(), partition.partition(), partition.offset())
                .expect("offset known to be valid");
        }
        // Add new partition
        partition_list
            .add_partition_offset(&self.topic_name, partition_id, initial_offset)
            .expect("offset known to be valid");
        self.consumer
            .assign(&partition_list)
            .expect("assignment known to be valid");

        // Since librdkafka v1.6.0, we need to recreate all partition queues
        // after every call to `self.consumer.assign`.
        let context = self.consumer.context().clone();
        for pc in &mut self.partition_consumers {
            pc.partition_queue = self
                .consumer
                .split_partition_queue(&self.topic_name, pc.pid)
                .expect("partition known to be valid");
            pc.partition_queue.set_nonempty_callback({
                let context = context.clone();
                move || context.activate()
            });
        }

        let mut partition_queue = self
            .consumer
            .split_partition_queue(&self.topic_name, partition_id)
            .expect("partition known to be valid");
        partition_queue.set_nonempty_callback(move || context.activate());
        self.partition_consumers
            .push_front(PartitionConsumer::new(partition_id, partition_queue));
        assert_eq!(
            self.consumer
                .assignment()
                .unwrap()
                .elements_for_topic(&self.topic_name)
                .len(),
            self.partition_consumers.len()
        );
    }
/// Creates a new partition queue for `partition_id`.
Creates a new partition queue for `partition_id`.
Creates a new partition queue for ` partition_id ` .
fn create_partition_queue(&mut self, partition_id: i32, initial_offset: Offset) {
        info!(
            "Activating Kafka queue for {} [{}] (source {}) on worker {}",
            self.topic_name, partition_id, self.id, self.worker_id
        );
        // Collect old partition assignments
        let tpl = self.consumer.assignment().unwrap();
        // Create list from assignments
        let mut partition_list = TopicPartitionList::new();
        for partition in tpl.elements_for_topic(&self.topic_name) {
            partition_list
                .add_partition_offset(partition.topic(), partition.partition(), partition.offset())
                .expect("offset known to be valid");
        }
        // Add new partition
        partition_list
            .add_partition_offset(&self.topic_name, partition_id, initial_offset)
            .expect("offset known to be valid");
        self.consumer
            .assign(&partition_list)
            .expect("assignment known to be valid");
        // Since librdkafka v1.6.0, we need to recreate all partition queues
        // after every call to `self.consumer.assign`.
        let context = self.consumer.context().clone();
        for pc in &mut self.partition_consumers {
            pc.partition_queue = self
                .consumer
                .split_partition_queue(&self.topic_name, pc.pid)
                .expect("partition known to be valid");
            pc.partition_queue.set_nonempty_callback({
                let context = context.clone();
                move || context.activate()
            });
        }
        let mut partition_queue = self
            .consumer
            .split_partition_queue(&self.topic_name, partition_id)
            .expect("partition known to be valid");
        partition_queue.set_nonempty_callback(move || context.activate());
        self.partition_consumers
            .push_front(PartitionConsumer::new(partition_id, partition_queue));
        assert_eq!(
            self.consumer
                .assignment()
                .unwrap()
                .elements_for_topic(&self.topic_name)
                .len(),
            self.partition_consumers.len()
        );
    }
fn create_partition_queue ( & mut self , partition_id : i32 , initial_offset : Offset ) { info ! ( " " self topic_name partition_id self id self worker_id ) ; // Collect old partition assignments let tpl = self . consumer . assignment ( ) . unwrap ( ) ; // Create list from assignments let mut partition_list = TopicPartitionList :: new ( ) ; for partition in tpl . elements_for_topic ( & self . topic_name ) { partition_list . add_partition_offset ( partition . topic ( ) , partition . partition ( ) , partition . offset ( ) ) . expect ( " " ) ; } // Add new partition partition_list . add_partition_offset ( & self . topic_name , partition_id , initial_offset ) . expect ( " " ) ; self . consumer . assign ( & partition_list ) . expect ( " " ) ; // Since librdkafka v1.6.0, we need to recreate all partition queues // after every call to `self.consumer.assign`. let context = self . consumer . context ( ) . clone ( ) ; for pc in & mut self . partition_consumers { pc . partition_queue = self . consumer . split_partition_queue ( & self . topic_name , pc . pid ) . expect ( " " ) ; pc . partition_queue . set_nonempty_callback ( { let context = context . clone ( ) ; move | | context . activate ( ) } ) ; } let mut partition_queue = self . consumer . split_partition_queue ( & self . topic_name , partition_id ) . expect ( " " ) ; partition_queue . set_nonempty_callback ( move | | context . activate ( ) ) ; self . partition_consumers . push_front ( PartitionConsumer :: new ( partition_id , partition_queue ) ) ; assert_eq ! ( self consumer assignment ( ) unwrap ( ) elements_for_topic ( self topic_name ) len ( ) self partition_consumers len ( ) ) ; }
Creates a new partition queue for `partition_id`.
Creates a new partition queue for ` partition_id ` .
&mut self
partition_id
i32
initial_offset
Offset
&mut self
partition_id
i32
initial_offset
Offset