# S3 Pulse Source Kafka Connect Worker

This project builds a Docker image for a Kafka Connect worker pre-installed with the [Streamthoughts Kafka Connect File Pulse](https://github.com/streamthoughts/kafka-connect-file-pulse) plugin (v2.16.0). It is specifically configured to ingest files from an Amazon S3 bucket into a Kafka topic.

## Overview

The worker uses a custom `entry_point.sh` to programmatically generate the connector configuration based on environment variables and submits it to the Kafka Connect REST API upon startup. It also includes logic to monitor the connector status and restart it if necessary.

## Configuration

The behavior of the connector and the worker can be customized using the following environment variables.

### Required Variables

These variables must be provided at runtime for the connector to function correctly.

| Variable | Description |
| :--- | :--- |
| `KAFKA_S3_BUCKET` | The name of the S3 bucket to read files from. |
| `KAFKA_TOPIC` | The destination Kafka topic where records will be produced. |
| `CONNECT_BOOTSTRAP_SERVERS` | Comma-separated list of Kafka bootstrap servers. |
| `CONNECT_GROUP_UNIQUE_IDENTIFIER` | A unique identifier for this Connect cluster group. Used to generate internal topic names (offsets, configs, status). |
| `CONNECT_REST_PORT` | The port the Kafka Connect REST API listens on. |

### S3 Source Configuration

| Variable | Default | Description |
| :--- | :--- | :--- |
| `KAFKA_S3_REGION` | `us-east-1` | The AWS region of the S3 bucket. |
| `KAFKA_S3_BUCKET_OBJECT_PREFIX` | `kafka_incoming` | The folder/prefix within the bucket to monitor for new files. |
| `KAFKA_S3_FILE_EXTENSION` | `json` | The file extension to filter files by (regex pattern ends with this). |
| `KAFKA_S3_ACTION_AFTER_READ` | `delete` | Action to take after processing a file. Options: `delete` or `move`. If `move`, files go to `processed/` or `failed/` prefixes. |
| `KAFKA_S3_STATUS_IDENTIFIER` | `uri+lastModified` | Strategy to uniquely identify files to prevent re-reading. |
| `SKIP_HEADER_ROWS` | `0` | Number of lines to skip at the beginning of the file (useful for CSVs with headers). |

### Message Transformation

| Variable | Default | Description |
| :--- | :--- | :--- |
| `KAFKA_MESSAGE_KEY_FIELDNAME` | *(empty)* | If set, extracts this field from the JSON value and sets it as the Kafka message key. |
| `KAFKA_MESSAGE_RENAME_FROM_FIELDNAME`| *(empty)* | The name of a field in the input JSON to rename. Must be used with `_TO_FIELDNAME`. |
| `KAFKA_MESSAGE_RENAME_TO_FIELDNAME` | *(empty)* | The new name for the field specified in `_FROM_FIELDNAME`. |

### Worker & Runtime Configuration

| Variable | Default | Description |
| :--- | :--- | :--- |
| `CONNECTOR_NAME` | `s3_pulse_source` | The name of the connector instance created in Kafka Connect. |
| `Environment` | *(unset)* | If set to `local`, advertises `localhost`. Otherwise, attempts to fetch the Fargate task IP for `CONNECT_REST_ADVERTISED_HOST_NAME`. |
| `DEBUG_MODE` | `false` | If `true`, prints verbose curl output during connector configuration submission and exits immediately on failure. |
| `CONNECT_KEY_CONVERTER` | `StringConverter` | Class for key converter (default: `org.apache.kafka.connect.storage.StringConverter`). |
| `CONNECT_VALUE_CONVERTER` | `JsonConverter` | Class for value converter (default: `org.apache.kafka.connect.json.JsonConverter`). |
| `KAFKA_PRODUCER_MAX_REQUEST_SIZE` | `15728640` | Max request size for the Kafka producer. |

## Usage

### Building the Image

```bash
docker build -t s3-pulse-source .
```

### Running Locally

Ensure you have a `.env` file or pass environment variables to the container.

```bash
docker run -d \
  -e KAFKA_S3_BUCKET=my-data-bucket \
  -e KAFKA_TOPIC=my-output-topic \
  -e CONNECT_BOOTSTRAP_SERVERS=kafka:9092 \
  -e CONNECT_GROUP_UNIQUE_IDENTIFIER=dev_cluster \
  -e CONNECT_REST_PORT=8083 \
  -e Environment=local \
  -p 8083:8083 \
  s3-pulse-source
```

## Features

-   **Automatic Configuration**: The container waits for the worker to start and automatically posts the connector configuration defined in `config.template.json`.
-   **Health Check & Repair**: The entry point checks if the connector is running and attempts to restart it via the REST API if it fails or is not found.
-   **S3 Cleanup**: Supports deleting files or moving them to `processed`/`failed` folders after ingestion based on `KAFKA_S3_ACTION_AFTER_READ`.