1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
//! The crate-level error type.
use std::error::Error as StdError;
/// Errors returned by the Amazon Kinesis Data Streams broker.
///
/// One enum for the whole crate, variants by source, per the `RustStream` broker conventions.
/// The wrapped sources are boxed `std` errors formatted with their full cause chain, so the
/// public API does not leak the SDK's layered error types.
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum KinesisError {
/// Loading the AWS configuration failed.
#[error("aws config error: {0}")]
Config(String),
/// A stream admin call (describe, create, shard listing) failed.
#[error("kinesis stream error for '{stream}': {source}")]
Stream {
/// The stream the call was about.
stream: String,
/// The SDK's failure, with its cause chain.
#[source]
source: Box<dyn StdError + Send + Sync>,
},
/// Reading a shard failed permanently.
#[error("kinesis read error on '{stream}' shard '{shard}': {source}")]
Read {
/// The stream the shard belongs to.
stream: String,
/// The shard the read targeted.
shard: String,
/// The SDK's failure, with its cause chain.
#[source]
source: Box<dyn StdError + Send + Sync>,
},
/// A delivered record is KPL-aggregated, which this crate does not deaggregate yet;
/// failing loudly beats handing the handler an opaque protobuf blob.
#[error("kinesis record on shard '{shard}' is KPL-aggregated (unsupported)")]
AggregatedRecord {
/// The shard the record arrived on.
shard: String,
},
/// Writing a record failed.
#[error("kinesis publish error to '{stream}': {source}")]
Publish {
/// The stream the record targeted.
stream: String,
/// The SDK's failure, with its cause chain.
#[source]
source: Box<dyn StdError + Send + Sync>,
},
/// The lease store failed.
#[error("kinesis lease store error for stream '{stream}' shard '{shard}': {source}")]
Lease {
/// The stream the shard belongs to.
stream: String,
/// The shard whose lease was involved.
shard: String,
/// The store's failure.
#[source]
source: Box<dyn StdError + Send + Sync>,
},
/// The handle is used before `connect` filled the shared connection, or after `shutdown`.
#[error("kinesis broker is not connected")]
NotConnected,
/// The broker already reads this stream.
///
/// A broker leases and checkpoints a stream's shards under its one owner id, so a second
/// subscription to the stream on the same broker would share those leases with the first and
/// lose records. A service that reads one stream twice reads it through a second
/// `KinesisBroker` with a lease store of its own.
#[error(
"kinesis broker already reads stream '{stream}': a second subscription would share its \
shard leases and checkpoints; read the stream again through a second `KinesisBroker` \
with a lease store of its own"
)]
AlreadySubscribed {
/// The stream the refused subscription named.
stream: String,
},
/// A stream descriptor is invalid.
#[error("invalid kinesis descriptor: {0}")]
Invalid(String),
}
/// Formats an SDK error with its full cause chain and boxes it.
pub(crate) fn sdk_err<E, R>(
err: &aws_sdk_kinesis::error::SdkError<E, R>,
) -> Box<dyn StdError + Send + Sync>
where
E: StdError + Send + Sync + 'static,
R: std::fmt::Debug + Send + Sync + 'static,
{
Box::from(aws_sdk_kinesis::error::DisplayErrorContext(err).to_string())
}