kestra-io / kestra-io/plugin-scylladb
Implement ScyllaDB plugin: Query, Queries, Execute, Batch tasks and polling Trigger
- Dominant language
- Java
- Stars
- 0
- Forks
- 0
- Avg merge
- 5d 9h
- Merged PRs (30d)
- 3
Description
## Summary
Implement the core tasks and trigger for the ScyllaDB plugin, enabling Kestra users to run CQL (Cassandra Query Language) statements against a ScyllaDB cluster directly from a flow. The plugin covers SELECT queries with configurable fetch semantics, DML statement execution, batch operations, and a polling trigger — removing the need to shell out to cqlsh or maintain custom Python tasks for ScyllaDB interactions.
## Motivation
ScyllaDB is a high-performance, Cassandra-compatible NoSQL database widely adopted for real-time analytics, IoT data pipelines, and event-driven architectures. Without a dedicated plugin, Kestra users rely on generic Script tasks or HTTP tasks to interact with ScyllaDB, losing type safety, structured outputs, and built-in secret masking. A native plugin enables:
- Data engineering teams to build ingestion and query pipelines without leaving the Kestra ecosystem.
- Platform engineers to automate schema migrations, maintenance jobs, and health checks.
- Seamless composition with existing plugins (e.g. `plugin-jdbc`, `plugin-mongodb`) in mixed-database workflows.
## Context
The repository `kestra-io/plugin-scylladb` already exists and is scaffolded, but currently contains only placeholder classes. This issue tracks the full implementation of the plugin's first stable feature set.
Reference implementations to draw from:
- **plugin-mongodb** (`kestra-io/plugin-mongodb`) — flat package structure, FetchType on query tasks, connection abstraction pattern.
- **plugin-jdbc** (`kestra-io/plugin-jdbc`) — `Query` / `Queries` / `Batch` naming, trigger pattern with row-advancing semantics.
## API Reference
- **Official docs**: https://java-driver.docs.scylladb.com/stable/
- **Authentication**: Plain-text username/password via `PlainTextAuthProvider`; certificate-based TLS also supported.
- **Base URL pattern**: Native CQL protocol on port `9042` (not HTTP); contact points as `host:port` pairs.
- **SDK / client library**: ScyllaDB Java driver — `com.scylladb:java-driver-core:4.19.0.4`
## Gradle Dependencies
Add to `build.gradle`:
```groovy
// ScyllaDB Java driver (ScyllaDB fork of the DataStax OSS Cassandra driver)
implementation "com.scylladb:java-driver-core:4.19.0.4"
```
> Use the latest stable version available on Maven Central.
>
> **Kestra framework inclusions (do not add as dependencies):**
> Jackson serializers are provided by the Kestra platform — do not add `com.fasterxml.jackson.*`.
> The ScyllaDB driver uses its own native CQL protocol transport; no OkHttp or Apache HttpClient is needed.
## Plugin Structure
- **Repository**: `plugin-scylladb`
- **Namespace**: `io.kestra.plugin.scylladb`
- **Sub-plugins**: none (flat package — all tasks in the root `scylladb` package)
- **Categories**: `DATA`
## Suggested Tasks
1. **`ScyllaDbConnection`** — connection configuration POJO holding contact points, local datacenter, keyspace, credentials, and TLS options; exposes a `CqlSession connect(RunContext)` factory method shared by all tasks.
2. **`AbstractScyllaDbTask`** — abstract base task holding `ScyllaDbConnection connection` and the five mandatory Lombok annotations; used by all concrete tasks.
3. **`Query`** — execute a SELECT CQL statement; support `fetchType` (`FETCH_ONE`, `FETCH`, `STORE`) and return structured row outputs or an internal storage URI for large result sets.
4. **`Queries`** — execute a list of SELECT CQL statements sequentially; aggregate outputs per statement.
5. **`Execute`** — execute a non-SELECT CQL statement (INSERT / UPDATE / DELETE / DDL); return affected rows count where available.
6. **`Batch`** — execute a `BatchStatement` of multiple CQL statements (`LOGGED`, `UNLOGGED`, or `COUNTER` batch type).
7. **`Trigger`** — polling trigger; runs a configurable SELECT CQL query on a fixed interval; fires an execution when rows are returned; must advance state (e.g. delete or mark processed rows) to avoid infinite re-triggering.
8. Write unit + integration tests using Testcontainers with the official `scylladb/scylla` Docker image.
9. Add `package-info.java` with `@PluginSubGroup(category = PluginSubGroup.PluginCategory.DATA)`.
10. Add `metadata/index.yaml` and update `AGENTS.md` with the implemented classes.
## YAML Examples
### Example 1 — Fetch rows from a ScyllaDB table and store the result
```yaml
id: scylladb_query
namespace: company.team
inputs:
- id: keyspace
type: STRING
defaults: my_keyspace
tasks:
- id: query
type: io.kestra.plugin.scylladb.Query
connection:
contactPoints:
- "{{ secret('SCYLLADB_HOST') }}:9042"
localDatacenter: datacenter1
keyspace: "{{ inputs.keyspace }}"
username: "{{ secret('SCYLLADB_USERNAME') }}"
password: "{{ secret('SCYLLADB_PASSWORD') }}"
cql: |
SELECT sensor_id, ts, temperature
FROM measurements
WHERE date = '2024-01-15'
ALLOW FILTERING
fetchType: STORE
- id: log_count
type: io.kestra.plugin.core.log.Log
message: "Stored {{ outputs.query.size }} rows at {{ outputs.query.uri }}"
```
### Example 2 — Execute a DML statement (INSERT / UPDATE / DELETE)
```yaml
id: scylladb_execute
namespace: company.team
inputs:
- id: sensor_id
type: STRING
- id: temperature
type: FLOAT
tasks:
- id: insert_measurement
type: io.kestra.plugin.scylladb.Execute
connection:
contactPoints:
- "{{ secret('SCYLLADB_HOST') }}:9042"
localDatacenter: datacenter1
keyspace: iot
username: "{{ secret('SCYLLADB_USERNAME') }}"
password: "{{ secret('SCYLLADB_PASSWORD') }}"
cql: |
INSERT INTO measurements (sensor_id, ts, temperature)
VALUES ('{{ inputs.sensor_id }}', toTimestamp(now()), {{ inputs.temperature }})
```
### Example 3 — Trigger a flow when new rows appear in a ScyllaDB table
```yaml
id: scylladb_trigger
namespace: company.team
triggers:
- id: on_new_events
type: io.kestra.plugin.scylladb.Trigger
interval: PT1M
connection:
contactPoints:
- "{{ secret('SCYLLADB_HOST') }}:9042"
localDatacenter: datacenter1
keyspace: events
username: "{{ secret('SCYLLADB_USERNAME') }}"
password: "{{ secret('SCYLLADB_PASSWORD') }}"
cql: |
SELECT event_id, event_type, payload
FROM pending_events
WHERE processed = false
ALLOW FILTERING
fetchType: FETCH
tasks:
- id: handle_events
type: io.kestra.plugin.core.log.Log
message: "Processing {{ trigger.size }} new events"
```
## Acceptance Criteria
### Functional
- [ ] `ScyllaDbConnection` abstracts `CqlSession` creation and is shared across all tasks
- [ ] `Query` supports `FETCH_ONE`, `FETCH`, and `STORE` fetch types
- [ ] `Queries` executes multiple CQL SELECT statements and returns per-statement outputs
- [ ] `Execute` handles INSERT / UPDATE / DELETE / DDL statements
- [ ] `Batch` supports `LOGGED`, `UNLOGGED`, and `COUNTER` batch types
- [ ] `Trigger` fires only on non-empty result sets and advances state to avoid re-triggering
- [ ] Unit + integration tests pass using Testcontainers (`scylladb/scylla` image) — `./gradlew test`
- [ ] Build passes — `./gradlew build`
### Kestra Plugin Coding Standards
- [ ] HTTP calls use Kestra's internal HTTP client (`io.kestra.core.http.client`) — N/A here; native CQL driver used instead (no HTTP transport to add)
- [ ] All new properties use `Property` — no legacy `@PluginProperty(dynamic = true)` on new code
- [ ] Secret/credential properties (`username`, `password`) annotated with `@PluginProperty(secret = true)`
- [ ] Every property and output has a `@Schema` annotation
- [ ] Task classes carry the five mandatory Lombok annotations (`@SuperBuilder`, `@ToString`, `@EqualsAndHashCode`, `@Getter`, `@NoArgsConstructor`)
- [ ] Logging via `runContext.logger()` only
- [ ] JSON serialization uses Jackson mappers from `io.kestra.core.serializers`
- [ ] All `Property` fields support Kestra expression language (template rendering)
### Documentation & Structure
- [ ] `@Plugin(examples = ...)` entries each set `full = true` with a complete runnable flow (id + namespace + tasks/triggers)
- [ ] Sensitive values in examples use `{{ secret('SECRET_NAME') }}`
- [ ] `package-info.java` with `@PluginSubGroup(category = PluginSubGroup.PluginCategory.DATA)`
- [ ] `metadata/index.yaml` and plugin icon SVG present
- [ ] `AGENTS.md` updated to list all implemented classes
## Repository Setup Checklist
### 1. Add to Sanity check page
Add this plugin to the [Sanity check Notion page](https://www.notion.so/kestra-io/32736907f7b580cbb00dc7c061e624b1?v=32736907f7b58002ac2b000ccc63d8a2).
### 2. Run scoped Terraform apply
Run the following from `infra/terraform/github`:
```bash
terraform apply \
-target='github_repository.repo["plugin-scylladb"]' \
-target='github_issue_labels.plugins["plugin-scylladb"]' \
-target='github_repository_ruleset.branch["plugin-scylladb"]'
```
Contributor guide
No contributing guide indexed for this repository
Research direction
Start by inspecting the scaffolded classes in the io.kestra.plugin.scylladb package and build.gradle, then compare the plugin-mongodb and plugin-jdbc implementations. Run ./gradlew test to establish the baseline and use Testcontainers with the scylladb/scylla image for coverage. Done means all listed tasks, trigger behavior, metadata, documentation, and ./gradlew build satisfy the acceptance criteria.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- docker, java
- Domain
- databases
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 25/100