Cassandra Speculative Retry Storm Mitigation Guard

id: cassandra-speculative-retry-storm-mitigation-guard
namespace: company.database

description: |
  Continuously inspect Apache Cassandra tables for aggressive speculative retry configurations
  (such as ALWAYS or excessively low millisecond thresholds) that trigger read query duplication,
  track policy compliance drift across execution runs in Kestra KV, gate DDL mutations behind a human
  approval Pause, apply safe 99th-percentile (99p) or disabled retry policies via CQL ALTER TABLE,
  and verify post-remediation schema alignment.

inputs:
  - id: cassandra_host
    type: STRING
    displayName: Cassandra Host
    description: Hostname or IP address of a Cassandra contact point reachable from
      the Kestra worker.
    defaults: "localhost"

  - id: cassandra_port
    type: INT
    displayName: Cassandra CQL Port
    description: Native CQL port the Cassandra contact point listens on.
    defaults: 9042

  - id: cassandra_datacenter
    type: STRING
    displayName: Local Datacenter
    description: Datacenter treated as local by the CQL driver load balancing policy.
    defaults: "datacenter1"

  - id: cassandra_username
    type: STRING
    displayName: Cassandra Username
    description: Username for CQL authentication. Leave empty for clusters without
      authentication enabled. Password is read from the CASSANDRA_PASSWORD
      secret.
    defaults: ""

  - id: target_keyspaces
    type: ARRAY
    itemType: STRING
    displayName: Target Keyspaces
    description: Keyspaces to audit for speculative retry policy violations. Leave
      empty to audit all non-system keyspaces.
    defaults:
      - "ecommerce"
      - "analytics"

  - id: disallowed_policies
    type: ARRAY
    itemType: STRING
    displayName: Disallowed Retry Policies
    description: Speculative retry policy strings classified as unsafe and eligible
      for mitigation.
    defaults:
      - "ALWAYS"

  - id: max_fixed_ms_ceiling
    type: INT
    displayName: Minimum Fixed Latency Ceiling (ms)
    description: Threshold below which fixed-millisecond speculative retry
      configurations (e.g. 5ms, 10ms) are flagged as aggressive.
    defaults: 20

  - id: remediation_policy
    type: STRING
    displayName: Remediation Policy
    description: Safe speculative retry policy to apply via CQL ALTER TABLE (e.g.
      99p, 99PERCENTILE, or NONE).
    defaults: "99p"

  - id: min_consecutive_runs
    type: INT
    displayName: Minimum Consecutive Runs
    description: Number of consecutive audit runs a table must be observed with an
      unsafe policy before triggering automated alerts.
    defaults: 1

  - id: dry_run
    type: BOOL
    displayName: Dry Run Mode
    description: When true, audits tables and logs proposed CQL ALTER statements
      without executing DDL changes.
    defaults: true

triggers:
  - id: periodic_speculative_retry_guard
    type: io.kestra.plugin.core.trigger.Schedule
    description: Run automated Cassandra speculative retry policy audit every 4 hours.
    cron: "0 */4 * * *"
    disabled: true

  - id: on_demand_speculative_retry_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    description: Authenticated HTTP endpoint invoked by Prometheus read-latency
      alerts or SRE incident runbooks.
    key: "{{ secret('CASSANDRA_GUARD_WEBHOOK_KEY') }}"

tasks:
  - id: validate_inputs
    type: io.kestra.plugin.core.execution.Assert
    description: Assert operational inputs satisfy connectivity, threshold, and
      sanity limits.
    conditions:
      - "{{ inputs.cassandra_host | length > 0 }}"
      - "{{ inputs.cassandra_port > 0 and inputs.cassandra_port <= 65535 }}"
      - "{{ inputs.max_fixed_ms_ceiling >= 0 }}"
      - "{{ inputs.remediation_policy | length > 0 }}"
      - "{{ inputs.min_consecutive_runs >= 1 }}"
    errorMessage: "Invalid configuration inputs. Ensure host is specified, port is
      within 1-65535, ceiling is non-negative, and remediation policy is
      non-empty."

  - id: scan_tables_speculative_retry
    type: io.kestra.plugin.cassandra.standard.Query
    description: Read table metadata from system_schema.tables to inspect
      speculative_retry configurations.
    session:
      endpoints:
        - hostname: "{{ inputs.cassandra_host }}"
          port: "{{ inputs.cassandra_port }}"
      localDatacenter: "{{ inputs.cassandra_datacenter }}"
      username: "{{ inputs.cassandra_username }}"
      password: "{{ secret('CASSANDRA_PASSWORD') }}"
    cql: "SELECT keyspace_name, table_name, speculative_retry FROM
      system_schema.tables;"
    fetchType: FETCH

  - id: get_previous_audit_state
    type: io.kestra.plugin.core.kv.Get
    description: Retrieve previous speculative retry compliance tracker from Kestra
      KV store for sequential trend analysis.
    key: "cassandra_speculative_retry_{{ inputs.cassandra_host | slugify }}"
    errorOnMissing: false

  - id: inspect_speculative_retry_policies
    type: io.kestra.plugin.scripts.python.Script
    description: Analyze table speculative_retry settings against aggressive policy
      thresholds and construct remediation plan.
    containerImage: python:3.11-slim
    env:
      TABLE_ROWS: "{{ outputs.scan_tables_speculative_retry.rows | toJson }}"
      TARGET_KEYSPACES: "{{ inputs.target_keyspaces | toJson }}"
      DISALLOWED_POLICIES: "{{ inputs.disallowed_policies | toJson }}"
      MAX_FIXED_MS_CEILING: "{{ inputs.max_fixed_ms_ceiling }}"
      REMEDIATION_POLICY: "{{ inputs.remediation_policy }}"
      MIN_CONSECUTIVE_RUNS: "{{ inputs.min_consecutive_runs }}"
      PREVIOUS_STATE: "{{ (outputs.get_previous_audit_state.value ?? {}) | toJson }}"
      DRY_RUN: "{{ inputs.dry_run }}"
    script: |
      import os
      import json
      import re
      import datetime

      table_rows_raw = os.getenv("TABLE_ROWS", "[]")
      target_keyspaces_raw = os.getenv("TARGET_KEYSPACES", "[]")
      disallowed_policies_raw = os.getenv("DISALLOWED_POLICIES", "[]")
      max_fixed_ms_ceiling = int(os.getenv("MAX_FIXED_MS_CEILING", "20"))
      remediation_policy = os.getenv("REMEDIATION_POLICY", "99p")
      min_consecutive_runs = int(os.getenv("MIN_CONSECUTIVE_RUNS", "1"))
      previous_state_raw = os.getenv("PREVIOUS_STATE", "{}")
      dry_run = os.getenv("DRY_RUN", "true").lower() == "true"

      try:
          table_rows = json.loads(table_rows_raw)
      except Exception:
          table_rows = []

      try:
          target_keyspaces = json.loads(target_keyspaces_raw)
      except Exception:
          target_keyspaces = []

      try:
          disallowed_policies = json.loads(disallowed_policies_raw)
      except Exception:
          disallowed_policies = ["ALWAYS"]

      try:
          previous_state = json.loads(previous_state_raw) if isinstance(previous_state_raw, str) else previous_state_raw
          if not isinstance(previous_state, dict):
              previous_state = {}
      except Exception:
          previous_state = {}

      now = datetime.datetime.now(datetime.timezone.utc)
      now_iso = now.isoformat()

      system_prefixes = ("system", "system_schema", "system_auth", "system_distributed", "system_traces", "system_views")
      disallowed_normalized = [p.strip().upper() for p in disallowed_policies]

      prev_tables = previous_state.get("tables", {})
      new_table_tracker = {}

      flagged_tables = []
      remediation_commands = []
      table_names = []
      scanned_count = 0

      for row in table_rows:
          ks = str(row.get("keyspace_name", "")).strip()
          tbl = str(row.get("table_name", "")).strip()
          policy = str(row.get("speculative_retry", "")).strip()

          if not ks or not tbl:
              continue

          if any(ks == sp or ks.startswith(sp + "_") for sp in system_prefixes):
              continue

          if target_keyspaces and ks not in target_keyspaces:
              continue

          scanned_count += 1
          full_name = f"{ks}.{tbl}"

          is_disallowed = policy.upper() in disallowed_normalized
          is_aggressive_ms = False
          ms_match = re.match(r"^(\d+)\s*ms$", policy, re.IGNORECASE)
          if ms_match:
              val = int(ms_match.group(1))
              if val < max_fixed_ms_ceiling:
                  is_aggressive_ms = True

          is_unsafe = is_disallowed or is_aggressive_ms

          prev_entry = prev_tables.get(full_name, {})
          prev_consecutive = prev_entry.get("consecutive_runs", 0)

          consecutive_runs = (prev_consecutive + 1) if is_unsafe else 0
          new_table_tracker[full_name] = {
              "consecutive_runs": consecutive_runs,
              "speculative_retry": policy,
              "last_evaluated": now_iso
          }

          if is_unsafe and consecutive_runs >= min_consecutive_runs:
              if is_disallowed:
                  reason = f"Policy '{policy}' duplicates coordinator reads on every query"
              else:
                  reason = f"Fixed threshold '{policy}' is below safe ceiling of {max_fixed_ms_ceiling}ms"

              cql_cmd = f"ALTER TABLE {ks}.{tbl} WITH speculative_retry = '{remediation_policy}';"
              flagged_tables.append({
                  "keyspace": ks,
                  "table": tbl,
                  "full_name": full_name,
                  "current_policy": policy,
                  "target_policy": remediation_policy,
                  "reason": reason,
                  "cql": cql_cmd,
                  "consecutive_runs": consecutive_runs
              })
              remediation_commands.append(cql_cmd)
              table_names.append(full_name)

      has_violations = len(flagged_tables) > 0
      overall_status = "RETRY_STORM_RISK" if has_violations else "HEALTHY"

      if has_violations:
          summary_text = (
              f"Detected {len(flagged_tables)} table(s) configured with aggressive speculative retry policies "
              f"across {len(target_keyspaces) if target_keyspaces else 'all'} keyspace(s). "
              f"Remediation action: {'DRY RUN' if dry_run else f'ALTER TO {remediation_policy}'}."
          )
      else:
          summary_text = (
              f"All {scanned_count} audited table(s) comply with safe speculative retry standards. "
              f"Zero retry storm risks detected."
          )

      audit_manifest = {
          "status": overall_status,
          "scanned_table_count": scanned_count,
          "flagged_count": len(flagged_tables),
          "flagged_tables": flagged_tables,
          "table_names": table_names,
          "evaluated_at": now_iso,
          "summary": summary_text
      }

      new_state_snapshot = {
          "host": os.getenv("CASSANDRA_HOST", "localhost"),
          "last_audit": now_iso,
          "tables": new_table_tracker
      }

      print("::" + json.dumps({
          "outputs": {
              "overall_status": overall_status,
              "has_violations": has_violations,
              "scanned_count": scanned_count,
              "flagged_count": len(flagged_tables),
              "flagged_tables": flagged_tables,
              "table_names": table_names,
              "remediation_commands": remediation_commands,
              "new_state_json": json.dumps(new_state_snapshot),
              "audit_manifest": audit_manifest,
              "summary_text": summary_text
          }
      }) + "::")

  - id: save_audit_state
    type: io.kestra.plugin.core.kv.Set
    description: Persist updated table policy tracking in Kestra KV store for
      sequential trend analysis.
    key: "cassandra_speculative_retry_{{ inputs.cassandra_host | slugify }}"
    kvType: JSON
    value: "{{ outputs.inspect_speculative_retry_policies.vars.new_state_json }}"

  - id: remediation_gate
    type: io.kestra.plugin.core.flow.If
    description: Apply CQL ALTER statements to align speculative retry policies when
      violations are detected.
    condition: "{{ outputs.inspect_speculative_retry_policies.vars.has_violations }}"
    then:
      - id: dry_run_check
        type: io.kestra.plugin.core.flow.If
        description: Enforce human operator approval gate and execute DDL modifications
          when dry_run is disabled.
        condition: "{{ not inputs.dry_run }}"
        then:
          - id: pause_for_remediation_approval
            type: io.kestra.plugin.core.flow.Pause
            description: Pause execution for operator verification before altering Cassandra
              table definitions. Review flagged tables and current policies
              before resuming.

          - id: apply_remediation_loop
            type: io.kestra.plugin.core.flow.Loop
            description: Safely iterate over each flagged table and execute CQL ALTER TABLE
              command.
            values: "{{ outputs.inspect_speculative_retry_policies.vars.remediation_commands
              }}"
            tasks:
              - id: remediate_table_speculative_retry
                type: io.kestra.plugin.cassandra.standard.Query
                description: Execute CQL ALTER TABLE to apply safe speculative retry policy.
                session:
                  endpoints:
                    - hostname: "{{ inputs.cassandra_host }}"
                      port: "{{ inputs.cassandra_port }}"
                  localDatacenter: "{{ inputs.cassandra_datacenter }}"
                  username: "{{ inputs.cassandra_username }}"
                  password: "{{ secret('CASSANDRA_PASSWORD') }}"
                cql: "{{ item.value }}"
                fetchType: NONE

          - id: verify_speculative_retry_remediation
            type: io.kestra.plugin.cassandra.standard.Query
            description: Re-query system_schema.tables to verify that flagged tables have
              been updated to the safe policy.
            session:
              endpoints:
                - hostname: "{{ inputs.cassandra_host }}"
                  port: "{{ inputs.cassandra_port }}"
              localDatacenter: "{{ inputs.cassandra_datacenter }}"
              username: "{{ inputs.cassandra_username }}"
              password: "{{ secret('CASSANDRA_PASSWORD') }}"
            cql: "SELECT keyspace_name, table_name, speculative_retry FROM
              system_schema.tables;"
            fetchType: FETCH

          - id: verify_remediation_results
            type: io.kestra.plugin.scripts.python.Script
            description: Assert that none of the target tables remain with aggressive
              speculative retry policies.
            containerImage: python:3.11-slim
            env:
              REVERIFY_ROWS: "{{ outputs.verify_speculative_retry_remediation.rows | toJson
                }}"
              FLAGGED_TABLES: "{{ outputs.inspect_speculative_retry_policies.vars.table_names
                | toJson }}"
              REMEDIATION_POLICY: "{{ inputs.remediation_policy }}"
            script: |
              import os
              import json
              import re
              import sys

              rows = json.loads(os.getenv("REVERIFY_ROWS", "[]"))
              flagged = json.loads(os.getenv("FLAGGED_TABLES", "[]"))
              target_policy = os.getenv("REMEDIATION_POLICY", "99p")

              def normalize_policy(policy):
                  # Cassandra canonicalizes speculative_retry when read back from system_schema.tables:
                  # 'NONE' is accepted on ALTER but always reads back as 'NEVER', and '99PERCENTILE' /
                  # '99 PERCENTILE' always reads back as '99p'. Compare on that canonical form so a
                  # legitimately-applied policy isn't flagged as still-failing just because the caller
                  # used an accepted alias instead of the exact canonical spelling.
                  p = policy.strip()
                  if p.upper() in ("NEVER", "NONE"):
                      return "NEVER"
                  if p.upper() == "ALWAYS":
                      return "ALWAYS"
                  m = re.match(r"^([0-9.]+)\s*p(?:ercentile)?$", p, re.IGNORECASE)
                  if m:
                      return f"{float(m.group(1)):g}p"
                  m = re.match(r"^([0-9.]+)\s*ms$", p, re.IGNORECASE)
                  if m:
                      return f"{float(m.group(1)):g}ms"
                  return p.upper()

              still_failing = []
              for r in rows:
                  full_name = f"{r.get('keyspace_name')}.{r.get('table_name')}"
                  if full_name in flagged:
                      curr = str(r.get('speculative_retry', '')).strip()
                      if curr.upper() == "ALWAYS" or normalize_policy(curr) != normalize_policy(target_policy):
                          still_failing.append((full_name, curr))

              if still_failing:
                  print(f"FAILED: The following tables were not successfully updated: {still_failing}", file=sys.stderr)
                  sys.exit(1)
              print(f"SUCCESS: All {len(flagged)} table(s) verified safely aligned to {target_policy}.")

          - id: log_remediation_summary
            type: io.kestra.plugin.core.log.Log
            description: Record verified post-remediation schema alignment telemetry.
            message: >-
              Successfully remediated {{
              outputs.inspect_speculative_retry_policies.vars.flagged_count }}
              Cassandra table(s): {{
              outputs.inspect_speculative_retry_policies.vars.table_names }}.
              All tables now verified with safe speculative retry policy '{{
              inputs.remediation_policy }}'.
        else:
          - id: log_dry_run_plan
            type: io.kestra.plugin.core.log.Log
            description: Record dry-run detection summary without modifying database
              schemas.
            message: >-
              DRY RUN: Detected {{
              outputs.inspect_speculative_retry_policies.vars.flagged_count }}
              table(s) with aggressive speculative retry policies: {{
              outputs.inspect_speculative_retry_policies.vars.table_names }}.
              Proposed CQL alterations: {{
              outputs.inspect_speculative_retry_policies.vars.remediation_commands
              }}. Set inputs.dry_run to false to execute automated remediation
              behind operator approval.

  - id: notify_if_violations
    type: io.kestra.plugin.core.flow.If
    description: Only post to Slack when at least one table actually has an
      aggressive speculative retry policy, so a clean scheduled audit produces
      no notification noise.
    condition: "{{ outputs.inspect_speculative_retry_policies.vars.has_violations }}"
    then:
      - id: notify_slack_audit
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatch structured Slack notification with speculative retry
          compliance status and remediation telemetry.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "attachments": [
              {
                "color": "#E01E5A",
                "title": ":satellite: Cassandra Speculative Retry Guard, {{ inputs.cassandra_host }}:{{ inputs.cassandra_port }}",
                "text": "{{ outputs.inspect_speculative_retry_policies.vars.summary_text }}",
                "fields": [
                  {
                    "title": "Cluster Host",
                    "value": "`{{ inputs.cassandra_host }}:{{ inputs.cassandra_port }}`",
                    "short": true
                  },
                  {
                    "title": "Status",
                    "value": "*RETRY_STORM_RISK*",
                    "short": true
                  },
                  {
                    "title": "Flagged Tables",
                    "value": "{{ outputs.inspect_speculative_retry_policies.vars.flagged_count }}",
                    "short": true
                  },
                  {
                    "title": "Target Policy",
                    "value": "`{{ inputs.remediation_policy }}`",
                    "short": true
                  },
                  {
                    "title": "Remediation Mode",
                    "value": "`{{ inputs.dry_run ? 'DRY_RUN (Audit Only)' : 'APPLIED' }}`",
                    "short": true
                  },
                  {
                    "title": "Execution ID",
                    "value": "`{{ execution.id }}`",
                    "short": true
                  }
                ]
              }
            ]
          }

errors:
  - id: alert_on_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Dispatch emergency alert when speculative retry guard execution
      fails unexpectedly.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "attachments": [
          {
            "color": "#E01E5A",
            "title": ":rotating_light: Cassandra Speculative Retry Guard Failed",
            "text": "Execution `{{ execution.id }}` failed while inspecting or mitigating speculative retry policies on `{{ inputs.cassandra_host }}:{{ inputs.cassandra_port }}`.\nCheck node reachability, CQL permissions, or worker logs."
          }
        ]
      }

In Apache Cassandra, speculative retry is designed to eliminate high tail read latencies by allowing the coordinator node to issue a speculative read request to a second replica if the primary replica fails to respond within a defined threshold. However, when database tables are configured with aggressive speculative retry policies, most notably speculative_retry = 'ALWAYS' or excessively low millisecond thresholds (e.g., 5ms or 10ms), the coordinator duplicates read I/O across cluster nodes on every single query.

Under peak production query workloads, aggressive speculative retries double or triple coordinator read traffic, rapidly saturating JVM ReadStage thread pool queues and exhausting operating system page cache. When a temporary network blip or garbage collection pause occurs, duplicate speculative requests overwhelm struggling nodes, creating a self-reinforcing "retry storm" that triggers widespread ReadTimeoutException errors and brings down live customer microservices.

This blueprint provides an automated Day-2 operations guard that inspects system_schema.tables for aggressive speculative retry policies across user keyspaces, tracks policy drift across executions in Kestra KV, gates DDL changes behind a human approval Pause, issues safe CQL ALTER TABLE statements (aligning policies to 99p or NONE), verifies post-remediation schema conformity, and publishes detailed Slack audit telemetry.

How it works

Pre-flight Assertion: Validates operational inputs (io.kestra.plugin.core.execution.Assert), confirming valid hostnames, CQL port bounds (1-65535), and non-empty remediation policies.
Schema Metadata Collection: scan_tables_speculative_retry (io.kestra.plugin.cassandra.standard.Query) queries system_schema.tables over the native CQL binary protocol to extract keyspace names, table names, and current speculative_retry policies.
Historical State Retrieval: get_previous_audit_state (io.kestra.plugin.core.kv.Get) retrieves prior execution audit records from the Kestra KV store to identify chronic policy violations across runs.
Triage and Policy Analysis: A Python inspection script (inspect_speculative_retry_policies) excludes built-in system keyspaces (system, system_schema, system_auth), filters by target keyspaces, and flags tables where speculative_retry is set to ALWAYS or a fixed-latency threshold below max_fixed_ms_ceiling. It formats the proposed ALTER TABLE <keyspace>.<table> WITH speculative_retry = '<remediation_policy>' DDL statements.
Cross-Run State Persistence: save_audit_state (io.kestra.plugin.core.kv.Set) updates Kestra KV with consecutive violation counters and policy snapshots for every evaluated table.
Human Approval Gate: When unsafe tables are detected and dry_run is disabled, the workflow halts at pause_for_remediation_approval (io.kestra.plugin.core.flow.Pause), presenting operators with flagged table names, current aggressive policies, and proposed safe settings.
Write-Path Schema Alteration: Upon operator approval, apply_remediation_loop (io.kestra.plugin.core.flow.Loop) iterates through each flagged table and executes the safe ALTER TABLE CQL statement via io.kestra.plugin.cassandra.standard.Query.
Post-Remediation Verification: verify_speculative_retry_remediation re-queries system_schema.tables, and verify_remediation_results asserts that all target tables have been updated to the safe policy, confirming zero remaining retry storm risks.
Audit & Escalation Alerting: Structured Slack alerts (notify_slack_audit) report cluster health status, flagged table counts, and remediation modes, while flow-level errors capture cluster connectivity failures.

What you get

Continuous protection against Cassandra read thread pool exhaustion and cascading retry storms.
Automated detection of tables configured with ALWAYS or dangerous sub-20ms speculative retry floors.
Safe human-in-the-loop approval gate before executing DDL schema modifications on production clusters.
Automated CQL ALTER TABLE remediation aligning tables to recommended 99th-percentile (99p) thresholds.
Post-remediation schema re-query asserting complete elimination of aggressive policies.
Historical state persistence in Kestra KV store for compliance tracking and drift auditing.

Who it's for

Database Reliability Engineers (DBREs) and SREs managing multi-node Apache Cassandra or ScyllaDB clusters.
Platform Engineers maintaining distributed backend microservices subject to read latency spikes.
Data Infrastructure Teams seeking automated schema governance and performance guardrails.

Why orchestrate this with Kestra

Cassandra's command-line tools like cqlsh require manual query execution and lack automated scheduling, cross-run state tracking, or centralized audit logs. Custom cron scripts running cqlsh create single points of failure without built-in pause-and-resume approval mechanics or credential encryption.

Orchestrating this guard with Kestra delivers a declarative workflow with native CQL driver support, approval gates (Pause), cross-execution KV state tracking, secret masking, and real-time Slack escalation.

Prerequisites

An accessible Apache Cassandra cluster (version 3.11, 4.x, or 5.x) reachable from the Kestra worker.
CQL administrative credentials with permissions to query system_schema.tables and alter table schemas.
A Slack Incoming Webhook URL for incident alerting.

Secrets

SLACK_WEBHOOK_URL: Slack Incoming Webhook endpoint used for compliance notifications and error alerts.
CASSANDRA_GUARD_WEBHOOK_KEY: Secret authentication key protecting the on-demand webhook trigger.
CASSANDRA_PASSWORD: Password for Cassandra native authentication (if enabled on the cluster).

Quick start

Configure SLACK_WEBHOOK_URL and CASSANDRA_GUARD_WEBHOOK_KEY in your Kestra namespace secrets.
Import this blueprint and set cassandra_host, cassandra_port, and target_keyspaces to match your cluster.
Run a manual execution with dry_run: true to review current table configurations without altering schemas.
Enable the periodic_speculative_retry_guard schedule trigger to audit cluster tables every 4 hours.
When ready for automated alignment, set dry_run: false to allow operator-approved CQL table alterations.

How to extend

Tombstone Threshold Alignment: Chain an additional inspection step to audit tombstone_threshold and unchecked_tombstone_compaction settings on flagged tables.
PagerDuty Incident Escalation: Integrate io.kestra.plugin.pagerduty.notifications.PagerDutyAlert to page on-call DBREs when multiple tables exhibit chronic retry storm risks.
Nodetool Metric Sampling: Add a io.kestra.plugin.scripts.shell.Commands task to sample live JMX org.apache.cassandra.metrics:type=ThreadPools,path=request,scope=ReadStage metrics before and after DDL execution.

Cassandra Speculative Retry Storm Mitigation and Safe Alignment Guard

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Cassandra Speculative Retry Storm Mitigation and Safe Alignment Guard

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Infrastructure Blueprints

More Data Blueprints

New to Kestra?