Adds S3 source connector (#659)

* Adds S3 source connector

This commit introduces a new S3 source connector that monitors an S3 bucket for new files and streams their content to a Kafka topic.

The connector supports:
- Configuration via environment variables
- File monitoring with configurable polling interval
- Optional content download and chunking for large files
- Metadata extraction for file information

This replaces the previous S3 source implementation with a custom file watcher for more flexibility and control over the data ingestion process.

* Updates S3 source configuration

Updates the S3 source configuration to remove the app.yaml file and consolidate settings into the library.json.

Configures the deployment type to 'Service' and increases the CPU and Memory allocation.
Adds 'MAX_MB_PER_MESSAGE' and 'DOWNLOAD_CONTENT' environment variables.

* Solve 3 comments

Refactors S3 source configuration by renaming `S3_FOLDER_PATH` to `S3_FOLDER_PREFIX` and making it optional.

Removes the required `S3_FILE_FORMAT` and `S3_FILE_COMPRESSION` parameters.

Updates default value for `MAX_MB_PER_MESSAGE` and `DOWNLOAD_CONTENT`.

Adds `POLL_INTERVAL_SECONDS` for configuring poll frequency.

* Adds support for custom S3 endpoints

Enables the S3 source to connect to S3-compatible storage
like MinIO by introducing the AWS_ENDPOINT_URL option.

This allows users to specify a custom endpoint for the S3 client,
making the source more flexible and adaptable to various
deployment scenarios.

Also updates tests to reflect that the service now consumes json files.

---------

Co-authored-by: Tomas Neubauer <tomas@quix.ai>

Author: JotaBlanco

python/sources/s3_source/README.md

@@ -1,37 +1,132 @@
-# S3 Source Connector
+# S3 Configuration File Source
 
-[This connector](https://github.com/quixio/quix-samples/tree/main/python/sources/s3_source) demonstrates how to connect to Amazon S3 to read files into a Kafka topic.
+A Quix Streams source service that monitors Amazon S3 buckets for configuration files and streams their content to Kafka topics for processing.
 
-## How to run
+## Overview
 
-Create a [Quix](https://portal.cloud.quix.io/signup?utm_campaign=github) account or log-in and visit the `Connectors` tab to use this connector.
+This service continuously monitors a specified S3 bucket and folder path for new configuration files. When files are detected, it downloads and publishes their content to the `s3-data` Kafka topic for downstream processing by the XML-to-JSON transformer.
 
-Clicking `Set up connector` allows you to enter your connection details and runtime parameters.
+## Features
 
-Then either: 
-* click `Test connection & deploy` to deploy the pre-built and configured container into Quix. 
+- **File Monitoring**: Continuous polling of S3 bucket for new files
+- **Compression Support**: Handles gzip-compressed files automatically
+- **Format Support**: Primarily designed for XML configuration files
+- **State Management**: Tracks processed files to avoid duplicates
+- **Error Handling**: Robust error handling for S3 connectivity issues
+- **Configurable Polling**: Adjustable polling interval for monitoring
 
-* or click `Customise connector` to inspect or alter the code before deployment.
+## How it Works
 
-## Environment variables
+1. **Monitoring**: Polls the specified S3 bucket/folder at regular intervals
+2. **File Detection**: Identifies new or modified files since last check
+3. **Download**: Retrieves file content from S3
+4. **Decompression**: Automatically handles gzip decompression if needed
+5. **Publishing**: Sends file content to Kafka topic for processing
+6. **State Tracking**: Records processed files to prevent reprocessing
 
-This connector uses the following environment variables:
+## Environment Variables
 
-- **output**: The output topic to stream Segment data into
-- **S3_BUCKET**: The URI or URL to your S3 bucket
-- **S3_REGION**: The region of your S3 bucket
-- **S3_SECRET**: Your AWS secret
-- **S3_ACCESS_KEY_ID**: Your AWS Access Key
-- **S3_FOLDER_PATH**: The path to the S3 folder to consume
-- **S3_FILE_FORMAT**: The file format of the files
-- **S3_FILE_COMPRESSION**: The type of file compression used for the files
+- **output**: Name of the output Kafka topic (default: `s3-data`)
+- **S3_BUCKET**: S3 bucket URI (e.g., `s3://quix-test-bucket/configurations/`)
+- **S3_REGION**: AWS region of the S3 bucket (e.g., `eu-west-2`)
+- **S3_SECRET**: AWS Secret Access Key (stored as secret)
+- **S3_ACCESS_KEY_ID**: AWS Access Key ID (stored as secret)
+- **S3_FOLDER_PATH**: Folder path within bucket to monitor (e.g., `configurations`)
+- **S3_FILE_FORMAT**: Expected file format (e.g., `xml`)
+- **S3_FILE_COMPRESSION**: Compression type (e.g., `gzip`)
+- **POLL_INTERVAL_SECONDS**: Polling interval in seconds (default: 30)
 
-## Contribute
+## Configuration Example
 
-Submit forked projects to the Quix [GitHub](https://github.com/quixio/quix-samples) repo. Any new project that we accept will be attributed to you and you'll receive $200 in Quix credit.
+Based on the current deployment configuration:
 
-## Open source
+```yaml
+S3_BUCKET: "s3://quix-test-bucket/configurations/"
+S3_REGION: "eu-west-2"
+S3_FOLDER_PATH: "configurations"
+S3_FILE_FORMAT: "xml"
+S3_FILE_COMPRESSION: "gzip"
+POLL_INTERVAL_SECONDS: "30"
+```
 
-This project is open source under the Apache 2.0 license and available in our [GitHub](https://github.com/quixio/quix-samples) repo.
+## Data Flow
 
-Please star us and mention us on social to show your appreciation.
+The service fits into the larger configuration management pipeline:
+
+```
+S3 Bucket → S3 Source → s3-data topic → XML-to-JSON → configurations topic → Configuration Sink → Configuration API
+```
+
+## Development
+
+### Local Development
+
+```bash
+# Install dependencies
+pip install -r requirements.txt
+
+# Set environment variables
+export S3_BUCKET="s3://your-bucket/path/"
+export S3_REGION="your-region"
+export S3_ACCESS_KEY_ID="your-access-key"
+export S3_SECRET="your-secret-key"
+export S3_FOLDER_PATH="configurations"
+export output="s3-data"
+
+# Run the service
+python main.py
+```
+
+### Docker Build
+
+```bash
+docker build -t s3-source .
+```
+
+## File Processing
+
+### Supported File Types
+- **XML files**: Primary configuration format
+- **Gzip compression**: Automatic decompression support
+- **Text-based formats**: Any text-based configuration files
+
+### File Structure
+The service expects files in the following S3 structure:
+```
+s3://bucket-name/configurations/
+├── config1.xml.gz
+├── config2.xml.gz
+└── subfolder/
+    └── config3.xml.gz
+```
+
+## Error Handling
+
+- **S3 Connection Errors**: Automatic retry with exponential backoff
+- **Authentication Failures**: Clear error logging for credential issues
+- **File Access Errors**: Graceful handling of permission or availability issues
+- **Compression Errors**: Error handling for corrupted compressed files
+
+## State Management
+
+The service maintains persistent state to track:
+- Last processed timestamp
+- File checksums/ETags to detect changes
+- Processing status of individual files
+
+State is preserved across service restarts through Quix's state management system.
+
+## Monitoring
+
+The service provides logging for:
+- File discovery and processing events
+- S3 API interactions and errors
+- Kafka message publishing status
+- Performance metrics and polling intervals
+
+## Integration
+
+This service integrates with:
+- **Upstream**: S3 bucket containing configuration files
+- **Downstream**: XML-to-JSON transformation service via `s3-data` topic
+- **Monitoring**: Quix platform logging and metrics systems

python/sources/s3_source/library.json

@@ -55,42 +55,42 @@
       "Required": true
   },
   {
-      "Name": "S3_FOLDER_PATH",
+      "Name": "S3_FOLDER_PREFIX",
       "Type": "EnvironmentVariable",
       "InputType": "FreeText",
-      "Description": "The path to the S3 folder to consume",
+      "Description": "The folder prefix within the S3 bucket to monitor (optional)",
       "DefaultValue": "",
-      "Required": true
+      "Required": false
   },
   {
-      "Name": "S3_FILE_FORMAT",
+      "Name": "POLL_INTERVAL_SECONDS",
       "Type": "EnvironmentVariable",
       "InputType": "FreeText",
-      "Description": "The file format of the files",
-      "DefaultValue": "json",
-      "Required": true
+      "Description": "How often to check S3 for new files (in seconds)",
+      "DefaultValue": "30",
+      "Required": false
   },
   {
-      "Name": "S3_FILE_COMPRESSION",
+      "Name": "MAX_MB_PER_MESSAGE",
       "Type": "EnvironmentVariable",
       "InputType": "FreeText",
-      "Description": "The type of file compression used for the files",
-      "DefaultValue": "gzip",
-      "Required": true
+      "Description": "Max size of the message in MB",
+      "DefaultValue": "1",
+      "Required": false
   },
   {
-      "Name": "S3_ENDPOINT_URL",
+      "Name": "DOWNLOAD_CONTENT",
       "Type": "EnvironmentVariable",
       "InputType": "FreeText",
-      "Description": "Custom S3 endpoint URL for MinIO or other S3-compatible services (optional)",
-      "DefaultValue": "",
+      "Description": "Whether to publish the content of the file to Kafka or just the metadata",
+      "DefaultValue": "True",
       "Required": false
   }
   ],
   "DeploySettings": {
-    "DeploymentType": "Job",
-    "CpuMillicores": 100,
-    "MemoryInMb": 100,
+    "DeploymentType": "Service",
+    "CpuMillicores": 250,
+    "MemoryInMb": 250,
     "Replicas": 1,
     "PublicAccess": false,
     "ValidateConnection": false

python/sources/s3_source/main.py

@@ -1,51 +1,50 @@
-from quixstreams import Application
-from quixstreams.sources.community.file.s3 import S3FileSource
-import time
-
 import os
+import logging
+from quixstreams import Application
+from s3_file_watcher import S3FileWatcher
 from dotenv import load_dotenv
-load_dotenv()
-
-app = Application()
-
-# create an output topic
-output_topic = app.topic(os.environ['output'])
 
-# Custom setters for S3 file records (must be regular functions for pickling)
-def key_setter(row):
-    """Use record ID as the message key, or None if not present."""
-    return str(row.get('id', '')) if 'id' in row else None
-
-def value_setter(row):
-    """Use the whole record as the message value."""
-    return row
-
-def timestamp_setter(row):
-    """Use current time as message timestamp."""
-    return int(time.time() * 1000)
-
-# Build S3FileSource kwargs
-source_kwargs = {
-    'filepath': os.environ['S3_FOLDER_PATH'],
-    'bucket': os.environ['S3_BUCKET'],
-    'aws_access_key_id': os.environ['S3_ACCESS_KEY_ID'],
-    'aws_secret_access_key': os.environ['S3_SECRET'],
-    'region_name': os.environ['S3_REGION'],
-    'file_format': os.environ['S3_FILE_FORMAT'],
-    'compression': os.environ.get('S3_FILE_COMPRESSION'),
-    'key_setter': key_setter,
-    'value_setter': value_setter,
-    'timestamp_setter': timestamp_setter,
-}
-
-# Support custom endpoint for MinIO or other S3-compatible services
-if 'S3_ENDPOINT_URL' in os.environ:
-    source_kwargs['endpoint_url'] = os.environ['S3_ENDPOINT_URL']
-
-# create the S3 file source
-source = S3FileSource(**source_kwargs)
+# Load environment variables
+load_dotenv()
 
-app.add_source(source, topic=output_topic)
+# Configuration
+AWS_ACCESS_KEY_ID = os.getenv("S3_ACCESS_KEY_ID")
+AWS_SECRET_ACCESS_KEY = os.getenv("S3_SECRET")
+S3_BUCKET_NAME = os.environ["S3_BUCKET"]
+S3_FOLDER_PREFIX = os.getenv("S3_FOLDER_PREFIX", "")
+AWS_REGION = os.getenv("S3_REGION", "us-east-1")
+AWS_ENDPOINT_URL = os.getenv("AWS_ENDPOINT_URL")  # For MinIO or custom S3-compatible endpoints
+TOPIC_NAME = os.environ["output"]
+POLL_INTERVAL = int(os.getenv("POLL_INTERVAL_SECONDS", "30"))
+
+# Logging setup
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+# Create Quix Application
+app = Application(consumer_group="s3_file_watcher_v1.2", auto_create_topics=True)
+
+# Create S3 File Watcher Source
+s3_file_watcher = S3FileWatcher(
+    name="s3_file_watcher",
+    bucket_name=S3_BUCKET_NAME,
+    folder_prefix=S3_FOLDER_PREFIX,
+    aws_access_key_id=AWS_ACCESS_KEY_ID,
+    aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
+    region_name=AWS_REGION,
+    endpoint_url=AWS_ENDPOINT_URL,
+    poll_interval=POLL_INTERVAL
+)
+
+# Define the topic using the "output" environment variable
+topic = app.topic(TOPIC_NAME)
+
+# Add source to application
+app.add_source(s3_file_watcher, topic)
 
 if __name__ == "__main__":
-    app.run()
+    try:
+        logging.basicConfig(level=logging.INFO)
+        app.run()
+    except KeyboardInterrupt:
+        print("\nProgram interrupted by user. Exiting gracefully.")

python/sources/s3_source/requirements.txt

@@ -1,2 +1,3 @@
-quixstreams[s3]==3.23.1
+quixstreams==3.16.1
 python-dotenv
+boto3