1use 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 #[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 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
1458pub(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 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}