rocketmq_client_rust/consumer/
default_lite_pull_consumer_builder.rs

1// Copyright 2023 The RocketMQ Rust Authors
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::sync::Arc;
16
17use cheetah_string::CheetahString;
18use rocketmq_common::common::consumer::consume_from_where::ConsumeFromWhere;
19use rocketmq_common::common::message::message_enum::MessageRequestMode;
20use rocketmq_common::common::mix_all;
21use rocketmq_error::RocketMQResult;
22use rocketmq_remoting::protocol::heartbeat::message_model::MessageModel;
23use rocketmq_remoting::runtime::RPCHook;
24use rocketmq_rust::ArcMut;
25
26use crate::base::client_config::ClientConfig;
27use crate::consumer::allocate_message_queue_strategy::AllocateMessageQueueStrategy;
28use crate::consumer::consumer_impl::default_lite_pull_consumer_impl::default_lite_pull_consume_timestamp;
29use crate::consumer::consumer_impl::default_lite_pull_consumer_impl::validate_lite_pull_consume_from_where;
30use crate::consumer::consumer_impl::default_lite_pull_consumer_impl::LitePullConsumerConfig;
31use crate::consumer::default_lite_pull_consumer::DefaultLitePullConsumer;
32use crate::consumer::rebalance_strategy::allocate_message_queue_averagely::AllocateMessageQueueAveragely;
33use crate::trace::trace_dispatcher::TraceDispatcher;
34
35pub(crate) const MIN_AUTOCOMMIT_INTERVAL_MILLIS: u64 = 1000;
36
37/// Builder for creating a [`DefaultLitePullConsumer`] with customized configuration.
38///
39/// # Examples
40///
41/// ```rust,ignore
42/// use rocketmq_client::consumer::default_lite_pull_consumer::DefaultLitePullConsumer;
43///
44/// let consumer = DefaultLitePullConsumer::builder()
45///     .consumer_group("my_consumer_group")
46///     .name_server_addr("127.0.0.1:9876")
47///     .pull_batch_size(32)
48///     .auto_commit(true)
49///     .build()?;
50/// ```
51pub struct DefaultLitePullConsumerBuilder {
52    // Client configuration
53    name_server_addr: Option<CheetahString>,
54    client_ip: Option<CheetahString>,
55    instance_name: Option<CheetahString>,
56    namespace: Option<CheetahString>,
57    access_channel: Option<CheetahString>,
58    use_tls: bool,
59
60    // Consumer configuration
61    consumer_group: Option<CheetahString>,
62    message_model: MessageModel,
63    consume_from_where: ConsumeFromWhere,
64    consume_timestamp: Option<CheetahString>,
65    allocate_message_queue_strategy: Arc<dyn AllocateMessageQueueStrategy + Send + Sync>,
66    unit_mode: bool,
67
68    // Pull configuration
69    pull_batch_size: i32,
70    pull_thread_nums: usize,
71    connect_broker_by_user: bool,
72    default_broker_id: u64,
73
74    // Flow control
75    pull_threshold_for_queue: i64,
76    pull_threshold_size_for_queue: i32,
77    pull_threshold_for_all: i64,
78    consume_max_span: i64,
79
80    // Backoff delays
81    pull_time_delay_millis_when_exception: u64,
82    pull_time_delay_millis_when_cache_flow_control: u64,
83    pull_time_delay_millis_when_broker_flow_control: u64,
84    broker_suspend_max_time_millis: u64,
85    consumer_timeout_millis_when_suspend: u64,
86    consumer_pull_timeout_millis: u64,
87
88    // Poll configuration
89    poll_timeout_millis: u64,
90
91    // Auto-commit
92    auto_commit: bool,
93    auto_commit_interval_millis: u64,
94
95    // Miscellaneous
96    topic_metadata_check_interval_millis: u64,
97    message_request_mode: MessageRequestMode,
98
99    // Advanced
100    rpc_hook: Option<Arc<dyn RPCHook>>,
101    trace_dispatcher: Option<Arc<dyn TraceDispatcher + Send + Sync>>,
102    enable_msg_trace: bool,
103    custom_trace_topic: Option<CheetahString>,
104}
105
106impl Default for DefaultLitePullConsumerBuilder {
107    fn default() -> Self {
108        Self::new()
109    }
110}
111
112impl DefaultLitePullConsumerBuilder {
113    /// Creates a new builder with default configuration.
114    pub fn new() -> Self {
115        Self {
116            name_server_addr: None,
117            client_ip: None,
118            instance_name: None,
119            namespace: None,
120            access_channel: None,
121            use_tls: false,
122            consumer_group: None,
123            message_model: MessageModel::Clustering,
124            consume_from_where: ConsumeFromWhere::ConsumeFromLastOffset,
125            consume_timestamp: Some(default_lite_pull_consume_timestamp()),
126            allocate_message_queue_strategy: Arc::new(AllocateMessageQueueAveragely),
127            unit_mode: false,
128            pull_batch_size: 10,
129            pull_thread_nums: 20,
130            connect_broker_by_user: false,
131            default_broker_id: mix_all::MASTER_ID,
132            pull_threshold_for_queue: 1000,
133            pull_threshold_size_for_queue: 100,
134            pull_threshold_for_all: 10000,
135            consume_max_span: 2000,
136            pull_time_delay_millis_when_exception: 1000,
137            pull_time_delay_millis_when_cache_flow_control: 50,
138            pull_time_delay_millis_when_broker_flow_control: 20,
139            broker_suspend_max_time_millis: 20_000,
140            consumer_timeout_millis_when_suspend: 30_000,
141            consumer_pull_timeout_millis: 10_000,
142            poll_timeout_millis: 5000,
143            auto_commit: true,
144            auto_commit_interval_millis: 5000,
145            topic_metadata_check_interval_millis: 30000,
146            message_request_mode: MessageRequestMode::Pull,
147            rpc_hook: None,
148            trace_dispatcher: None,
149            enable_msg_trace: false,
150            custom_trace_topic: None,
151        }
152    }
153
154    /// Sets the name server address (required).
155    ///
156    /// # Examples
157    ///
158    /// ```rust,ignore
159    /// builder.name_server_addr("127.0.0.1:9876");
160    /// ```
161    pub fn name_server_addr(mut self, addr: impl Into<CheetahString>) -> Self {
162        self.name_server_addr = Some(addr.into());
163        self
164    }
165
166    /// Sets the consumer group name (required).
167    ///
168    /// # Examples
169    ///
170    /// ```rust,ignore
171    /// builder.consumer_group("my_consumer_group");
172    /// ```
173    pub fn consumer_group(mut self, group: impl Into<CheetahString>) -> Self {
174        self.consumer_group = Some(group.into());
175        self
176    }
177
178    /// Sets the client IP address (optional, auto-detected by default).
179    pub fn client_ip(mut self, ip: impl Into<CheetahString>) -> Self {
180        self.client_ip = Some(ip.into());
181        self
182    }
183
184    /// Sets the instance name (optional, auto-generated by default).
185    pub fn instance_name(mut self, name: impl Into<CheetahString>) -> Self {
186        self.instance_name = Some(name.into());
187        self
188    }
189
190    /// Sets the namespace (optional).
191    pub fn namespace(mut self, namespace: impl Into<CheetahString>) -> Self {
192        self.namespace = Some(namespace.into());
193        self
194    }
195
196    /// Sets whether client remoting connections should use TLS.
197    pub fn use_tls(mut self, use_tls: bool) -> Self {
198        self.use_tls = use_tls;
199        self
200    }
201
202    /// Sets the message model (default: Clustering).
203    pub fn message_model(mut self, model: MessageModel) -> Self {
204        self.message_model = model;
205        self
206    }
207
208    /// Sets where to start consuming from when no offset exists (default: LastOffset).
209    pub fn consume_from_where(mut self, consume_from: ConsumeFromWhere) -> Self {
210        self.consume_from_where = consume_from;
211        self
212    }
213
214    /// Sets the timestamp to consume from (for CONSUME_FROM_TIMESTAMP mode).
215    pub fn consume_timestamp(mut self, timestamp: impl Into<CheetahString>) -> Self {
216        self.consume_timestamp = Some(timestamp.into());
217        self
218    }
219
220    /// Sets the message queue allocation strategy.
221    pub fn allocate_message_queue_strategy(
222        mut self,
223        strategy: Arc<dyn AllocateMessageQueueStrategy + Send + Sync>,
224    ) -> Self {
225        self.allocate_message_queue_strategy = strategy;
226        self
227    }
228
229    /// Sets whether the subscription group runs in unit mode (default: false).
230    pub fn unit_mode(mut self, unit_mode: bool) -> Self {
231        self.unit_mode = unit_mode;
232        self
233    }
234
235    /// Sets the number of messages to pull in a single request (default: 10, range: 1-1024).
236    pub fn pull_batch_size(mut self, size: i32) -> Self {
237        self.pull_batch_size = size;
238        self
239    }
240
241    /// Sets the number of concurrent pull threads (default: 20).
242    pub fn pull_thread_nums(mut self, nums: usize) -> Self {
243        self.pull_thread_nums = nums;
244        self
245    }
246
247    /// Sets whether pulls should always use the configured default broker ID.
248    pub fn connect_broker_by_user(mut self, connect: bool) -> Self {
249        self.connect_broker_by_user = connect;
250        self
251    }
252
253    /// Sets the broker ID used when user-controlled broker selection is enabled.
254    pub fn default_broker_id(mut self, broker_id: u64) -> Self {
255        self.default_broker_id = broker_id;
256        self
257    }
258
259    /// Sets the maximum number of messages cached per queue (default: 1000).
260    pub fn pull_threshold_for_queue(mut self, threshold: i64) -> Self {
261        self.pull_threshold_for_queue = threshold;
262        self
263    }
264
265    /// Sets the maximum size in MiB of messages cached per queue (default: 100).
266    pub fn pull_threshold_size_for_queue(mut self, threshold: i32) -> Self {
267        self.pull_threshold_size_for_queue = threshold;
268        self
269    }
270
271    /// Sets the maximum total number of cached messages across all queues (default: 10000).
272    pub fn pull_threshold_for_all(mut self, threshold: i64) -> Self {
273        self.pull_threshold_for_all = threshold;
274        self
275    }
276
277    /// Sets the maximum offset span allowed in a process queue (default: 2000).
278    pub fn consume_max_span(mut self, span: i64) -> Self {
279        self.consume_max_span = span;
280        self
281    }
282
283    /// Sets the delay when pull encounters an exception (default: 1000ms).
284    pub fn pull_time_delay_millis_when_exception(mut self, delay: u64) -> Self {
285        self.pull_time_delay_millis_when_exception = delay;
286        self
287    }
288
289    /// Sets the delay when cache flow control is triggered (default: 50ms).
290    pub fn pull_time_delay_millis_when_cache_flow_control(mut self, delay: u64) -> Self {
291        self.pull_time_delay_millis_when_cache_flow_control = delay;
292        self
293    }
294
295    /// Sets the delay when broker flow control is triggered (default: 20ms).
296    pub fn pull_time_delay_millis_when_broker_flow_control(mut self, delay: u64) -> Self {
297        self.pull_time_delay_millis_when_broker_flow_control = delay;
298        self
299    }
300
301    /// Sets the maximum time that the broker may suspend a long-poll pull request (default:
302    /// 20000ms).
303    pub fn broker_suspend_max_time_millis(mut self, timeout: u64) -> Self {
304        self.broker_suspend_max_time_millis = timeout;
305        self
306    }
307
308    /// Sets the consumer-side timeout for suspended long-poll pull requests (default: 30000ms).
309    pub fn consumer_timeout_millis_when_suspend(mut self, timeout: u64) -> Self {
310        self.consumer_timeout_millis_when_suspend = timeout;
311        self
312    }
313
314    /// Sets the lite pull RPC timeout in milliseconds (default: 10000ms).
315    pub fn consumer_pull_timeout_millis(mut self, timeout: u64) -> Self {
316        self.consumer_pull_timeout_millis = timeout;
317        self
318    }
319
320    /// Sets the default poll timeout in milliseconds (default: 5000).
321    pub fn poll_timeout_millis(mut self, timeout: u64) -> Self {
322        self.poll_timeout_millis = timeout;
323        self
324    }
325
326    /// Sets whether to automatically commit offsets (default: true).
327    pub fn auto_commit(mut self, enable: bool) -> Self {
328        self.auto_commit = enable;
329        self
330    }
331
332    /// Sets the interval between automatic offset commits (default: 5000ms, minimum: 1000ms).
333    pub fn auto_commit_interval_millis(mut self, interval: u64) -> Self {
334        if interval >= MIN_AUTOCOMMIT_INTERVAL_MILLIS {
335            self.auto_commit_interval_millis = interval;
336        }
337        self
338    }
339
340    /// Sets the interval for checking topic metadata changes (default: 30000ms).
341    pub fn topic_metadata_check_interval_millis(mut self, interval: u64) -> Self {
342        self.topic_metadata_check_interval_millis = interval;
343        self
344    }
345
346    /// Sets the message request mode (default: Pull).
347    pub fn message_request_mode(mut self, mode: MessageRequestMode) -> Self {
348        self.message_request_mode = mode;
349        self
350    }
351
352    /// Sets the RPC hook for request/response interception.
353    pub fn rpc_hook(mut self, hook: Arc<dyn RPCHook>) -> Self {
354        self.rpc_hook = Some(hook);
355        self
356    }
357
358    /// Enables message trace with the default trace topic.
359    pub fn enable_msg_trace(mut self) -> Self {
360        self.enable_msg_trace = true;
361        self
362    }
363
364    /// Enables message trace with a custom trace topic.
365    pub fn enable_msg_trace_with_topic(mut self, trace_topic: impl Into<CheetahString>) -> Self {
366        self.enable_msg_trace = true;
367        self.custom_trace_topic = Some(trace_topic.into());
368        self
369    }
370
371    /// Sets a custom trace dispatcher.
372    pub fn trace_dispatcher(mut self, dispatcher: Arc<dyn TraceDispatcher + Send + Sync>) -> Self {
373        self.trace_dispatcher = Some(dispatcher);
374        self
375    }
376
377    /// Builds the [`DefaultLitePullConsumer`].
378    pub fn build(self) -> RocketMQResult<DefaultLitePullConsumer> {
379        let Some(consumer_group) = self.consumer_group else {
380            return Err(mq_client_err!("consumer_group is required"));
381        };
382        validate_lite_pull_consume_from_where(self.consume_from_where)?;
383
384        let mut client_config = ClientConfig::default();
385        client_config.set_enable_stream_request_type(true);
386        if let Some(name_server_addr) = self.name_server_addr {
387            client_config.set_namesrv_addr(name_server_addr);
388        }
389
390        if let Some(ip) = self.client_ip {
391            client_config.set_client_ip(ip);
392        }
393
394        if let Some(name) = self.instance_name {
395            client_config.set_instance_name(name);
396        }
397
398        if let Some(namespace) = self.namespace {
399            client_config.set_namespace(namespace);
400        }
401        client_config.set_use_tls(self.use_tls);
402
403        let consumer_config = LitePullConsumerConfig {
404            consumer_group,
405            message_model: self.message_model,
406            consume_from_where: self.consume_from_where,
407            consume_timestamp: self.consume_timestamp,
408            allocate_message_queue_strategy: self.allocate_message_queue_strategy,
409            unit_mode: self.unit_mode,
410            pull_batch_size: self.pull_batch_size,
411            pull_thread_nums: self.pull_thread_nums,
412            connect_broker_by_user: self.connect_broker_by_user,
413            default_broker_id: self.default_broker_id,
414            pull_threshold_for_queue: self.pull_threshold_for_queue,
415            pull_threshold_size_for_queue: self.pull_threshold_size_for_queue,
416            pull_threshold_for_all: self.pull_threshold_for_all,
417            consume_max_span: self.consume_max_span,
418            pull_time_delay_millis_when_exception: self.pull_time_delay_millis_when_exception,
419            pull_time_delay_millis_when_cache_flow_control: self.pull_time_delay_millis_when_cache_flow_control,
420            pull_time_delay_millis_when_broker_flow_control: self.pull_time_delay_millis_when_broker_flow_control,
421            broker_suspend_max_time_millis: self.broker_suspend_max_time_millis,
422            consumer_timeout_millis_when_suspend: self.consumer_timeout_millis_when_suspend,
423            consumer_pull_timeout_millis: self.consumer_pull_timeout_millis,
424            poll_timeout_millis: self.poll_timeout_millis,
425            auto_commit: self.auto_commit,
426            auto_commit_interval_millis: self.auto_commit_interval_millis,
427            topic_metadata_check_interval_millis: self.topic_metadata_check_interval_millis,
428            message_request_mode: self.message_request_mode,
429        };
430
431        Ok(DefaultLitePullConsumer::new(
432            ArcMut::new(client_config),
433            ArcMut::new(consumer_config),
434            self.rpc_hook,
435            self.trace_dispatcher,
436            self.enable_msg_trace,
437            self.custom_trace_topic,
438        ))
439    }
440}
441
442#[cfg(test)]
443mod tests {
444    use super::*;
445
446    #[test]
447    fn build_without_consumer_group_returns_error() {
448        let result = DefaultLitePullConsumerBuilder::new().build();
449        match result {
450            Ok(_) => panic!("builder should reject missing consumer group"),
451            Err(error) => assert!(error.to_string().contains("consumer_group is required")),
452        }
453    }
454
455    #[test]
456    fn build_without_name_server_addr_uses_default_client_config() {
457        let consumer = DefaultLitePullConsumerBuilder::new()
458            .consumer_group("lite_pull_group")
459            .build()
460            .expect("builder should use default client config");
461
462        assert_eq!(consumer.consumer_group().as_str(), "lite_pull_group");
463    }
464
465    #[test]
466    fn build_enables_stream_request_type_like_java_lite_pull_constructor() {
467        let consumer = DefaultLitePullConsumerBuilder::new()
468            .consumer_group("lite_pull_group")
469            .build()
470            .expect("builder should use Java LitePull client id mode");
471
472        assert!(consumer.client_config().is_enable_stream_request_type());
473        assert!(consumer.client_config().build_mq_client_id().ends_with("@STREAM"));
474    }
475
476    #[test]
477    fn build_use_tls_initializes_client_config_like_java_client_config() {
478        let consumer = DefaultLitePullConsumerBuilder::new()
479            .consumer_group("lite_pull_tls_group")
480            .use_tls(true)
481            .build()
482            .expect("builder should apply TLS flag");
483
484        assert!(consumer.is_use_tls());
485        assert!(consumer.client_config().is_use_tls());
486    }
487
488    #[test]
489    #[allow(deprecated)]
490    fn build_rejects_unsupported_consume_from_where_like_java_lite_pull() {
491        let result = DefaultLitePullConsumerBuilder::new()
492            .consumer_group("lite_pull_group")
493            .consume_from_where(ConsumeFromWhere::ConsumeFromMinOffset)
494            .build();
495
496        match result {
497            Ok(_) => panic!("Java LitePull rejects legacy consumeFromWhere values"),
498            Err(error) => assert!(error.to_string().contains("Invalid ConsumeFromWhere Value")),
499        }
500    }
501
502    #[test]
503    fn default_pull_threshold_for_all_matches_java() {
504        let consumer = DefaultLitePullConsumerBuilder::new()
505            .consumer_group("lite_pull_group")
506            .build()
507            .expect("builder should use Java default pull threshold");
508
509        assert_eq!(consumer.consumer_config().pull_threshold_for_all, 10000);
510        assert_eq!(LitePullConsumerConfig::default().pull_threshold_for_all, 10000);
511    }
512
513    #[test]
514    fn pull_thread_nums_defaults_and_overrides_match_java_lite_pull() {
515        let default_consumer = DefaultLitePullConsumerBuilder::new()
516            .consumer_group("lite_pull_group")
517            .build()
518            .expect("builder should use Java default pull thread count");
519
520        assert_eq!(default_consumer.consumer_config().pull_thread_nums, 20);
521        assert_eq!(LitePullConsumerConfig::default().pull_thread_nums, 20);
522
523        let custom_consumer = DefaultLitePullConsumerBuilder::new()
524            .consumer_group("lite_pull_group")
525            .pull_thread_nums(7)
526            .build()
527            .expect("builder should preserve configured pull thread count");
528
529        assert_eq!(custom_consumer.consumer_config().pull_thread_nums, 7);
530    }
531
532    #[tokio::test]
533    async fn broker_selection_defaults_and_overrides_match_java_lite_pull() {
534        let default_consumer = DefaultLitePullConsumerBuilder::new()
535            .consumer_group("lite_pull_group")
536            .build()
537            .expect("builder should use Java default broker selection");
538
539        assert!(!default_consumer.is_connect_broker_by_user().await);
540        assert_eq!(default_consumer.default_broker_id().await, mix_all::MASTER_ID);
541        assert!(!LitePullConsumerConfig::default().connect_broker_by_user);
542        assert_eq!(LitePullConsumerConfig::default().default_broker_id, mix_all::MASTER_ID);
543
544        let custom_consumer = DefaultLitePullConsumerBuilder::new()
545            .consumer_group("lite_pull_group")
546            .connect_broker_by_user(true)
547            .default_broker_id(2)
548            .build()
549            .expect("builder should preserve configured broker selection");
550
551        assert!(custom_consumer.is_connect_broker_by_user().await);
552        assert_eq!(custom_consumer.default_broker_id().await, 2);
553    }
554
555    #[tokio::test]
556    async fn unit_mode_defaults_and_overrides_match_java_lite_pull() {
557        let default_consumer = DefaultLitePullConsumerBuilder::new()
558            .consumer_group("lite_pull_group")
559            .build()
560            .expect("builder should use Java default unit mode");
561
562        assert!(!default_consumer.is_unit_mode().await);
563        assert!(!default_consumer.consumer_config().unit_mode);
564        assert!(!LitePullConsumerConfig::default().unit_mode);
565
566        let custom_consumer = DefaultLitePullConsumerBuilder::new()
567            .consumer_group("lite_pull_group")
568            .unit_mode(true)
569            .build()
570            .expect("builder should preserve configured unit mode");
571
572        assert!(custom_consumer.is_unit_mode().await);
573        assert!(custom_consumer.consumer_config().unit_mode);
574    }
575
576    #[test]
577    fn long_polling_timeout_defaults_and_overrides_match_java_lite_pull() {
578        let default_consumer = DefaultLitePullConsumerBuilder::new()
579            .consumer_group("lite_pull_group")
580            .build()
581            .expect("builder should use Java default long polling timeouts");
582
583        assert_eq!(
584            default_consumer.consumer_config().broker_suspend_max_time_millis,
585            20_000
586        );
587        assert_eq!(
588            default_consumer.consumer_config().consumer_timeout_millis_when_suspend,
589            30_000
590        );
591        assert_eq!(default_consumer.consumer_config().consumer_pull_timeout_millis, 10_000);
592        assert_eq!(LitePullConsumerConfig::default().broker_suspend_max_time_millis, 20_000);
593        assert_eq!(
594            LitePullConsumerConfig::default().consumer_timeout_millis_when_suspend,
595            30_000
596        );
597        assert_eq!(LitePullConsumerConfig::default().consumer_pull_timeout_millis, 10_000);
598
599        let custom_consumer = DefaultLitePullConsumerBuilder::new()
600            .consumer_group("lite_pull_group")
601            .broker_suspend_max_time_millis(12_345)
602            .consumer_timeout_millis_when_suspend(23_456)
603            .consumer_pull_timeout_millis(34_567)
604            .build()
605            .expect("builder should keep configured long polling timeouts");
606
607        assert_eq!(custom_consumer.consumer_config().broker_suspend_max_time_millis, 12_345);
608        assert_eq!(
609            custom_consumer.consumer_config().consumer_timeout_millis_when_suspend,
610            23_456
611        );
612        assert_eq!(custom_consumer.consumer_config().consumer_pull_timeout_millis, 34_567);
613    }
614
615    #[test]
616    fn default_consume_timestamp_matches_java_shape() {
617        let consumer = DefaultLitePullConsumerBuilder::new()
618            .consumer_group("lite_pull_group")
619            .build()
620            .expect("builder should use Java default consume timestamp");
621        let timestamp = consumer
622            .consumer_config()
623            .consume_timestamp
624            .as_ref()
625            .expect("Java LitePull default consume timestamp should be present");
626
627        assert!(rocketmq_common::utils::util_all::parse_date(
628            timestamp,
629            rocketmq_common::utils::util_all::YYYYMMDDHHMMSS
630        )
631        .is_some());
632        let config = LitePullConsumerConfig::default();
633        let config_timestamp = config
634            .consume_timestamp
635            .as_ref()
636            .expect("Java LitePull config default consume timestamp should be present");
637        assert!(rocketmq_common::utils::util_all::parse_date(
638            config_timestamp,
639            rocketmq_common::utils::util_all::YYYYMMDDHHMMSS
640        )
641        .is_some());
642    }
643
644    #[test]
645    fn auto_commit_interval_below_java_minimum_keeps_default() {
646        let consumer = DefaultLitePullConsumerBuilder::new()
647            .consumer_group("lite_pull_group")
648            .auto_commit_interval_millis(MIN_AUTOCOMMIT_INTERVAL_MILLIS - 1)
649            .build()
650            .expect("builder should ignore invalid auto commit interval");
651
652        assert_eq!(consumer.consumer_config().auto_commit_interval_millis, 5000);
653    }
654
655    #[test]
656    fn auto_commit_interval_below_java_minimum_keeps_previous_valid_value() {
657        let consumer = DefaultLitePullConsumerBuilder::new()
658            .consumer_group("lite_pull_group")
659            .auto_commit_interval_millis(2000)
660            .auto_commit_interval_millis(MIN_AUTOCOMMIT_INTERVAL_MILLIS - 1)
661            .build()
662            .expect("builder should keep previous valid auto commit interval");
663
664        assert_eq!(consumer.consumer_config().auto_commit_interval_millis, 2000);
665    }
666}
rocketmq_client_rust/consumer/default_lite_pull_consumer_builder.rs

rocketmq_client_rust/consumer/
default_lite_pull_consumer_builder.rs
