Voltdb-procedure

The emitter voltdb-procedure invokes VoltDB stored procedure. It employs at-least-once delivery guarantees.

For each event it will synchronously or asynchronously invoke stored procedure passing event data as arguments. When using the Java API the argument to the sink is a VoltRequest object that contains: - parameters: an array of Object-s that correspond to stored procedure arguments. For example, to invoke a stored procedure that has public long run(long id, int anotherId, String someString) the Object[] should contain Long, Integer and String or compatible types as defined in the VoltDB documentation. - procedureName: name of the stored procedure to run for this specific event.

A call that fails because the connection was lost, the call timed out, the response was lost, the server is unavailable for a reason other than a pause, or the client could not send it is retried with exponential backoff. The number of retries and the delays are set by the retry configuration nested within the client configuration. A synchronous call (asyncCall: false) waits on the worker during the backoff. A call that VoltDB rejects, or that fails in a way a delay cannot fix, is not retried.

The voltdb-procedure operator requires VoltDBResource to be configured either in yaml or in java, see examples.

JAVAYAML

// if resource is configured in yaml
voltStreamBuilder.configureResource("primary-cluster", VoltDBResourceConfigBuilder.class);
// or
voltStreamBuilder.configureResource("primary-cluster",
         VoltDBResourceConfigBuilder.class,
         new Consumer<VoltDBResourceConfigBuilder>() {
             @Override
             public void consume(VoltDBResourceConfigBuilder configurator) {
                 configurator
                   .addToServers("localhost", 12122)
                   .withClientBuilder(vcb -> builder -> {
                       builder.withMaxOutstandingTransactions(42000);
                       builder.withMaxTransactionsPerSecond(23);
                       builder.withRequestTimeout(Duration.ofSeconds(5));
                       builder.withAuthBuilder(authBuilder -> authBuilder
                          .withUsername("admin")
                          .withPassword("admin123"));
                       builder.withSslBuilder(sslBuilder -> sslBuilder
                          .withTrustStoreFile("c:/Users32/trust.me")
                          .withTrustStorePassword("got2have"));
                       builder.withRetryBuilder(retryBuilder -> retryBuilder
                          .withRetries(4)
                          .withBackoffDelay(Duration.ofSeconds(2))
                          .withMaxBackoffDelay(Duration.ofSeconds(11)));
                   })
             }
         });
voltStreamBuilder
        .consumeFromSource(...)
        .terminateWithEmitter(VoltProcedureEmitterConfig.<Event, EnhancedEvent>builder()
            .withClientReferenceName("primary-cluster")
            .withExceptionHandler(exceptionHandler)
            .withRequestMapper(event -> new Object[] { event.id() })
            .withComputeFunction((event, result) -> {
                if (result != null && result.advanceRow()) {
                    return new EnhancedEvent(event.id(), result.getString("name"));
                }
                return null;
            })
        )
        .emit(VoltProcedureTrigger.onEachResponse)
        ...

public record EnhancedEvent(int id, String name) {}

resources:
- name: primary-cluster
  voltdb-client:
    servers: localhost:12122
    client:
        maxOutstandingTransactions: 42000
        maxTransactionsPerSecond: 23
        requestTimeout: PT5S
        auth:
            username: admin
            password: admin123
        ssl:
            trustStoreFile: c:/Users32/trust.me
            trustStorePassword: got2have
        retry:
            retries: 4
            backoffDelay: 2s
            maxBackoffDelay: 11s
emitter:
    voltdb-procedure:
      voltClientResource: primary-cluster
      requestMapper: "com.example.Event::toRequest"
      computeFunction: "com.example.Event::toResult"
      triggers:
        - trigger: org.voltdb.stream.plugin.volt.api.VoltProcedureTrigger#onEachResponse
          ...

Java dependency management

Add this declaration to your dependency management system to access the configuration DSL for this plugin in Java.

MavenGradle

<dependency>
    <groupId>org.voltdb</groupId>
    <artifactId>volt-stream-plugin-volt-api</artifactId>
    <version>1.8.7</version>
</dependency>

implementation group: 'org.voltdb', name: 'volt-stream-plugin-volt-api', version: '1.8.7'

Properties

requestMapper

Maps the input event to a VoltProcedureRequest.

Java: .withRequestMapper(order -> VoltProcedureRequest.createWithParameters( "InsertOrder", order.id(), order.amount()))

YAML — the JS function may return a plain JS object matching VoltProcedureRequest's shape: requestMapper: | function process(order) { return { procedureName: "InsertOrder", parameters: [order.id, order.amount] }; }

If omitted, a default mapper accepts either a VoltProcedureRequest directly or a Map with 'procedureName' and 'parameters' entries. Type: object

computeFunction

The function computes result event for next operator in pipeline Required.

Type: object

asyncCall

Synchronously/asynchronously call to VoltDB. By default procedures are executed asynchronously. Type: boolean

Default value: true

voltClientResource

Client resource reference to be used when connecting to VoltDb cluster Required.

Type: object

exceptionHandler

A custom exception handler enabling interception of all errors related to this processor. Type: object

Available Triggers

org.voltdb.stream.api.pipeline.emitter.ExceptionTrigger.onError

When used, an emitter will create a sub stream of all events that were consumed.

org.voltdb.stream.plugin.volt.api.VoltProcedureTrigger.onEachResponse

Triggered on each response from VoltDB.

JSON Schema

You can validate or explore the configuration using its JSON Schema.

Metrics

Volt metrics

Metric enum: org.voltdb.stream.plugin.volt.metrics.VoltMetric

Prometheus name	Type	Description
voltsp_volt_affinity_reads_total	counter	Number of VoltDB affinity reads.
voltsp_volt_affinity_writes_total	counter	Number of VoltDB affinity writes.
voltsp_volt_procedure_bytes_read_bytes	counter	Number of bytes read by the VoltDB client.
voltsp_volt_procedure_bytes_written_bytes	counter	Number of bytes written by the VoltDB client.
voltsp_volt_procedure_invocation_aborts_total	counter	Number of VoltDB procedure invocation aborts.
voltsp_volt_procedure_invocation_completed_total	counter	Number of completed VoltDB procedure invocations.
voltsp_volt_procedure_invocation_errors_total	counter	Number of VoltDB procedure invocation errors.
voltsp_volt_procedure_invocation_retry_total	counter	Number of VoltDB procedure invocation retries.
voltsp_volt_procedure_invocation_timeouts_total	counter	Number of VoltDB procedure invocation timeouts.
voltsp_volt_roundrobin_reads_total	counter	Number of VoltDB round-robin reads.
voltsp_volt_roundrobin_writes_total	counter	Number of VoltDB round-robin writes.
voltsp_volt_procedure_invocation_latency_seconds	histogram	VoltDB procedure invocation latency.
voltsp_volt_sink_info	info	Metadata describing a VoltDB sink.

Volt tags

Tag enum: org.voltdb.stream.plugin.volt.metrics.VoltTag

Metrics reported by this component may include these tags. Not every metric includes every tag; tags are present only when they are relevant to the measurement.

Tag	Description
connection_id	VoltDB client connection id.
partition_id	VoltDB partition id.
procedure_name	VoltDB procedure name.
table_name	VoltDB table name.
volt_batch_size	Configured VoltDB batch size.