kestra-io / kestra-io/plugin-scylladb

Implement ScyllaDB plugin: Query, Queries, Execute, Batch tasks and polling Trigger

Open
#3 0 comments 0 reactions 0 assignees View on GitHub
area/plugin good first issue
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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.