KasprAgents - Distributed Stream Processors

What Is a KasprAgent?

A KasprAgent is a stream processor defined as a Kubernetes custom resource. It listens to incoming event streams from Kafka topics or in-memory channels, applies custom logic using a programmable pipeline of operations, and optionally produces results to output topics, channels, or sinks.

Key Features:

• Declarative configuration via Kubernetes CRDs
• Horizontal scalability and automatic partition rebalancing
• Flexible input/output (Kafka topics, channels, custom sinks)
• Programmable transformation pipeline (Python functions)
• Built-in support for batching, error handling, and stateful processing

Why use KasprAgents?

• Version, manage, and deploy stream processors like any other Kubernetes resource
• Integrate with existing Kafka infrastructure
• Simplify complex event-driven workflows

Example Use Cases

• ETL Pipelines: Transform and enrich data from Kafka topics before loading into databases
• Anomaly Detection: Process sensor streams and emit alerts for outlier events
• Event Enrichment: Join events with reference tables and output enriched results

Quickstart

Create a KasprAgent

A minimal agent can be created by specifying an input topic and a simple mapping processor. This example echoes incoming events to an output topic:

apiVersion: kaspr.io/v1alpha1
kind: KasprAgent
metadata:
  name: echo-agent
spec:
  input:
    topic:
      name: demo-input
  output:
    topics:
      - name: demo-output
  processors:
    pipeline:
      - echo
    operations:
      - name: echo
        map:
          python: |
            def echo(value):
                return value

Apply it using:

kubectl apply -f echo-agent.yaml

Partition Isolation

Set spec.isolatedPartitions: true when an agent should process each assigned Kafka partition in isolation.

apiVersion: kaspr.io/v1alpha1
kind: KasprAgent
metadata:
  name: partition-local-agent
spec:
  isolatedPartitions: true
  input:
    topic:
      name: orders
  processors:
    pipeline:
      - normalize
    operations:
      - name: normalize
        map:
          python: |
            def normalize(value):
                return value

With partition isolation enabled, the runtime starts a dedicated agent worker for each input partition assigned to the pod. When Kafka rebalances partitions, workers for revoked partitions are stopped and workers for newly assigned partitions are started.

Use this mode when your processing depends on partition-local ordering or partition-scoped state. Leave it disabled for the default shared-consumer behavior.

Components of a KasprAgent

Input

Agents must define their input source. This is either:

• A Kafka topic, defined with name or pattern, and serialization settings.
• An in-memory channel, useful for internal communication or testing.

You can also set declare: true to have the agent auto-create the topic if missing.

input:
  declare: true
  topic:
    name: my-topic
    keySerializer: json
    valueSerializer: json

Input Buffering

KasprAgents can buffer multiple input events before processing them as a batch. This is useful for scenarios where processing efficiency improves with larger batches, such as bulk database operations or machine learning inference.

input:
  topic:
    name: sensor-data
  take:
    max: 100           # Process up to 100 events at once
    within: 5s         # Or process after 5 seconds, whichever comes first

When buffering is enabled, your processor functions receive a list of events instead of individual events:

processors:
  operations:
    - name: batch-process
      map:
        python: |
          def process_batch(events):
              # events is now a list of values
              processed = []
              for event in events:
                  event["batch_size"] = len(events)
                  processed.append(event)
              return processed

Buffering is particularly beneficial for:

• Bulk operations: Database inserts, API calls with batch endpoints
• Statistical analysis: Computing aggregates over windows of data
• ML inference: Batch prediction for better GPU utilization
• I/O optimization: Reducing network round trips

Output

KasprAgents can emit output to:

• Kafka topics, defined with static names or dynamic functions.
• In-memory channels for internal use.
• Custom sinks defined by Python code

Each output target can define:

• declare: true to automatically create missing topics or channels
• keySelector, valueSelector, and optional headersSelector
• predicate for filtering what values to allow to pass through
• partitionSelector for custom partitioning
• ack: true for delivery guarantees

Example:

output:
  topics:
    - name: my-output-topic
      declare: true
      keySelector:
        python: |
          def get_key(value):
              return value["id"]
      predicate:
        python: |
          def should_send(value):
              return "id" in value

Processors

The heart of a KasprAgent is its processors section, which defines the transformation pipeline.

A processor pipeline is:

• Declared as a sequence of named operations
• Backed by user-defined Python map, filter, and optional topicSend side effects
• Optional tables can be attached for stateful logic

Example:

processors:
  pipeline:
    - validate
    - enrich
  init:
    python: |
      def init():
          print("Initializing agent...")
  operations:
    - name: validate
      filter:
        python: |
          def validate(value):
              return value.get("is_valid", False)
    - name: enrich
      map:
        python: |
          def enrich(value):
              value["processed"] = True
              return value

topicSend in Agent Pipelines

Use topicSend inside a processor operation when the agent needs to publish a Kafka side effect from the middle of the pipeline.

• Use filter when you want to discard the pipeline value.
• Use predicate on topicSend when you only want to decide whether a Kafka message should be sent.
• Use passThrough: true when the input value to topicSend should also be the output value passed to the next operation.
• With passThrough: true, the pipeline continues with the original input whether the send happened or was skipped by predicate.
• With passThrough: false, topicSend behaves as a terminal side effect for that pipeline branch.

Example:

processors:
  pipeline:
    - prepare
    - publish-schema-proposal
    - finalize
  operations:
    - name: prepare
      map:
        python: |
          def prepare(value):
              return {
                  "original": value,
                  "schema_proposal": {
                      "entity": value["entity"],
                      "schema": value.get("candidate_schema"),
                  },
              }
    - name: publish-schema-proposal
      topicSend:
        name: schema-proposals
        passThrough: true
        predicate:
          python: |
            def should_send(value):
                return value["schema_proposal"]["schema"] is not None
        keySelector:
          python: |
            def select_key(value):
                return value["schema_proposal"]["entity"]
        valueSelector:
          python: |
            def select_value(value):
                return value["schema_proposal"]
    - name: finalize
      map:
        python: |
          def finalize(value):
              value["original"]["schema_proposal_checked"] = True
              return value["original"]

In this pattern, predicate decides whether the side-effect publish happens, while passThrough decides whether the pipeline continues with the same input value.

Initialization

The init block lets you define startup logic, load configuration, prepare resources, or validate environment variables.

init:
  python: |
    import os
    if "MY_VAR" not in os.environ:
        raise RuntimeError("Missing MY_VAR")

Using External Python Packages

KasprAgents can use third-party Python libraries, but those packages must be installed at the KasprApp level first. Add them under spec.pythonPackages on the app, then reference them from your agent code. See the Python Packages guide for cache options, private registries, and installation policy details.

For example, this app installs the Stripe SDK and provides the API key through an environment variable:

apiVersion: kaspr.io/v1alpha1
kind: KasprApp
metadata:
  name: payments-app
spec:
  bootstrapServers: kafka:9092
  storage:
    type: persistent-claim
    class: standard
    size: 1Gi
  template:
    kasprContainer:
      env:
        - name: STRIPE_API_KEY
          valueFrom:
            secretKeyRef:
              name: stripe-api-key
              key: api-key
  pythonPackages:
    packages:
      - stripe==10.17.0

Once the package is available in the app container, import and configure it in the agent. A simple pattern is to initialize the SDK once in init, then reuse it in the operation handler:

apiVersion: kaspr.io/v1alpha1
kind: KasprAgent
metadata:
  name: create-payment-intent
  labels:
    kaspr.io/app: payments-app
spec:
  input:
    topic:
      name: checkout-requests
  output:
    topics:
      - name: payment-results
  processors:
    pipeline:
      - create-payment
    init:
      python: |
        import os
        import stripe
 
        stripe_api_key = os.getenv("STRIPE_API_KEY")
        if not stripe_api_key:
            raise RuntimeError("Missing STRIPE_API_KEY")
 
        stripe.api_key = stripe_api_key
    operations:
      - name: create-payment
        map:
          python: |
            def create_payment(value):
                intent = stripe.PaymentIntent.create(
                    amount=value["amount"],
                    currency=value.get("currency", "usd"),
                    customer=value["customer_id"],
                    automatic_payment_methods={"enabled": True},
                    metadata={"order_id": value["order_id"]},
                )
 
                return {
                    "order_id": value["order_id"],
                    "payment_intent_id": intent["id"],
                    "status": intent["status"],
                    "client_secret": intent["client_secret"],
                }

This keeps each event handler focused on per-event work only. The SDK setup happens once during startup, which reduces overhead and makes failures like missing credentials surface immediately.

In production, prefer these practices when using external packages:

• Store API keys and tokens in Kubernetes Secrets, not inline YAML.
• Pin package versions so agent behavior stays reproducible across deployments.

Package Complex Agent Logic as a Standalone Module

For larger agents, a useful pattern is to build the domain logic as a normal Python package and treat Kaspr as the runtime harness around it. In this model, the KasprAgent is responsible for reading events, calling your package code, and routing the results, while the business logic lives in a reusable module with its own tests.

This approach works well when you want to:

• Develop and test the core application logic outside of the Kaspr CRD lifecycle.
• Reuse the same logic across multiple Kaspr apps or agents.
• Keep agent YAML focused on streaming concerns such as input, output, routing, and tables.
• Reduce the amount of inline Python embedded directly in the CRD.

For example, imagine a package named payments_core that contains the Stripe-specific payment workflow:

from payments_core.checkout import PaymentService
 
service = PaymentService()
 
def create_payment(value):
    return service.create_payment_intent(value)

In the agent, you can import that package and call its methods from your operation handlers:

processors:
  pipeline:
    - create-payment
  init:
    python: |
      from payments_core.checkout import PaymentService
 
      service = PaymentService()
  operations:
    - name: create-payment
      map:
        python: |
          def create_payment(value):
              return service.create_payment_intent(value)

This keeps Kaspr-specific code thin and makes the package independently testable with ordinary Python unit and integration tests. You can validate payment rules, error handling, and request transformation without needing to deploy a KasprAgent for every test run.

Handler Context (app and event)

Inside init and operation handlers (map, filter, etc.), Kaspr provides a context dict in scope.

app = context["app"]
event = context["event"]

• context["app"] gives access to the underlying app instance.
• context["event"] gives the current event object.

For graceful shutdown logic, check app.should_stop in long-running handlers and exit cleanly when it becomes True.

def process(value):
    app = context["app"]
    if app.should_stop:
        # Flush state, release resources, then return.
        return value
    return value

event exposes both deserialized data and broker-level metadata:

• event.key, event.value: deserialized Python-native key/value.
• event.message.key, event.message.value: raw serialized payloads from Kafka.
• event.message.partition: source partition id.
• event.message.offset: offset within the partition.
• event.message.timestamp: message timestamp (Unix epoch).

def inspect(value):
    event = context["event"]
    print("decoded", event.key, event.value)
    print("raw", event.message.key, event.message.value)
    print("position", event.message.partition, event.message.offset)
    return value

When using input buffering with take, context["event"] is a list of event objects (aligned with the buffered values).

def process_batch(values):
    events = context["event"]  # list[Event]
    for value, event in zip(values, events):
        print(event.message.partition, event.message.offset, value)
    return values

Table Access

Agents can reference state tables and use them as inputs to their operations.

tables:
  - name: my-table
    paramName: table

map:
  python: |
    def enrich(value, table):
        value["status"] = table.get(value["id"], "unknown")
        return value

Partitioning and Routing

Kafka partitioning is automatically leveraged if your input topics are partitioned. You can control output partitioning via partitionSelector.

You may also implement repartitioning by emitting records with a newly computed key via keySelector and routing to a new topic.

Processing Failures

When a KasprAgent encounters an exception during event processing, the system maintains exactly-once semantics by acknowledging the source message and not reprocessing it. While this prevents duplicate processing, it raises the question of what happens to failed events.

There are several approaches to handling failed events, each with trade-offs:

Acknowledgment (Current Behavior)

• The failed message is acknowledged and marked as complete
• Ensures exactly-once processing semantics
• Failed events are not reprocessed automatically

Retry Strategies

• Retrying requires stopping topic processing to maintain message ordering
• The next offset cannot be processed until the failed event is resolved
• Moving events to the “back of the queue” breaks topic ordering

Instance Restart

• Crashing the instance forces human intervention
• Not ideal given the frequency of code errors and unexpected exceptions
• Better to log errors and notify operations teams for manual replay

Explicit Error Handling

Agents may emit to dead-letter queues (DLQs) using conditional logic in predicate or return values that match failure conditions.

output:
  topics:
    - name: dead-letter-topic
      predicate:
        python: |
          def should_send(value):
              return "error" in value

Concurrency and Scaling

KasprAgents scale by deploying multiple KasprApp replicas. Kafka ensures partition-based routing, meaning each message lands on exactly one instance. Kafka automatically distributes topic partitions across available agent instances using consumer group rebalancing. When you scale agent replicas:

• Adding instances: New agents join the consumer group and Kafka redistributes partitions to balance the load
• Removing instances: Kafka detects unavailable agents and reassigns their partitions to remaining instances
• Partition assignment: Each partition is consumed by exactly one agent instance at a time, ensuring ordered processing

This automatic rebalancing means agents can scale horizontally without manual intervention, with processing automatically redistributed as the cluster size changes.

Consuming a Join Channel

Agents can consume the output of a KasprJoin by referencing its output channel name in the input configuration. This allows agents to process joined records from two tables without managing the join logic themselves.

apiVersion: kaspr.io/v1alpha1
kind: KasprAgent
metadata:
  name: process-joined-data
  labels:
    kaspr.io/app: my-app
spec:
  input:
    channel:
      name: orders-products-joined  # matches the KasprJoin's outputChannel
  processors:
    pipeline:
      - handle-joined
    operations:
      - name: handle-joined
        map:
          entrypoint: process
          python: |
            def process(value):
                left_record = value["left"]
                right_record = value["right"]
                return {"merged": {**left_record, **right_record}}

The joined value has the structure {"left": ..., "right": ...} where left is the record from the left table and right is the matching record from the right table. See the Joins guide for full details.

FAQ & Troubleshooting

Q: Why isn’t my agent processing events? A: Check that your input topic/channel exists and is correctly configured. Use declare: true to auto-create topics. Review agent logs for errors.

Q: How do I debug processor errors? A: Enable detailed logging in your processor functions. Use DLQs to capture failed events for analysis.

Q: Can I use global state in processor functions? A: Use tables for persistent or shared state.

Q: How do I handle schema changes in input events? A: Update your processor functions to handle new fields or formats. Consider versioning your agent resources.