VoltSP Advanced Tutorial (Java SDK)

1

What is VoltSP?

▶ Try in Sandbox instead — skip the local install. This launches a real VoltSP + VoltDB stack in a live container in your browser, so you can run every step for real. →

Welcome to the VoltSP Hello World tutorial! VoltSP is VoltDB's stream processing engine for building real-time data pipelines. It connects sources (like Kafka) to sinks (like VoltDB stored procedures) with transformation logic in between.

Pipeline Architecture

Every VoltSP pipeline follows a simple three-stage pattern:

  Source  ──>  Processing  ──>  Sink
(Kafka, Gen)   (.processWith)   (stdout, Kafka, VoltDB)

A Source produces data (from Kafka, a generator, a file, etc.). Processing functions transform each record using .processWith() — you can chain as many as you need. A Sink consumes the final output (stdout, Kafka, VoltDB stored procedure, etc.).

Key Concepts

VoltPipeline — the interface you implement. It has one method: define(VoltStreamBuilder stream)
VoltStreamBuilder — the fluent API for wiring source → processing → sink
Sources — factory for data sources (generators, Kafka, files)
Sinks — factory for data sinks (stdout, Kafka, VoltDB)
Configuration — YAML files provide runtime settings (Kafka brokers, TPS, etc.)

Note: VoltSP pipelines can be defined in Java (packaged as JARs) or entirely in YAML with no code at all. Both approaches are run with the voltsp CLI — no application server needed.

Project Setup

Let's start by setting up the Maven project. Navigate to the project directory in your container terminal — it's pre-installed and ready to build:

Shell

cd /opt/trainingUI/data/voltsp-hello-world
ls -la

Project Structure

voltsp-hello-world/
├── pom.xml
├── ddl/
│   └── create_greetings.sql
└── src/main/
    ├── java/hello/
    │   ├── GreetingPipeline.java           # Step 3
    │   ├── EnrichedGreetingPipeline.java   # Step 5
    │   ├── KafkaGreetingPipeline.java      # Step 7
    │   ├── KafkaToKafkaPipeline.java       # Step 8
    │   └── KafkaToVoltPipeline.java        # Step 9
    └── resources/
        ├── configuration.yaml             # GreetingPipeline
        ├── configuration-enriched.yaml    # EnrichedGreetingPipeline
        ├── configuration-kafka.yaml       # KafkaGreetingPipeline
        ├── configuration-kafka-to-kafka.yaml  # KafkaToKafkaPipeline
        └── configuration-volt.yaml        # KafkaToVoltPipeline

Maven Dependencies

The project needs these VoltSP dependencies (all provided scope — VoltSP ships them at runtime) plus the Kafka client library:

XML (pom.xml)

<!-- VoltSP Core API -->
<dependency>
  <groupId>org.voltdb</groupId>
  <artifactId>volt-stream-api</artifactId>
  <version>1.7.2-dev</version>
  <scope>provided</scope>
</dependency>

<!-- Connectors API (Sources, Sinks) -->
<dependency>
  <groupId>org.voltdb</groupId>
  <artifactId>volt-stream-connectors-api</artifactId>
  <version>1.7.2-dev</version>
  <scope>provided</scope>
</dependency>

<!-- Kafka Plugin -->
<dependency>
  <groupId>org.voltdb</groupId>
  <artifactId>volt-stream-plugin-kafka-api</artifactId>
  <version>1.7.2-dev</version>
  <scope>provided</scope>
</dependency>

<!-- VoltDB Procedure Plugin -->
<dependency>
  <groupId>org.voltdb</groupId>
  <artifactId>volt-stream-plugin-volt-api</artifactId>
  <version>1.7.2-dev</version>
  <scope>provided</scope>
</dependency>

<!-- Kafka Client (for Serializer/Deserializer classes) -->
<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-clients</artifactId>
  <version>3.4.0</version>
  <scope>provided</scope>
</dependency>

Note: Dependencies are scope=provided because VoltSP provides them at runtime. Your JAR only needs your pipeline code.

Your First Pipeline

Let's build the simplest possible pipeline: generate greeting messages, convert them to uppercase, and print to the console.

GreetingPipeline.java

Java

package hello;

import java.util.concurrent.ThreadLocalRandom;

import org.voltdb.stream.api.Sources;
import org.voltdb.stream.api.Sinks;
import org.voltdb.stream.api.pipeline.VoltPipeline;
import org.voltdb.stream.api.pipeline.VoltStreamBuilder;

public class GreetingPipeline implements VoltPipeline {

  private static final String[] NAMES = {
      "Alice", "Bob", "Charlie", "Diana", "Eve"
  };

  @Override
  public void define(VoltStreamBuilder stream) {
    int tps = stream.getExecutionContext()
        .configurator()
        .findByPath("tps").orElse(5);

    stream
      .withName("greeting-pipeline")
      .consumeFromSource(
          Sources.<String>generateAtRate(tps,
              () -> "Hello, "
                  + NAMES[ThreadLocalRandom.current()
                      .nextInt(NAMES.length)]
                  + "!"))
      .processWith(String::toUpperCase)
      .processWith(s -> s + "\n")
      .terminateWithSink(Sinks.stdout());
  }
}

Code Walkthrough

Let's break down each part:

implements VoltPipeline — Every pipeline implements this interface
define(VoltStreamBuilder stream) — The single method where you wire your pipeline
Sources.generateAtRate(tps, supplier) — Generates strings at 5 events/second
.processWith(String::toUpperCase) — Transforms each message to uppercase
.processWith(s -> s + "\n") — Adds a newline (stdout doesn't add one automatically)
.terminateWithSink(Sinks.stdout()) — Prints each result to the console

The getExecutionContext().configurator().findByPath("tps") reads the tps value from configuration.yaml. If not found, it defaults to 5.

Note: The pipeline reads configuration at startup, not per-event. Source, process, and sink are wired once in define().

4

Build & Run

Now let's build the project and run our first pipeline.

Build with Maven

Shell

cd /opt/trainingUI/data/voltsp-hello-world
mvn clean package

This compiles your pipeline classes and packages them into target/voltsp-hello-world-1.0-SNAPSHOT.jar.

Run with VoltSP

Shell

cd /opt/trainingUI/data/voltsp-hello-world
export CP=target/voltsp-hello-world-1.0-SNAPSHOT.jar
voltsp -c src/main/resources/configuration.yaml

Expected Output

You should see uppercase greetings printed at 5 per second:

HELLO, CHARLIE!
HELLO, ALICE!
HELLO, BOB!
HELLO, DIANA!
HELLO, ALICE!
HELLO, EVE!
...

Note: Press Ctrl+C to stop the pipeline. The generator runs indefinitely until stopped.

Configuration

The configuration.yaml file is minimal:

YAML

className: hello.GreetingPipeline

# Generator rate (events per second)
tps: 5

Try changing tps to 10 or 20 and restarting — you'll see messages arrive faster.

Processing Functions

VoltSP lets you chain multiple .processWith() stages. Each stage receives the output of the previous one and can transform, enrich, or filter records.

Chaining Stages

Here's a pipeline that chains four processing stages:

Java

stream
  .withName("enriched-greeting-pipeline")
  .consumeFromSource(
      Sources.<String>generateAtRate(tps,
          () -> "Hello, "
              + NAMES[ThreadLocalRandom.current()
                  .nextInt(NAMES.length)]
              + "!"))

  // Stage 1: convert to uppercase
  .processWith(String::toUpperCase)

  // Stage 2: append a timestamp
  .processWith(msg ->
      msg + " [" + System.currentTimeMillis() + "]")

  // Stage 3: filter -- return null to drop
  .processWith(msg ->
      msg.contains("EVE") ? null : msg)

  // Stage 4: add newline for stdout
  .processWith(s -> s + "\n")

  .terminateWithSink(Sinks.stdout());

How It Works

Stage 1: Converts "Hello, Alice!" → "HELLO, ALICE!"
Stage 2: Adds a timestamp → "HELLO, ALICE! [1707234567890]"
Stage 3: Drops messages containing "EVE" by returning null

Note: Returning null from a processWith() stage drops the record — it won't reach downstream stages or the sink. This is how you implement filtering in VoltSP.

Try It

Shell

cd /opt/trainingUI/data/voltsp-hello-world
export CP=target/voltsp-hello-world-1.0-SNAPSHOT.jar
voltsp -c src/main/resources/configuration-enriched.yaml

Expected output — notice Eve's greeting is missing:

HELLO, CHARLIE! [1707234567890]
HELLO, ALICE! [1707234567891]
HELLO, BOB! [1707234567892]
HELLO, DIANA! [1707234567893]
HELLO, BOB! [1707234567894]
...

YAML Configuration

VoltSP uses YAML configuration files for runtime settings. You've already seen a simple one — now let's understand how configuration works in more detail.

Configuration File

Every configuration file specifies the pipeline className and any custom settings:

YAML

className: com.example.MyPipeline

# Source configuration (for Kafka, file, etc.)
source:
  kafka:
    bootstrapServers:
      - "localhost:9092"
    topicNames:
      - "my-topic"
    groupId: "my-group"

# Custom application settings
tps: 100
myCustomSetting: "value"

Reading Configuration in Java

Access any YAML key with the configurator API:

Java

ConfigurationContext config =
    stream.getExecutionContext().configurator();

// Read a top-level key with a default value
int tps = config.findByPath("tps").orElse(100);

// Read a nested key
String servers = config
    .findByPath("source.kafka.bootstrapServers")
    .orElse("localhost:9092");

The findByPath() method navigates the YAML tree using dot-separated paths. The orElse() provides a fallback if the key is missing.

Multiple Configuration Files

configuration.yaml — GreetingPipeline (Step 4)
configuration-enriched.yaml — EnrichedGreetingPipeline (Step 5)
configuration-kafka.yaml — KafkaGreetingPipeline (Step 7)
configuration-kafka-to-kafka.yaml — KafkaToKafkaPipeline (Step 8)
configuration-volt.yaml — KafkaToVoltPipeline (Step 9)

Switch configurations by passing a different -c flag to voltsp.

Note: Kafka source settings (bootstrapServers, topicNames, groupId, startingOffset) can be set in YAML or programmatically in Java. YAML values are applied automatically — Java builder calls override them.

Kafka Source

Now let's replace the generator with a real Kafka source. This pipeline reads messages from a Kafka topic, transforms them, and prints the results.

KafkaGreetingPipeline.java

Java

public class KafkaGreetingPipeline implements VoltPipeline {
  @Override
  public void define(VoltStreamBuilder stream) {
    stream
      .withName("kafka-greeting-pipeline")
      .consumeFromSource(
          KafkaSourceConfigBuilder.<String>builder()
            .withStartingOffset(KafkaStartingOffset.EARLIEST)
            .withKeyDeserializer(StringDeserializer.class)
            .withValueDeserializer(StringDeserializer.class))
      .processWith(record ->
          record.getValue().toUpperCase() + "\n")
      .terminateWithSink(Sinks.stdout());
  }
}

Key Differences from Generator

KafkaSourceConfigBuilder replaces Sources.generateAtRate()
The builder specifies offset strategy, key/value deserializers
Kafka connection details (bootstrapServers, topicNames, groupId) come from the YAML config file
The .processWith() receives a Kafka ConsumerRecord — use .getValue() to get the message

Configuration

YAML

className: hello.KafkaGreetingPipeline

source:
  kafka:
    bootstrapServers:
      - "localhost:9092"
    topicNames:
      - "greetings"
    groupId: "voltsp-hello"
    startingOffset: "EARLIEST"

sink:
  kafka:
    topicName: "greetings-output"
    bootstrapServers: "localhost:9092"

outputTopic: "greetings-output"

Test It

Start Kafka (if not already running), create the topic, and send some messages:

Shell

# Create the greetings topic
kafka-topics.sh --create --topic greetings --bootstrap-server localhost:9092

# Send test messages
kafka-console-producer.sh --topic greetings --bootstrap-server localhost:9092
> Hello, World!
> Hi from Kafka!
> Greetings, VoltSP!

In another terminal, start the pipeline:

Shell

cd /opt/trainingUI/data/voltsp-hello-world
export CP=target/voltsp-hello-world-1.0-SNAPSHOT.jar
voltsp -c src/main/resources/configuration-kafka.yaml

HELLO, WORLD!
HI FROM KAFKA!
GREETINGS, VOLTSP!

Kafka Sink

Now let's complete the Kafka-to-Kafka pattern: read from one topic, transform, and write to another topic.

KafkaToKafkaPipeline.java

Java

public class KafkaToKafkaPipeline implements VoltPipeline {
  @Override
  public void define(VoltStreamBuilder stream) {
    String outputTopic = stream.getExecutionContext()
        .configurator()
        .findByPath("outputTopic")
        .orElse("greetings-output");

    stream
      .withName("kafka-to-kafka-pipeline")
      .consumeFromSource(
          KafkaSourceConfigBuilder.<String>builder()
            .withStartingOffset(KafkaStartingOffset.EARLIEST)
            .withKeyDeserializer(StringDeserializer.class)
            .withValueDeserializer(StringDeserializer.class))
      .processWith(record -> record.getValue().toUpperCase())
      .terminateWithSink(
          KafkaSinkConfigBuilder.<String>builder()
            .withTopicName(outputTopic)
            .withValueSerializer(StringSerializer.class));
  }
}

How the Sink Works

KafkaSinkConfigBuilder replaces Sinks.stdout()
.withTopicName() — the output Kafka topic
.withValueSerializer() — how to serialize the output (StringSerializer for plain text)
Sink Kafka connection uses the same YAML config (sink.kafka.bootstrapServers) or falls back to the source bootstrap servers

Test It

Shell

# Create the output topic
kafka-topics.sh --create --topic greetings-output --bootstrap-server localhost:9092

Shell

# Run the pipeline
cd /opt/trainingUI/data/voltsp-hello-world
export CP=target/voltsp-hello-world-1.0-SNAPSHOT.jar
voltsp -c src/main/resources/configuration-kafka-to-kafka.yaml

In another terminal, send messages to the input topic:

Shell

kafka-console-producer.sh --topic greetings \
  --bootstrap-server localhost:9092
> hello world
> voltsp is awesome
> stream processing rocks

Type messages and press Enter to send each one. Then in another terminal, consume from the output topic:

Shell

kafka-console-consumer.sh --topic greetings-output \
  --bootstrap-server localhost:9092 --from-beginning

You should see uppercase versions of whatever you send to the greetings topic.

Note: This Kafka → VoltSP → Kafka pattern is the foundation of production stream processing. VoltSP handles consumer group management, offset commits, and exactly-once delivery automatically.

VoltDB Integration

The final piece: connecting VoltSP to VoltDB. Instead of writing to stdout or Kafka, we'll call a VoltDB stored procedure for each event.

Step 1: Create the VoltDB Schema

First, create the greetings table and the InsertGreeting procedure in VoltDB:

SQL

CREATE TABLE greetings (
  sender VARCHAR(64) NOT NULL,
  recipient VARCHAR(64) NOT NULL,
  message VARCHAR(256) NOT NULL,
  greeting_time TIMESTAMP DEFAULT NOW
);
PARTITION TABLE greetings ON COLUMN sender;

CREATE PROCEDURE InsertGreeting
  PARTITION ON TABLE greetings COLUMN sender
AS
  INSERT INTO greetings (sender, recipient, message, greeting_time)
  VALUES (?, ?, ?, TO_TIMESTAMP(MILLISECOND, ?));

Step 2: The Pipeline

Java

public class KafkaToVoltPipeline implements VoltPipeline {
  @Override
  public void define(VoltStreamBuilder stream) {
    stream
      .withName("kafka-to-volt-pipeline")
      .configureResource("voltdb",
          VoltDBResourceConfigBuilder.class)
      .consumeFromSource(
          KafkaSourceConfigBuilder.<String>builder()
            .withStartingOffset(KafkaStartingOffset.EARLIEST)
            .withKeyDeserializer(StringDeserializer.class)
            .withValueDeserializer(StringDeserializer.class))
      .processWith(record -> {
          String[] parts = record.getValue().split("\\|", 3);
          String sender = parts.length > 0 ? parts[0].trim() : "Unknown";
          String recipient = parts.length > 1 ? parts[1].trim() : "World";
          String message = parts.length > 2 ? parts[2].trim() : record.getValue();
          return new VoltProcedureRequest("InsertGreeting",
              new Object[]{ sender, recipient,
                  message, System.currentTimeMillis() });
      })
      .terminateWithSink(
          VoltProcedureSinkConfigBuilder.builder()
            .withVoltClientResourceName("voltdb"));
  }
}

Key Concepts

configureResource("voltdb", VoltDBResourceConfigBuilder.class) — registers a VoltDB connection resource
VoltProcedureRequest — wraps a stored procedure name and its parameters as an Object[]
VoltProcedureSinkConfigBuilder — sends each VoltProcedureRequest to VoltDB
.withVoltClientResourceName("voltdb") — links the sink to the resource configured above

Step 3: Run & Test

Send pipe-delimited messages to Kafka (sender|recipient|message):

Shell

kafka-console-producer.sh --topic greetings --bootstrap-server localhost:9092
> Alice|Bob|Hello from the stream!
> Charlie|Diana|VoltSP is working!

Shell

cd /opt/trainingUI/data/voltsp-hello-world
export CP=target/voltsp-hello-world-1.0-SNAPSHOT.jar
voltsp -c src/main/resources/configuration-volt.yaml

Verify the data arrived in VoltDB:

SQL

SELECT * FROM greetings ORDER BY greeting_time DESC;

SENDER   RECIPIENT  MESSAGE                      GREETING_TIME
-------- ---------- ---------------------------- --------------------------
Alice    Bob        Hello from the stream!       2024-02-06 12:34:56.789000
Charlie  Diana      VoltSP is working!           2024-02-06 12:34:57.123000

Read more: VoltDB Procedure Sink
Read more: VoltDB Client Resource

10

Next Steps

Congratulations! You've built a complete VoltSP streaming pipeline — from simple generators to a production-ready Kafka-to-VoltDB flow.

What You Learned

VoltPipeline interface — the entry point for every Java pipeline
Sources — generators and Kafka consumers
processWith() — chainable transformation stages (return null to filter)
Sinks — stdout, Kafka topics, and VoltDB stored procedures
YAML configuration — externalize settings with findByPath()
YAML-only pipelines — define entire pipelines without Java code
VoltProcedureRequest — map stream data to stored procedure calls

Going Further

Windowing — aggregate events over time windows (tumbling, sliding)
Emitters — fan-out one event to multiple procedure calls
Multi-branch pipelines — split processing into parallel paths
ONNX integration — run ML inference inside your pipeline
Metrics — monitor throughput and latency with --metrics-port

Production Deployment

In production, VoltSP runs as a containerized process alongside your Kafka cluster and VoltDB cluster. Typical deployment:

Shell

export CP=/app/voltsp-hello-world.jar
voltsp -c /etc/voltsp/configuration.yaml \
  -p 8 --metrics-port 11785 \
  hello.KafkaToVoltPipeline

The --parallelism flag controls the number of consumer threads (default: auto based on CPU count). The --metrics-port exposes Prometheus-compatible metrics.

Prerequisites

What is VoltSP?

Pipeline Architecture

Key Concepts

Project Setup

Project Structure

Maven Dependencies

Your First Pipeline

GreetingPipeline.java

Code Walkthrough

Build & Run

Build with Maven

Run with VoltSP

Expected Output

Configuration

Processing Functions

Chaining Stages

How It Works

Try It

YAML Configuration

Configuration File

Reading Configuration in Java

Multiple Configuration Files

Kafka Source

KafkaGreetingPipeline.java

Key Differences from Generator

Configuration

Test It

Kafka Sink

KafkaToKafkaPipeline.java

How the Sink Works

Test It

VoltDB Integration

Step 1: Create the VoltDB Schema

Step 2: The Pipeline

Key Concepts

Step 3: Run & Test

Next Steps

What You Learned

Going Further

Production Deployment