feat: Add Redis queue support for S3 event notifications (#1189)

* feat: Add Redis queue support for S3 event notifications

- Add Redis client dependency to Cargo.toml
- Extend Amazon S3 source to support Redis pub/sub for event notifications
- Add redis_url and redis_channel configuration options
- Implement RedisContext with async pub/sub support
- Update change_stream to prefer Redis over SQS when both are configured
- Add comprehensive documentation for MinIO Redis setup
- Update Python spec to include Redis configuration fields

This enables MinIO users to receive S3-compatible event notifications
without requiring AWS SQS, addressing issue #599.

* refactor: address maintainer feedback for Redis configuration

- Remove timeout logic in Redis polling to prevent error masking
- Wrap Redis URL and channel in dedicated RedisConfig/RedisNotification structs
- Improve code organization and follow project patterns

Author: belloibrahv

Cargo.lock

@@ -152,5 +152,6 @@ azure_storage_blobs = { version = "0.21.0", default-features = false, features =
     "hmac_rust",
 ] }
 serde_path_to_error = "0.1.17"
+redis = { version = "0.31.0", features = ["tokio-comp", "connection-manager"] }
 expect-test = "1.5.0"
 encoding_rs = "0.8.35"

Cargo.toml

@@ -86,6 +86,33 @@ This is how to setup:
 
 AWS's [Guide of Configuring a Bucket for Notifications](https://docs.aws.amazon.com/AmazonS3/latest/userguide/ways-to-add-notification-config-to-bucket.html#step1-create-sqs-queue-for-notification) provides more details.
 
+#### (Alternative) Setup Redis for event notifications (MinIO)
+
+For MinIO setups that don't use AWS SQS, you can configure MinIO to publish event notifications to Redis:
+
+*   Configure MinIO to publish events to Redis by setting environment variables:
+    ```bash
+    export MINIO_NOTIFY_REDIS_ENABLE="on"
+    export MINIO_NOTIFY_REDIS_ADDRESS="redis-endpoint.example.net:6379"
+    export MINIO_NOTIFY_REDIS_KEY="bucketevents"
+    export MINIO_NOTIFY_REDIS_FORMAT="namespace"
+    ```
+    Replace the values with your Redis server details.
+
+*   Alternatively, use the `mc` command-line tool:
+    ```bash
+    mc alias set myminio http://minio.example.com:9000 ACCESSKEY SECRETKEY
+    mc admin config set myminio/ notify_redis \
+      address="redis-endpoint.example.net:6379" \
+      key="bucketevents" \
+      format="namespace"
+    mc admin service restart myminio
+    ```
+
+*   Ensure your Redis server is accessible and configured to accept connections from MinIO.
+
+MinIO's [Redis Notification Settings](https://min.io/docs/minio/linux/reference/minio-server/settings/notifications/redis.html) documentation provides more details on configuration options.
+
 ### Spec
 
 The spec takes the following fields:
@@ -113,6 +140,17 @@ The spec takes the following fields:
 
     :::
 
+*   `redis_url` (`str`, optional): if provided, the source will receive change event notifications via Redis pub/sub. This is particularly useful for MinIO setups that publish events to Redis instead of SQS.
+
+*   `redis_channel` (`str`, optional): the Redis channel to subscribe to for event notifications. Required when `redis_url` is provided.
+
+    :::info
+
+    Redis pub/sub is preferred over SQS when both are configured. This allows MinIO users to receive S3-compatible event notifications without requiring AWS SQS.
+    The Redis implementation expects S3 event notifications in the same JSON format as SQS messages.
+
+    :::
+
 ### Schema
 
 The output is a [*KTable*](/docs/core/data_types#ktable) with the following sub fields:

docs/docs/sources/amazons3.md

@@ -35,6 +35,16 @@ class GoogleDrive(op.SourceSpec):
     recent_changes_poll_interval: datetime.timedelta | None = None
 
 
+@dataclass
+class RedisNotification:
+    """Redis pub/sub configuration for event notifications."""
+
+    # Redis server URL (e.g., "redis://localhost:6379")
+    redis_url: str
+    # Redis channel name for pub/sub notifications
+    redis_channel: str
+
+
 class AmazonS3(op.SourceSpec):
     """Import data from an Amazon S3 bucket. Supports optional prefix and file filtering by glob patterns."""
 
@@ -46,6 +56,7 @@ class AmazonS3(op.SourceSpec):
     included_patterns: list[str] | None = None
     excluded_patterns: list[str] | None = None
     sqs_queue_url: str | None = None
+    redis: RedisNotification | None = None
 
 
 class AzureBlob(op.SourceSpec):