feat(sink): add AWS Kinesis data stream sink

## Summary
- Add `KinesisSink` that publishes events to AWS Kinesis streams
- Sends `event.payload` as JSON data blob
- Supports dynamic stream name and partition key via event metadata

## Configuration
```yaml
sink:
  type: kinesis
  region: us-east-1
  stream_name: events  # optional default
```

## Routing
Stream and partition key are resolved from:
1. `stream` and `partition_key` fields in event metadata
2. Fallback to config values

Author: psteinroe

Cargo.lock

@@ -6,6 +6,7 @@ version = "0.1.0"
 [features]
 default            = []
 sink-kafka         = ["dep:rdkafka"]
+sink-kinesis       = ["dep:aws-sdk-kinesis", "dep:aws-config"]
 sink-nats          = ["dep:async-nats"]
 sink-rabbitmq      = ["dep:lapin"]
 sink-redis-streams = ["dep:redis"]
@@ -57,6 +58,7 @@ uuid         = { version = "1.19.0", default-features = false, features = ["v4"]
 # Optional sink dependencies.
 async-nats = { version = "0.38", optional = true }
 aws-config = { version = "1.5", optional = true }
+aws-sdk-kinesis = { version = "1.60", optional = true }
 aws-sdk-sns = { version = "1.60", optional = true }
 aws-sdk-sqs = { version = "1.60", optional = true }
 lapin = { version = "2.5", optional = true }

Cargo.toml

@@ -24,6 +24,9 @@ use crate::sink::sqs::SqsSinkConfig;
 #[cfg(feature = "sink-sns")]
 use crate::sink::sns::SnsSinkConfig;
 
+#[cfg(feature = "sink-kinesis")]
+use crate::sink::kinesis::KinesisSinkConfig;
+
 /// Sink destination configuration.
 ///
 /// Determines where replicated events are sent.
@@ -72,4 +75,9 @@ pub enum SinkConfig {
     #[cfg(feature = "sink-sns")]
     #[serde(rename = "sns")]
     Sns(SnsSinkConfig),
+
+    /// AWS Kinesis sink for data stream publishing.
+    #[cfg(feature = "sink-kinesis")]
+    #[serde(rename = "kinesis")]
+    Kinesis(KinesisSinkConfig),
 }

src/config/sink.rs

@@ -181,6 +181,19 @@ async fn run_pipeline(config: &PipelineConfig) -> EtlResult<()> {
             })?;
             AnySink::Sns(s)
         }
+
+        #[cfg(feature = "sink-kinesis")]
+        SinkConfig::Kinesis(cfg) => {
+            use crate::sink::kinesis::KinesisSink;
+            let s = KinesisSink::new(cfg.clone()).await.map_err(|e| {
+                etl::etl_error!(
+                    etl::error::ErrorKind::InvalidData,
+                    "Failed to create Kinesis sink",
+                    e.to_string()
+                )
+            })?;
+            AnySink::Kinesis(s)
+        }
     };
 
     // Create PgStream as an ETL destination
@@ -266,6 +279,11 @@ fn log_sink_config(config: &SinkConfig) {
         SinkConfig::Sns(_cfg) => {
             debug!("using sns sink");
         }
+
+        #[cfg(feature = "sink-kinesis")]
+        SinkConfig::Kinesis(_cfg) => {
+            debug!("using kinesis sink");
+        }
     }
 }