Skip to main content

meta_srv/
error.rs

1// Copyright 2023 Greptime Team
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 common_error::define_into_tonic_status;
16use common_error::ext::{BoxedError, ErrorExt, RetryHint};
17use common_error::status_code::StatusCode;
18use common_macro::stack_trace_debug;
19use common_meta::DatanodeId;
20use common_procedure::ProcedureId;
21use common_runtime::JoinError;
22use common_wal::kafka::rskafka_client_error_to_retry_hint;
23use snafu::{Location, Snafu};
24use store_api::storage::RegionId;
25use table::metadata::TableId;
26use tokio::sync::mpsc::error::SendError;
27use tonic::codegen::http;
28use uuid::Uuid;
29
30use crate::metasrv::SelectTarget;
31use crate::pubsub::Message;
32use crate::service::mailbox::Channel;
33
34#[derive(Snafu)]
35#[snafu(visibility(pub))]
36#[stack_trace_debug]
37pub enum Error {
38    #[snafu(display("Failed to choose items"))]
39    ChooseItems {
40        #[snafu(implicit)]
41        location: Location,
42        #[snafu(source)]
43        error: rand::distr::weighted::Error,
44    },
45
46    #[snafu(display("Exceeded deadline, operation: {}", operation))]
47    ExceededDeadline {
48        #[snafu(implicit)]
49        location: Location,
50        operation: String,
51    },
52
53    #[snafu(display("The target peer is unavailable temporally: {}", peer_id))]
54    PeerUnavailable {
55        #[snafu(implicit)]
56        location: Location,
57        peer_id: u64,
58    },
59
60    #[snafu(display("Failed to list active frontends"))]
61    ListActiveFrontends {
62        #[snafu(implicit)]
63        location: Location,
64        source: common_meta::error::Error,
65    },
66
67    #[snafu(display("Failed to list active datanodes"))]
68    ListActiveDatanodes {
69        #[snafu(implicit)]
70        location: Location,
71        source: common_meta::error::Error,
72    },
73
74    #[snafu(display("Failed to list active flownodes"))]
75    ListActiveFlownodes {
76        #[snafu(implicit)]
77        location: Location,
78        source: common_meta::error::Error,
79    },
80
81    #[snafu(display("No available frontend"))]
82    NoAvailableFrontend {
83        #[snafu(implicit)]
84        location: Location,
85    },
86
87    #[snafu(display("Another migration procedure is running for region: {}", region_id))]
88    MigrationRunning {
89        #[snafu(implicit)]
90        location: Location,
91        region_id: RegionId,
92    },
93
94    #[snafu(display("Another drain procedure is running for datanode: {}", node_id))]
95    DrainDatanodeRunning {
96        #[snafu(implicit)]
97        location: Location,
98        node_id: DatanodeId,
99    },
100
101    #[snafu(display(
102        "The region migration procedure is completed for region: {}, target_peer: {}",
103        region_id,
104        target_peer_id
105    ))]
106    RegionMigrated {
107        #[snafu(implicit)]
108        location: Location,
109        region_id: RegionId,
110        target_peer_id: u64,
111    },
112
113    #[snafu(display("The region migration procedure aborted, reason: {}", reason))]
114    MigrationAbort {
115        #[snafu(implicit)]
116        location: Location,
117        reason: String,
118    },
119
120    #[snafu(display(
121        "Drain datanode {} aborted, failed regions: {:?}",
122        node_id,
123        failed_regions
124    ))]
125    DrainDatanodeAbort {
126        node_id: DatanodeId,
127        failed_regions: Vec<RegionId>,
128        #[snafu(implicit)]
129        location: Location,
130    },
131
132    #[snafu(display(
133        "Drain datanode {} exceeded the maximum rescan rounds: {}",
134        node_id,
135        max_rounds
136    ))]
137    DrainDatanodeRescanExceeded {
138        node_id: DatanodeId,
139        max_rounds: u32,
140        #[snafu(implicit)]
141        location: Location,
142    },
143
144    #[snafu(display(
145        "Another procedure is operating the region: {} on peer: {}",
146        region_id,
147        peer_id
148    ))]
149    RegionOperatingRace {
150        #[snafu(implicit)]
151        location: Location,
152        peer_id: DatanodeId,
153        region_id: RegionId,
154    },
155
156    #[snafu(display("Failed to init ddl manager"))]
157    InitDdlManager {
158        #[snafu(implicit)]
159        location: Location,
160        source: common_meta::error::Error,
161    },
162
163    #[snafu(display("Failed to init reconciliation manager"))]
164    InitReconciliationManager {
165        #[snafu(implicit)]
166        location: Location,
167        source: common_meta::error::Error,
168    },
169
170    #[snafu(display("Failed to create default catalog and schema"))]
171    InitMetadata {
172        #[snafu(implicit)]
173        location: Location,
174        source: common_meta::error::Error,
175    },
176
177    #[snafu(display("Failed to allocate next sequence number"))]
178    NextSequence {
179        #[snafu(implicit)]
180        location: Location,
181        source: common_meta::error::Error,
182    },
183
184    #[snafu(display("Failed to set next sequence number"))]
185    SetNextSequence {
186        #[snafu(implicit)]
187        location: Location,
188        source: common_meta::error::Error,
189    },
190
191    #[snafu(display("Failed to peek sequence number"))]
192    PeekSequence {
193        #[snafu(implicit)]
194        location: Location,
195        source: common_meta::error::Error,
196    },
197
198    #[snafu(display("Failed to start telemetry task"))]
199    StartTelemetryTask {
200        #[snafu(implicit)]
201        location: Location,
202        source: common_runtime::error::Error,
203    },
204
205    #[snafu(display("Failed to submit ddl task"))]
206    SubmitDdlTask {
207        #[snafu(implicit)]
208        location: Location,
209        source: common_meta::error::Error,
210    },
211
212    #[snafu(display("Failed to submit reconcile procedure"))]
213    SubmitReconcileProcedure {
214        #[snafu(implicit)]
215        location: Location,
216        source: common_meta::error::Error,
217    },
218
219    #[snafu(display("Failed to invalidate table cache"))]
220    InvalidateTableCache {
221        #[snafu(implicit)]
222        location: Location,
223        source: common_meta::error::Error,
224    },
225
226    #[snafu(display("Failed to list catalogs"))]
227    ListCatalogs {
228        #[snafu(implicit)]
229        location: Location,
230        source: BoxedError,
231    },
232
233    #[snafu(display("Failed to list {}'s schemas", catalog))]
234    ListSchemas {
235        #[snafu(implicit)]
236        location: Location,
237        catalog: String,
238        source: BoxedError,
239    },
240
241    #[snafu(display("Failed to list {}.{}'s tables", catalog, schema))]
242    ListTables {
243        #[snafu(implicit)]
244        location: Location,
245        catalog: String,
246        schema: String,
247        source: BoxedError,
248    },
249
250    #[snafu(display("Failed to join a future"))]
251    Join {
252        #[snafu(implicit)]
253        location: Location,
254        #[snafu(source)]
255        error: JoinError,
256    },
257
258    #[snafu(display(
259        "Failed to request {}, required: {}, but only {} available",
260        select_target,
261        required,
262        available
263    ))]
264    NoEnoughAvailableNode {
265        #[snafu(implicit)]
266        location: Location,
267        required: usize,
268        available: usize,
269        select_target: SelectTarget,
270    },
271
272    #[snafu(display("Failed to send shutdown signal"))]
273    SendShutdownSignal {
274        #[snafu(source)]
275        error: SendError<()>,
276    },
277
278    #[snafu(display("Failed to shutdown {} server", server))]
279    ShutdownServer {
280        #[snafu(implicit)]
281        location: Location,
282        source: servers::error::Error,
283        server: String,
284    },
285
286    #[snafu(display("Empty key is not allowed"))]
287    EmptyKey {
288        #[snafu(implicit)]
289        location: Location,
290    },
291
292    #[snafu(display("Failed to execute via Etcd"))]
293    EtcdFailed {
294        #[snafu(source)]
295        error: etcd_client::Error,
296        #[snafu(implicit)]
297        location: Location,
298    },
299
300    #[snafu(display("Failed to connect to Etcd"))]
301    ConnectEtcd {
302        #[snafu(source)]
303        error: etcd_client::Error,
304        #[snafu(implicit)]
305        location: Location,
306    },
307
308    #[snafu(display("Failed to read file: {}", path))]
309    FileIo {
310        #[snafu(source)]
311        error: std::io::Error,
312        #[snafu(implicit)]
313        location: Location,
314        path: String,
315    },
316
317    #[snafu(display("Failed to bind address {}", addr))]
318    TcpBind {
319        addr: String,
320        #[snafu(source)]
321        error: std::io::Error,
322        #[snafu(implicit)]
323        location: Location,
324    },
325
326    #[snafu(display("Failed to start gRPC server"))]
327    StartGrpc {
328        #[snafu(source)]
329        error: tonic::transport::Error,
330        #[snafu(implicit)]
331        location: Location,
332    },
333
334    #[snafu(display("Failed to start http server"))]
335    StartHttp {
336        #[snafu(implicit)]
337        location: Location,
338        source: servers::error::Error,
339    },
340
341    #[snafu(display("Failed to parse address {}", addr))]
342    ParseAddr {
343        addr: String,
344        #[snafu(source)]
345        error: std::net::AddrParseError,
346    },
347
348    #[snafu(display("Invalid lease key: {}", key))]
349    InvalidLeaseKey {
350        key: String,
351        #[snafu(implicit)]
352        location: Location,
353    },
354
355    #[snafu(display("Invalid datanode stat key: {}", key))]
356    InvalidStatKey {
357        key: String,
358        #[snafu(implicit)]
359        location: Location,
360    },
361
362    #[snafu(display("Invalid inactive region key: {}", key))]
363    InvalidInactiveRegionKey {
364        key: String,
365        #[snafu(implicit)]
366        location: Location,
367    },
368
369    #[snafu(display("Invalid node scheduling key: {}", key))]
370    InvalidNodeSchedulingKey {
371        #[snafu(implicit)]
372        location: Location,
373        key: String,
374    },
375
376    #[snafu(display("Failed to parse lease key from utf8"))]
377    LeaseKeyFromUtf8 {
378        #[snafu(source)]
379        error: std::string::FromUtf8Error,
380        #[snafu(implicit)]
381        location: Location,
382    },
383
384    #[snafu(display("Failed to parse lease value from utf8"))]
385    LeaseValueFromUtf8 {
386        #[snafu(source)]
387        error: std::string::FromUtf8Error,
388        #[snafu(implicit)]
389        location: Location,
390    },
391
392    #[snafu(display("Failed to parse invalid region key from utf8"))]
393    InvalidRegionKeyFromUtf8 {
394        #[snafu(source)]
395        error: std::string::FromUtf8Error,
396        #[snafu(implicit)]
397        location: Location,
398    },
399
400    #[snafu(display("Failed to serialize to json: {}", input))]
401    SerializeToJson {
402        input: String,
403        #[snafu(source)]
404        error: serde_json::error::Error,
405        #[snafu(implicit)]
406        location: Location,
407    },
408
409    #[snafu(display("Failed to deserialize from json payload of length {}", input_len))]
410    DeserializeFromJson {
411        input_len: usize,
412        #[snafu(source)]
413        error: serde_json::error::Error,
414        #[snafu(implicit)]
415        location: Location,
416    },
417
418    #[snafu(display("Failed to serialize config"))]
419    SerializeConfig {
420        #[snafu(source)]
421        error: serde_json::error::Error,
422        #[snafu(implicit)]
423        location: Location,
424    },
425
426    #[snafu(display("Failed to parse number: {}", err_msg))]
427    ParseNum {
428        err_msg: String,
429        #[snafu(source)]
430        error: std::num::ParseIntError,
431        #[snafu(implicit)]
432        location: Location,
433    },
434
435    #[snafu(display("Failed to parse bool: {}", err_msg))]
436    ParseBool {
437        err_msg: String,
438        #[snafu(source)]
439        error: std::str::ParseBoolError,
440        #[snafu(implicit)]
441        location: Location,
442    },
443
444    #[snafu(display("Failed to downgrade region leader, region: {}", region_id))]
445    DowngradeLeader {
446        region_id: RegionId,
447        #[snafu(implicit)]
448        location: Location,
449        #[snafu(source)]
450        source: BoxedError,
451    },
452
453    #[snafu(display("Region's leader peer changed: {}", msg))]
454    LeaderPeerChanged {
455        msg: String,
456        #[snafu(implicit)]
457        location: Location,
458    },
459
460    #[snafu(display("Invalid arguments: {}", err_msg))]
461    InvalidArguments {
462        err_msg: String,
463        #[snafu(implicit)]
464        location: Location,
465    },
466
467    #[snafu(display("Manual GC is rejected because maintenance mode is enabled"))]
468    ManualGcRejectedByMaintenanceMode {
469        #[snafu(implicit)]
470        location: Location,
471    },
472
473    #[cfg(feature = "mysql_kvbackend")]
474    #[snafu(display("Failed to parse mysql url: {}", mysql_url))]
475    ParseMySqlUrl {
476        #[snafu(source)]
477        error: sqlx::error::Error,
478        mysql_url: String,
479        #[snafu(implicit)]
480        location: Location,
481    },
482
483    #[cfg(feature = "mysql_kvbackend")]
484    #[snafu(display("Failed to decode sql value"))]
485    DecodeSqlValue {
486        #[snafu(source)]
487        error: sqlx::error::Error,
488        #[snafu(implicit)]
489        location: Location,
490    },
491
492    #[snafu(display("Failed to find table route for {table_id}"))]
493    TableRouteNotFound {
494        table_id: TableId,
495        #[snafu(implicit)]
496        location: Location,
497    },
498
499    #[snafu(display("Failed to find table route for {region_id}"))]
500    RegionRouteNotFound {
501        region_id: RegionId,
502        #[snafu(implicit)]
503        location: Location,
504    },
505
506    #[snafu(display("Table info not found: {}", table_id))]
507    TableInfoNotFound {
508        table_id: TableId,
509        #[snafu(implicit)]
510        location: Location,
511    },
512
513    #[snafu(display("Datanode table not found: {}, datanode: {}", table_id, datanode_id))]
514    DatanodeTableNotFound {
515        table_id: TableId,
516        datanode_id: DatanodeId,
517        #[snafu(implicit)]
518        location: Location,
519    },
520
521    #[snafu(display("Metasrv has no leader at this moment"))]
522    NoLeader {
523        #[snafu(implicit)]
524        location: Location,
525    },
526
527    #[snafu(display("Leader lease expired"))]
528    LeaderLeaseExpired {
529        #[snafu(implicit)]
530        location: Location,
531    },
532
533    #[snafu(display("Leader lease changed during election"))]
534    LeaderLeaseChanged {
535        #[snafu(implicit)]
536        location: Location,
537    },
538
539    #[snafu(display("Table {} not found", name))]
540    TableNotFound {
541        name: String,
542        #[snafu(implicit)]
543        location: Location,
544    },
545
546    #[snafu(display("Unsupported selector type, {}", selector_type))]
547    UnsupportedSelectorType {
548        selector_type: String,
549        #[snafu(implicit)]
550        location: Location,
551    },
552
553    #[snafu(display("Unexpected, violated: {violated}"))]
554    Unexpected {
555        violated: String,
556        #[snafu(implicit)]
557        location: Location,
558    },
559
560    #[snafu(display("Failed to create gRPC channel"))]
561    CreateChannel {
562        #[snafu(implicit)]
563        location: Location,
564        source: common_grpc::error::Error,
565    },
566
567    #[snafu(display("Failed to batch get KVs from leader's in_memory kv store"))]
568    BatchGet {
569        #[snafu(source)]
570        error: tonic::Status,
571        #[snafu(implicit)]
572        location: Location,
573    },
574
575    #[snafu(display("Failed to batch range KVs from leader's in_memory kv store"))]
576    Range {
577        #[snafu(source)]
578        error: tonic::Status,
579        #[snafu(implicit)]
580        location: Location,
581    },
582
583    #[snafu(display("Response header not found"))]
584    ResponseHeaderNotFound {
585        #[snafu(implicit)]
586        location: Location,
587    },
588
589    #[snafu(display("The requested meta node is not leader, node addr: {}", node_addr))]
590    IsNotLeader {
591        node_addr: String,
592        #[snafu(implicit)]
593        location: Location,
594    },
595
596    #[snafu(display("Invalid http body"))]
597    InvalidHttpBody {
598        #[snafu(source)]
599        error: http::Error,
600        #[snafu(implicit)]
601        location: Location,
602    },
603
604    #[snafu(display(
605        "The number of retries for the grpc call {} exceeded the limit, {}",
606        func_name,
607        retry_num
608    ))]
609    ExceededRetryLimit {
610        func_name: String,
611        retry_num: usize,
612        #[snafu(implicit)]
613        location: Location,
614    },
615
616    #[snafu(display("Invalid utf-8 value"))]
617    InvalidUtf8Value {
618        #[snafu(source)]
619        error: std::string::FromUtf8Error,
620        #[snafu(implicit)]
621        location: Location,
622    },
623
624    #[snafu(display("Missing required parameter, param: {:?}", param))]
625    MissingRequiredParameter { param: String },
626
627    #[snafu(display("Failed to start procedure manager"))]
628    StartProcedureManager {
629        #[snafu(implicit)]
630        location: Location,
631        source: common_procedure::Error,
632    },
633
634    #[snafu(display("Failed to stop procedure manager"))]
635    StopProcedureManager {
636        #[snafu(implicit)]
637        location: Location,
638        source: common_procedure::Error,
639    },
640
641    #[snafu(display("Failed to wait procedure done"))]
642    WaitProcedure {
643        #[snafu(implicit)]
644        location: Location,
645        source: common_procedure::Error,
646    },
647
648    #[snafu(display("Failed to query procedure state"))]
649    QueryProcedure {
650        #[snafu(implicit)]
651        location: Location,
652        source: common_procedure::Error,
653    },
654
655    #[snafu(display("Procedure not found: {pid}"))]
656    ProcedureNotFound {
657        #[snafu(implicit)]
658        location: Location,
659        pid: String,
660    },
661
662    #[snafu(display("Failed to submit procedure"))]
663    SubmitProcedure {
664        #[snafu(implicit)]
665        location: Location,
666        source: common_procedure::Error,
667    },
668
669    #[snafu(display("A prune task for topic {} is already running", topic))]
670    PruneTaskAlreadyRunning {
671        topic: String,
672        #[snafu(implicit)]
673        location: Location,
674    },
675
676    #[snafu(display("Schema already exists, name: {schema_name}"))]
677    SchemaAlreadyExists {
678        schema_name: String,
679        #[snafu(implicit)]
680        location: Location,
681    },
682
683    #[snafu(display("Table already exists: {table_name}"))]
684    TableAlreadyExists {
685        table_name: String,
686        #[snafu(implicit)]
687        location: Location,
688    },
689
690    #[snafu(display("Pusher not found: {pusher_id}"))]
691    PusherNotFound {
692        pusher_id: String,
693        #[snafu(implicit)]
694        location: Location,
695    },
696
697    #[snafu(display("Failed to push message: {err_msg}"))]
698    PushMessage {
699        err_msg: String,
700        #[snafu(implicit)]
701        location: Location,
702    },
703
704    #[snafu(display("Mailbox already closed: {id}"))]
705    MailboxClosed {
706        id: u64,
707        #[snafu(implicit)]
708        location: Location,
709    },
710
711    #[snafu(display("Mailbox timeout: {id}"))]
712    MailboxTimeout {
713        id: u64,
714        #[snafu(implicit)]
715        location: Location,
716    },
717
718    #[snafu(display("Mailbox receiver got an error: {id}, {err_msg}"))]
719    MailboxReceiver {
720        id: u64,
721        err_msg: String,
722        #[snafu(implicit)]
723        location: Location,
724    },
725
726    #[snafu(display("Mailbox channel closed: {channel}"))]
727    MailboxChannelClosed {
728        channel: Channel,
729        #[snafu(implicit)]
730        location: Location,
731    },
732
733    #[snafu(display("Missing request header"))]
734    MissingRequestHeader {
735        #[snafu(implicit)]
736        location: Location,
737    },
738
739    #[snafu(display("Failed to register procedure loader, type name: {}", type_name))]
740    RegisterProcedureLoader {
741        type_name: String,
742        #[snafu(implicit)]
743        location: Location,
744        source: common_procedure::error::Error,
745    },
746
747    #[snafu(display(
748        "Received unexpected instruction reply, mailbox message: {}, reason: {}",
749        mailbox_message,
750        reason
751    ))]
752    UnexpectedInstructionReply {
753        mailbox_message: String,
754        reason: String,
755        #[snafu(implicit)]
756        location: Location,
757    },
758
759    #[snafu(display("Expected to retry later, reason: {}", reason))]
760    RetryLater {
761        reason: String,
762        #[snafu(implicit)]
763        location: Location,
764    },
765
766    #[snafu(display("Expected to retry later, reason: {}", reason))]
767    RetryLaterWithSource {
768        reason: String,
769        #[snafu(implicit)]
770        location: Location,
771        source: BoxedError,
772    },
773
774    #[snafu(display("Failed to convert proto data"))]
775    ConvertProtoData {
776        #[snafu(implicit)]
777        location: Location,
778        source: common_meta::error::Error,
779    },
780
781    // this error is used for custom error mapping
782    // please do not delete it
783    #[snafu(display("Other error"))]
784    Other {
785        source: BoxedError,
786        #[snafu(implicit)]
787        location: Location,
788    },
789
790    #[snafu(display("Table metadata manager error"))]
791    TableMetadataManager {
792        source: common_meta::error::Error,
793        #[snafu(implicit)]
794        location: Location,
795    },
796
797    #[snafu(display("Runtime switch manager error"))]
798    RuntimeSwitchManager {
799        source: common_meta::error::Error,
800        #[snafu(implicit)]
801        location: Location,
802    },
803
804    #[snafu(display("Keyvalue backend error"))]
805    KvBackend {
806        source: common_meta::error::Error,
807        #[snafu(implicit)]
808        location: Location,
809    },
810
811    #[snafu(display("Failed to publish message"))]
812    PublishMessage {
813        #[snafu(source)]
814        error: SendError<Message>,
815        #[snafu(implicit)]
816        location: Location,
817    },
818
819    #[snafu(display("Too many partitions"))]
820    TooManyPartitions {
821        #[snafu(implicit)]
822        location: Location,
823    },
824
825    #[snafu(display("Failed to create repartition subtasks"))]
826    RepartitionCreateSubtasks {
827        source: partition::error::Error,
828        #[snafu(implicit)]
829        location: Location,
830    },
831
832    #[snafu(display("Failed to manage the repartition GC requirement"))]
833    RepartitionGcRequirement {
834        source: common_meta::error::Error,
835        #[snafu(implicit)]
836        location: Location,
837    },
838
839    #[snafu(display("Failed to inspect persisted repartition procedures"))]
840    InspectRepartitionProcedures {
841        source: common_procedure::Error,
842        #[snafu(implicit)]
843        location: Location,
844    },
845
846    #[snafu(display(
847        "Metasrv GC must be enabled because the cluster has durable repartition state"
848    ))]
849    RepartitionGcRequired {
850        #[snafu(implicit)]
851        location: Location,
852    },
853
854    #[snafu(display(
855        "Source partition expression '{}' does not match any existing region",
856        expr
857    ))]
858    RepartitionSourceExprMismatch {
859        expr: String,
860        #[snafu(implicit)]
861        location: Location,
862    },
863
864    #[snafu(display(
865        "Failed to get the state receiver for repartition subprocedure {}",
866        procedure_id
867    ))]
868    RepartitionSubprocedureStateReceiver {
869        procedure_id: ProcedureId,
870        #[snafu(source)]
871        source: common_procedure::Error,
872        #[snafu(implicit)]
873        location: Location,
874    },
875
876    #[snafu(display("Unsupported operation {}", operation))]
877    Unsupported {
878        operation: String,
879        #[snafu(implicit)]
880        location: Location,
881    },
882
883    #[snafu(display("Unexpected table route type: {}", err_msg))]
884    UnexpectedLogicalRouteTable {
885        #[snafu(implicit)]
886        location: Location,
887        err_msg: String,
888        source: common_meta::error::Error,
889    },
890
891    #[snafu(display("Failed to save cluster info"))]
892    SaveClusterInfo {
893        #[snafu(implicit)]
894        location: Location,
895        source: common_meta::error::Error,
896    },
897
898    #[snafu(display("Invalid cluster info format"))]
899    InvalidClusterInfoFormat {
900        #[snafu(implicit)]
901        location: Location,
902        source: common_meta::error::Error,
903    },
904
905    #[snafu(display("Invalid datanode stat format"))]
906    InvalidDatanodeStatFormat {
907        #[snafu(implicit)]
908        location: Location,
909        source: common_meta::error::Error,
910    },
911
912    #[snafu(display("Invalid node info format"))]
913    InvalidNodeInfoFormat {
914        #[snafu(implicit)]
915        location: Location,
916        source: common_meta::error::Error,
917    },
918
919    #[snafu(display("Failed to serialize options to TOML"))]
920    TomlFormat {
921        #[snafu(implicit)]
922        location: Location,
923        #[snafu(source(from(common_config::error::Error, Box::new)))]
924        source: Box<common_config::error::Error>,
925    },
926
927    #[cfg(feature = "pg_kvbackend")]
928    #[snafu(display("Failed to execute via postgres, sql: {}", sql))]
929    PostgresExecution {
930        #[snafu(source)]
931        error: tokio_postgres::Error,
932        sql: String,
933        #[snafu(implicit)]
934        location: Location,
935    },
936
937    #[cfg(feature = "pg_kvbackend")]
938    #[snafu(display("Failed to get Postgres client"))]
939    GetPostgresClient {
940        #[snafu(implicit)]
941        location: Location,
942        #[snafu(source)]
943        error: deadpool::managed::PoolError<tokio_postgres::Error>,
944    },
945
946    #[cfg(any(feature = "pg_kvbackend", feature = "mysql_kvbackend"))]
947    #[snafu(display("Sql execution timeout, sql: {}, duration: {:?}", sql, duration))]
948    SqlExecutionTimeout {
949        #[snafu(implicit)]
950        location: Location,
951        sql: String,
952        duration: std::time::Duration,
953    },
954
955    #[cfg(feature = "pg_kvbackend")]
956    #[snafu(display("Failed to create connection pool for Postgres"))]
957    CreatePostgresPool {
958        #[snafu(source)]
959        error: deadpool_postgres::CreatePoolError,
960        #[snafu(implicit)]
961        location: Location,
962    },
963
964    #[cfg(feature = "pg_kvbackend")]
965    #[snafu(display("Failed to get connection from Postgres pool: {}", reason))]
966    GetPostgresConnection {
967        reason: String,
968        #[snafu(implicit)]
969        location: Location,
970    },
971
972    #[cfg(feature = "mysql_kvbackend")]
973    #[snafu(display("Failed to execute via mysql, sql: {}", sql))]
974    MySqlExecution {
975        #[snafu(source)]
976        error: sqlx::Error,
977        #[snafu(implicit)]
978        location: Location,
979        sql: String,
980    },
981
982    #[cfg(feature = "mysql_kvbackend")]
983    #[snafu(display("Failed to create mysql pool"))]
984    CreateMySqlPool {
985        #[snafu(source)]
986        error: sqlx::Error,
987        #[snafu(implicit)]
988        location: Location,
989    },
990
991    #[cfg(feature = "mysql_kvbackend")]
992    #[snafu(display("Failed to acquire mysql client from pool"))]
993    AcquireMySqlClient {
994        #[snafu(source)]
995        error: sqlx::Error,
996        #[snafu(implicit)]
997        location: Location,
998    },
999
1000    #[snafu(display("Handler not found: {}", name))]
1001    HandlerNotFound {
1002        name: String,
1003        #[snafu(implicit)]
1004        location: Location,
1005    },
1006
1007    #[snafu(display("Flow state handler error"))]
1008    FlowStateHandler {
1009        #[snafu(implicit)]
1010        location: Location,
1011        source: common_meta::error::Error,
1012    },
1013
1014    #[snafu(display("Failed to build wal provider"))]
1015    BuildWalProvider {
1016        #[snafu(implicit)]
1017        location: Location,
1018        source: common_meta::error::Error,
1019    },
1020
1021    #[snafu(display("Failed to build kafka client."))]
1022    BuildKafkaClient {
1023        #[snafu(implicit)]
1024        location: Location,
1025        #[snafu(source)]
1026        error: common_meta::error::Error,
1027    },
1028
1029    #[snafu(display(
1030        "Failed to build a Kafka partition client, topic: {}, partition: {}",
1031        topic,
1032        partition
1033    ))]
1034    BuildPartitionClient {
1035        topic: String,
1036        partition: i32,
1037        #[snafu(implicit)]
1038        location: Location,
1039        #[snafu(source)]
1040        error: rskafka::client::error::Error,
1041    },
1042
1043    #[snafu(display(
1044        "Failed to delete records from Kafka, topic: {}, partition: {}, offset: {}",
1045        topic,
1046        partition,
1047        offset
1048    ))]
1049    DeleteRecords {
1050        #[snafu(implicit)]
1051        location: Location,
1052        #[snafu(source)]
1053        error: rskafka::client::error::Error,
1054        topic: String,
1055        partition: i32,
1056        offset: u64,
1057    },
1058
1059    #[snafu(display("Failed to get offset from Kafka, topic: {}", topic))]
1060    GetOffset {
1061        topic: String,
1062        #[snafu(implicit)]
1063        location: Location,
1064        #[snafu(source)]
1065        error: rskafka::client::error::Error,
1066    },
1067
1068    #[snafu(display("Failed to update the TopicNameValue in kvbackend, topic: {}", topic))]
1069    UpdateTopicNameValue {
1070        topic: String,
1071        #[snafu(implicit)]
1072        location: Location,
1073        #[snafu(source)]
1074        source: common_meta::error::Error,
1075    },
1076
1077    #[snafu(display("Failed to build tls options"))]
1078    BuildTlsOptions {
1079        #[snafu(implicit)]
1080        location: Location,
1081        #[snafu(source)]
1082        source: common_meta::error::Error,
1083    },
1084
1085    #[snafu(display(
1086        "Repartition group {} source region missing, region id: {}",
1087        group_id,
1088        region_id
1089    ))]
1090    RepartitionSourceRegionMissing {
1091        group_id: Uuid,
1092        region_id: RegionId,
1093        #[snafu(implicit)]
1094        location: Location,
1095    },
1096
1097    #[snafu(display(
1098        "Repartition group {} target region missing, region id: {}",
1099        group_id,
1100        region_id
1101    ))]
1102    RepartitionTargetRegionMissing {
1103        group_id: Uuid,
1104        region_id: RegionId,
1105        #[snafu(implicit)]
1106        location: Location,
1107    },
1108
1109    #[snafu(display("Failed to serialize partition expression"))]
1110    SerializePartitionExpr {
1111        #[snafu(source)]
1112        source: partition::error::Error,
1113        #[snafu(implicit)]
1114        location: Location,
1115    },
1116
1117    #[snafu(display("Failed to deserialize partition expression"))]
1118    DeserializePartitionExpr {
1119        #[snafu(source)]
1120        source: partition::error::Error,
1121        #[snafu(implicit)]
1122        location: Location,
1123    },
1124
1125    #[snafu(display("Empty partition expression"))]
1126    EmptyPartitionExpr {
1127        #[snafu(implicit)]
1128        location: Location,
1129    },
1130
1131    #[snafu(display(
1132        "Partition expression mismatch, region id: {}, expected: {}, actual: {}",
1133        region_id,
1134        expected,
1135        actual
1136    ))]
1137    PartitionExprMismatch {
1138        region_id: RegionId,
1139        expected: String,
1140        actual: String,
1141        #[snafu(implicit)]
1142        location: Location,
1143    },
1144
1145    #[snafu(display("Failed to allocate regions for table: {}", table_id))]
1146    AllocateRegions {
1147        #[snafu(implicit)]
1148        location: Location,
1149        table_id: TableId,
1150        #[snafu(source)]
1151        source: common_meta::error::Error,
1152    },
1153
1154    #[snafu(display("Failed to deallocate regions for table: {}", table_id))]
1155    DeallocateRegions {
1156        #[snafu(implicit)]
1157        location: Location,
1158        table_id: TableId,
1159        #[snafu(source)]
1160        source: common_meta::error::Error,
1161    },
1162
1163    #[snafu(display("Failed to build create request for table: {}", table_id))]
1164    BuildCreateRequest {
1165        #[snafu(implicit)]
1166        location: Location,
1167        table_id: TableId,
1168        #[snafu(source)]
1169        source: common_meta::error::Error,
1170    },
1171
1172    #[snafu(display("Failed to allocate region routes for table: {}", table_id))]
1173    AllocateRegionRoutes {
1174        #[snafu(implicit)]
1175        location: Location,
1176        table_id: TableId,
1177        #[snafu(source)]
1178        source: common_meta::error::Error,
1179    },
1180
1181    #[snafu(display("Failed to allocate wal options for table: {}", table_id))]
1182    AllocateWalOptions {
1183        #[snafu(implicit)]
1184        location: Location,
1185        table_id: TableId,
1186        #[snafu(source)]
1187        source: common_meta::error::Error,
1188    },
1189}
1190
1191impl Error {
1192    /// Returns `true` if the error is retryable.
1193    pub fn is_retryable(&self) -> bool {
1194        self.retry_hint().is_retryable()
1195    }
1196}
1197
1198pub type Result<T> = std::result::Result<T, Error>;
1199
1200define_into_tonic_status!(Error);
1201
1202impl ErrorExt for Error {
1203    fn status_code(&self) -> StatusCode {
1204        match self {
1205            Error::EtcdFailed { .. }
1206            | Error::ConnectEtcd { .. }
1207            | Error::FileIo { .. }
1208            | Error::TcpBind { .. }
1209            | Error::SerializeConfig { .. }
1210            | Error::SerializeToJson { .. }
1211            | Error::DeserializeFromJson { .. }
1212            | Error::NoLeader { .. }
1213            | Error::LeaderLeaseExpired { .. }
1214            | Error::LeaderLeaseChanged { .. }
1215            | Error::CreateChannel { .. }
1216            | Error::BatchGet { .. }
1217            | Error::Range { .. }
1218            | Error::ResponseHeaderNotFound { .. }
1219            | Error::InvalidHttpBody { .. }
1220            | Error::ExceededRetryLimit { .. }
1221            | Error::SendShutdownSignal { .. }
1222            | Error::PushMessage { .. }
1223            | Error::MailboxClosed { .. }
1224            | Error::MailboxReceiver { .. }
1225            | Error::StartGrpc { .. }
1226            | Error::PublishMessage { .. }
1227            | Error::Join { .. }
1228            | Error::ChooseItems { .. }
1229            | Error::FlowStateHandler { .. }
1230            | Error::BuildWalProvider { .. }
1231            | Error::BuildPartitionClient { .. }
1232            | Error::BuildKafkaClient { .. } => StatusCode::Internal,
1233
1234            Error::DeleteRecords { .. }
1235            | Error::GetOffset { .. }
1236            | Error::PeerUnavailable { .. }
1237            | Error::PusherNotFound { .. } => StatusCode::Unexpected,
1238            Error::MailboxTimeout { .. } | Error::ExceededDeadline { .. } => StatusCode::Cancelled,
1239            Error::PruneTaskAlreadyRunning { .. }
1240            | Error::RetryLater { .. }
1241            | Error::MailboxChannelClosed { .. }
1242            | Error::IsNotLeader { .. } => StatusCode::IllegalState,
1243            Error::RetryLaterWithSource { source, .. } => source.status_code(),
1244            Error::SerializePartitionExpr { source, .. }
1245            | Error::DeserializePartitionExpr { source, .. } => source.status_code(),
1246
1247            Error::Unsupported { .. } => StatusCode::Unsupported,
1248
1249            Error::SchemaAlreadyExists { .. } => StatusCode::DatabaseAlreadyExists,
1250
1251            Error::TableAlreadyExists { .. } => StatusCode::TableAlreadyExists,
1252            Error::EmptyKey { .. }
1253            | Error::MissingRequiredParameter { .. }
1254            | Error::MissingRequestHeader { .. }
1255            | Error::InvalidLeaseKey { .. }
1256            | Error::InvalidStatKey { .. }
1257            | Error::InvalidInactiveRegionKey { .. }
1258            | Error::InvalidNodeSchedulingKey { .. }
1259            | Error::ParseNum { .. }
1260            | Error::ParseBool { .. }
1261            | Error::ParseAddr { .. }
1262            | Error::UnsupportedSelectorType { .. }
1263            | Error::InvalidArguments { .. }
1264            | Error::ManualGcRejectedByMaintenanceMode { .. }
1265            | Error::ProcedureNotFound { .. }
1266            | Error::TooManyPartitions { .. }
1267            | Error::TomlFormat { .. }
1268            | Error::HandlerNotFound { .. }
1269            | Error::LeaderPeerChanged { .. }
1270            | Error::RepartitionSourceRegionMissing { .. }
1271            | Error::RepartitionTargetRegionMissing { .. }
1272            | Error::PartitionExprMismatch { .. }
1273            | Error::RepartitionSourceExprMismatch { .. }
1274            | Error::EmptyPartitionExpr { .. } => StatusCode::InvalidArguments,
1275            Error::LeaseKeyFromUtf8 { .. }
1276            | Error::LeaseValueFromUtf8 { .. }
1277            | Error::InvalidRegionKeyFromUtf8 { .. }
1278            | Error::TableRouteNotFound { .. }
1279            | Error::TableInfoNotFound { .. }
1280            | Error::DatanodeTableNotFound { .. }
1281            | Error::InvalidUtf8Value { .. }
1282            | Error::UnexpectedInstructionReply { .. }
1283            | Error::Unexpected { .. }
1284            | Error::RegionOperatingRace { .. }
1285            | Error::RegionRouteNotFound { .. }
1286            | Error::MigrationAbort { .. }
1287            | Error::MigrationRunning { .. }
1288            | Error::RegionMigrated { .. }
1289            | Error::DrainDatanodeRunning { .. }
1290            | Error::DrainDatanodeAbort { .. }
1291            | Error::DrainDatanodeRescanExceeded { .. } => StatusCode::Unexpected,
1292            Error::TableNotFound { .. } => StatusCode::TableNotFound,
1293            Error::SaveClusterInfo { source, .. }
1294            | Error::InvalidClusterInfoFormat { source, .. }
1295            | Error::InvalidDatanodeStatFormat { source, .. }
1296            | Error::InvalidNodeInfoFormat { source, .. } => source.status_code(),
1297            Error::InvalidateTableCache { source, .. } => source.status_code(),
1298            Error::SubmitProcedure { source, .. }
1299            | Error::WaitProcedure { source, .. }
1300            | Error::QueryProcedure { source, .. } => source.status_code(),
1301            Error::ShutdownServer { source, .. } | Error::StartHttp { source, .. } => {
1302                source.status_code()
1303            }
1304            Error::StartProcedureManager { source, .. }
1305            | Error::StopProcedureManager { source, .. } => source.status_code(),
1306
1307            Error::ListCatalogs { source, .. }
1308            | Error::ListSchemas { source, .. }
1309            | Error::ListTables { source, .. } => source.status_code(),
1310            Error::StartTelemetryTask { source, .. } => source.status_code(),
1311
1312            Error::NextSequence { source, .. }
1313            | Error::SetNextSequence { source, .. }
1314            | Error::PeekSequence { source, .. } => source.status_code(),
1315            Error::DowngradeLeader { source, .. } => source.status_code(),
1316            Error::RegisterProcedureLoader { source, .. } => source.status_code(),
1317            Error::SubmitDdlTask { source, .. }
1318            | Error::SubmitReconcileProcedure { source, .. } => source.status_code(),
1319            Error::ConvertProtoData { source, .. }
1320            | Error::TableMetadataManager { source, .. }
1321            | Error::RuntimeSwitchManager { source, .. }
1322            | Error::KvBackend { source, .. }
1323            | Error::UnexpectedLogicalRouteTable { source, .. }
1324            | Error::UpdateTopicNameValue { source, .. } => source.status_code(),
1325            Error::ListActiveFrontends { source, .. }
1326            | Error::ListActiveDatanodes { source, .. }
1327            | Error::ListActiveFlownodes { source, .. } => source.status_code(),
1328            Error::NoAvailableFrontend { .. } | Error::RepartitionGcRequired { .. } => {
1329                StatusCode::IllegalState
1330            }
1331
1332            Error::InitMetadata { source, .. }
1333            | Error::InitDdlManager { source, .. }
1334            | Error::InitReconciliationManager { source, .. } => source.status_code(),
1335
1336            Error::BuildTlsOptions { source, .. } => source.status_code(),
1337            Error::Other { source, .. } => source.status_code(),
1338            Error::RepartitionCreateSubtasks { source, .. } => source.status_code(),
1339            Error::RepartitionGcRequirement { source, .. } => source.status_code(),
1340            Error::InspectRepartitionProcedures { source, .. } => source.status_code(),
1341            Error::RepartitionSubprocedureStateReceiver { source, .. } => source.status_code(),
1342            Error::AllocateRegions { source, .. } => source.status_code(),
1343            Error::DeallocateRegions { source, .. } => source.status_code(),
1344            Error::AllocateRegionRoutes { source, .. } => source.status_code(),
1345            Error::AllocateWalOptions { source, .. } => source.status_code(),
1346            Error::BuildCreateRequest { source, .. } => source.status_code(),
1347            Error::NoEnoughAvailableNode { .. } => StatusCode::RuntimeResourcesExhausted,
1348
1349            #[cfg(feature = "pg_kvbackend")]
1350            Error::CreatePostgresPool { .. }
1351            | Error::GetPostgresClient { .. }
1352            | Error::GetPostgresConnection { .. }
1353            | Error::PostgresExecution { .. } => StatusCode::Internal,
1354            #[cfg(feature = "mysql_kvbackend")]
1355            Error::MySqlExecution { .. }
1356            | Error::CreateMySqlPool { .. }
1357            | Error::ParseMySqlUrl { .. }
1358            | Error::DecodeSqlValue { .. }
1359            | Error::AcquireMySqlClient { .. } => StatusCode::Internal,
1360            #[cfg(any(feature = "pg_kvbackend", feature = "mysql_kvbackend"))]
1361            Error::SqlExecutionTimeout { .. } => StatusCode::Internal,
1362        }
1363    }
1364
1365    fn as_any(&self) -> &dyn std::any::Any {
1366        self
1367    }
1368
1369    fn retry_hint(&self) -> RetryHint {
1370        match self {
1371            Error::RetryLater { .. }
1372            | Error::RetryLaterWithSource { .. }
1373            | Error::MailboxTimeout { .. }
1374            | Error::NoEnoughAvailableNode { .. }
1375            | Error::NoLeader { .. }
1376            | Error::LeaderLeaseExpired { .. }
1377            | Error::LeaderLeaseChanged { .. }
1378            | Error::PeerUnavailable { .. } => RetryHint::Retryable,
1379
1380            Error::ConnectEtcd { error, .. } | Error::EtcdFailed { error, .. } => {
1381                common_meta::error::retry_hint_from_etcd_error(error)
1382            }
1383
1384            #[cfg(feature = "pg_kvbackend")]
1385            Error::PostgresExecution { error, .. } => {
1386                common_meta::error::retry_hint_from_postgres_error(error)
1387            }
1388            #[cfg(feature = "pg_kvbackend")]
1389            Error::GetPostgresClient { error, .. } => {
1390                common_meta::error::retry_hint_from_postgres_pool_error(error)
1391            }
1392            #[cfg(feature = "mysql_kvbackend")]
1393            Error::MySqlExecution { error, .. }
1394            | Error::CreateMySqlPool { error, .. }
1395            | Error::AcquireMySqlClient { error, .. } => {
1396                common_meta::error::retry_hint_from_sqlx_error(error)
1397            }
1398            #[cfg(any(feature = "pg_kvbackend", feature = "mysql_kvbackend"))]
1399            Error::SqlExecutionTimeout { .. } => RetryHint::Retryable,
1400
1401            Error::ListActiveFrontends { source, .. }
1402            | Error::ListActiveDatanodes { source, .. }
1403            | Error::ListActiveFlownodes { source, .. }
1404            | Error::InitDdlManager { source, .. }
1405            | Error::InitReconciliationManager { source, .. }
1406            | Error::InitMetadata { source, .. }
1407            | Error::NextSequence { source, .. }
1408            | Error::SetNextSequence { source, .. }
1409            | Error::PeekSequence { source, .. }
1410            | Error::SubmitDdlTask { source, .. }
1411            | Error::SubmitReconcileProcedure { source, .. }
1412            | Error::InvalidateTableCache { source, .. }
1413            | Error::ConvertProtoData { source, .. }
1414            | Error::TableMetadataManager { source, .. }
1415            | Error::RuntimeSwitchManager { source, .. }
1416            | Error::KvBackend { source, .. }
1417            | Error::UnexpectedLogicalRouteTable { source, .. }
1418            | Error::SaveClusterInfo { source, .. }
1419            | Error::InvalidClusterInfoFormat { source, .. }
1420            | Error::InvalidDatanodeStatFormat { source, .. }
1421            | Error::InvalidNodeInfoFormat { source, .. }
1422            | Error::FlowStateHandler { source, .. }
1423            | Error::BuildWalProvider { source, .. }
1424            | Error::BuildKafkaClient { error: source, .. }
1425            | Error::UpdateTopicNameValue { source, .. }
1426            | Error::BuildTlsOptions { source, .. }
1427            | Error::AllocateRegions { source, .. }
1428            | Error::DeallocateRegions { source, .. }
1429            | Error::BuildCreateRequest { source, .. }
1430            | Error::AllocateRegionRoutes { source, .. }
1431            | Error::AllocateWalOptions { source, .. }
1432            | Error::RepartitionGcRequirement { source, .. } => source.retry_hint(),
1433
1434            Error::Other { source, .. }
1435            | Error::ListCatalogs { source, .. }
1436            | Error::ListSchemas { source, .. }
1437            | Error::ListTables { source, .. }
1438            | Error::DowngradeLeader { source, .. } => source.retry_hint(),
1439
1440            Error::SubmitProcedure { source, .. }
1441            | Error::WaitProcedure { source, .. }
1442            | Error::QueryProcedure { source, .. }
1443            | Error::StartProcedureManager { source, .. }
1444            | Error::StopProcedureManager { source, .. }
1445            | Error::RegisterProcedureLoader { source, .. }
1446            | Error::InspectRepartitionProcedures { source, .. }
1447            | Error::RepartitionSubprocedureStateReceiver { source, .. } => source.retry_hint(),
1448
1449            Error::ShutdownServer { source, .. } | Error::StartHttp { source, .. } => {
1450                source.retry_hint()
1451            }
1452            Error::StartTelemetryTask { source, .. } => source.retry_hint(),
1453            Error::CreateChannel { source, .. } => source.retry_hint(),
1454            Error::RepartitionCreateSubtasks { source, .. } => source.retry_hint(),
1455            Error::SerializePartitionExpr { source, .. }
1456            | Error::DeserializePartitionExpr { source, .. } => source.retry_hint(),
1457
1458            Error::DeleteRecords { error, .. }
1459            | Error::BuildPartitionClient { error, .. }
1460            | Error::GetOffset { error, .. } => rskafka_client_error_to_retry_hint(error),
1461
1462            Error::PusherNotFound { .. }
1463            | Error::PushMessage { .. }
1464            | Error::ExceededDeadline { .. } => RetryHint::NonRetryable,
1465
1466            _ => RetryHint::NonRetryable,
1467        }
1468    }
1469}
1470
1471// for form tonic
1472pub(crate) fn match_for_io_error(err_status: &tonic::Status) -> Option<&std::io::Error> {
1473    let mut err: &(dyn std::error::Error + 'static) = err_status;
1474
1475    loop {
1476        if let Some(io_err) = err.downcast_ref::<std::io::Error>() {
1477            return Some(io_err);
1478        }
1479
1480        // h2::Error do not expose std::io::Error with `source()`
1481        // https://github.com/hyperium/h2/pull/462
1482        if let Some(h2_err) = err.downcast_ref::<h2::Error>()
1483            && let Some(io_err) = h2_err.get_io()
1484        {
1485            return Some(io_err);
1486        }
1487
1488        err = err.source()?;
1489    }
1490}
1491
1492#[cfg(test)]
1493mod tests {
1494    use std::time::Duration;
1495
1496    use common_error::ext::ErrorExt;
1497    use common_error::mock::MockError;
1498    use common_error::status_code::StatusCode;
1499    use rskafka::BackoffError;
1500    use rskafka::client::error::Error as KafkaClientError;
1501    use snafu::ResultExt;
1502
1503    use super::{
1504        BuildPartitionClientSnafu, DeallocateRegionsSnafu, DeleteRecordsSnafu, GetOffsetSnafu,
1505    };
1506
1507    fn retry_failed_kafka_error() -> KafkaClientError {
1508        KafkaClientError::RetryFailed(BackoffError::DeadlineExceded {
1509            deadline: Duration::from_secs(1),
1510            source: Box::new(std::io::Error::other("retry failed")),
1511        })
1512    }
1513
1514    #[test]
1515    fn test_deallocate_regions_is_retryable_when_source_is_retry_later() {
1516        let source = common_meta::error::Error::retry_later(MockError::new(StatusCode::Internal));
1517        let err = Err::<(), _>(source)
1518            .context(DeallocateRegionsSnafu { table_id: 1024_u32 })
1519            .unwrap_err();
1520
1521        assert!(err.is_retryable());
1522        assert!(err.retry_hint().is_retryable());
1523    }
1524
1525    #[test]
1526    fn test_deallocate_regions_is_not_retryable_when_source_is_not_retry_later() {
1527        let source = common_meta::error::UnexpectedSnafu {
1528            err_msg: "mock error",
1529        }
1530        .build();
1531        let err = Err::<(), _>(source)
1532            .context(DeallocateRegionsSnafu { table_id: 1024_u32 })
1533            .unwrap_err();
1534
1535        assert!(!err.is_retryable());
1536        assert!(!err.retry_hint().is_retryable());
1537    }
1538
1539    #[test]
1540    fn test_kafka_retry_failed_errors_are_retryable() {
1541        let delete_records_err = Err::<(), _>(retry_failed_kafka_error())
1542            .context(DeleteRecordsSnafu {
1543                topic: "test_topic",
1544                partition: 0,
1545                offset: 1024u64,
1546            })
1547            .unwrap_err();
1548        let build_partition_client_err = Err::<(), _>(retry_failed_kafka_error())
1549            .context(BuildPartitionClientSnafu {
1550                topic: "test_topic",
1551                partition: 0,
1552            })
1553            .unwrap_err();
1554        let get_offset_err = Err::<(), _>(retry_failed_kafka_error())
1555            .context(GetOffsetSnafu {
1556                topic: "test_topic",
1557            })
1558            .unwrap_err();
1559
1560        assert!(delete_records_err.is_retryable());
1561        assert!(build_partition_client_err.is_retryable());
1562        assert!(get_offset_err.is_retryable());
1563        assert!(delete_records_err.retry_hint().is_retryable());
1564        assert!(build_partition_client_err.retry_hint().is_retryable());
1565        assert!(get_offset_err.retry_hint().is_retryable());
1566    }
1567
1568    #[test]
1569    fn test_kafka_non_retry_failed_errors_are_not_retryable() {
1570        let delete_records_err = Err::<(), _>(KafkaClientError::InvalidResponse("invalid".into()))
1571            .context(DeleteRecordsSnafu {
1572                topic: "test_topic",
1573                partition: 0,
1574                offset: 1024u64,
1575            })
1576            .unwrap_err();
1577        let build_partition_client_err =
1578            Err::<(), _>(KafkaClientError::InvalidResponse("invalid".into()))
1579                .context(BuildPartitionClientSnafu {
1580                    topic: "test_topic",
1581                    partition: 0,
1582                })
1583                .unwrap_err();
1584        let get_offset_err = Err::<(), _>(KafkaClientError::InvalidResponse("invalid".into()))
1585            .context(GetOffsetSnafu {
1586                topic: "test_topic",
1587            })
1588            .unwrap_err();
1589
1590        assert!(!delete_records_err.is_retryable());
1591        assert!(!build_partition_client_err.is_retryable());
1592        assert!(!get_offset_err.is_retryable());
1593        assert!(!delete_records_err.retry_hint().is_retryable());
1594        assert!(!build_partition_client_err.retry_hint().is_retryable());
1595        assert!(!get_offset_err.retry_hint().is_retryable());
1596    }
1597}