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