Actions

Actions are isolated units of business logic that perform functions within Timed Data Sources, Transforms, and Data Sinks. Actions receive required data from the DeltaFile on a queue, perform the necessary logic, and issue a response that augments the DeltaFile so that it can be handed off to the next Action in the flow or published to a topic.

Actions are currently implemented on four platforms. Core actions and several plugins are implemented in Spring Boot and utilize a Java DeltaFi development kit. Python, Go, and C++ DeltaFi development kits have also been implemented. However, Actions can be developed in any language as long as certain integration interfaces are met.

Actions may be configured differently in each flow that uses them. The Action itself defines the configuration parameters that can be used to specialize it.

Actions Must Be Thread-Safe

An action is a shared, long-lived handler. The framework keeps a single instance of each action and may invoke it concurrently on multiple threads at once — up to the action's configured concurrency. Think of it like a web request handler serving many requests in parallel: the handler is shared, the data for each request is not.

The core rule that follows: an action must not carry per-DeltaFile state across or between invocations. Everything specific to one DeltaFile arrives as method arguments (the context, input, and parameters) and should stay in local variables. Shared state held on the action (instance fields in the Java/Python/C++ kits) is only safe when it is immutable or itself thread-safe — a collaborator set up once, such as an HTTP client or a JSON mapper. Mutable state that changes per DeltaFile races as soon as concurrency is above 1 and corrupts results across DeltaFiles. Default concurrency is 1, so a stateful action appears to work until someone raises its limit.

For practical do's and don'ts — including whether to manage threads inside an action and how to tune concurrency — see Multithreading in Action Kits.

Common Action Interfaces

All actions are derived from the common Action class, which is specialized for each action type. The Action interface gives access to some common services. Extend the proper Action class (see details below) and it will be automatically discovered and loaded by the framework when your plugin is installed.

Context

Execution methods for each Java action type are passed an ActionContext. The context gives you access to information about where the action is running and the DeltaFile(s) being processed.

// the did is the DeltaFile's id
UUID did = context.getDid();
// the name of the DeltaFile, typically the original filename
String deltaFileName = context.getDeltaFileName();
// the original dataSource of this DeltaFile
String dataSource = context.getDataSource();
// the name of the flow in which the action is being invoked
String flowName = context.getFlowName();
// the id of the flow in which the action is being invoked
UUID flowId = context.getFlowId();
// name of the Action as configured in a flow
String actionName = context.getActionName();
// id of the Action in the current flow
UUID actionId = context.getActionId();
// hostname where the Action is running
String hostname = context.getHostname();
// version of Core Actions or plugin containing the Action
String actionVersion = context.getActionVersion();
// when the Action began execution
OffsetDateTime startTime = context.getStartTime();
// system name from DeltaFi System Properties
String systemName = context.getSystemName();
// the optional join configuration in effect for the action
JoinConfiguration join = context.getJoin();
// the optional joined DeltaFile ids
List<UUID> joinedDids = context.getJoinedDids();
// the optional memo field used to pass bookmarking info to timed ingress actions
String memo = context.getMemo();

Execution methods for each Python action type are passed a Context. The context gives you access to information about where the action is running, the DeltaFile(s) being processed, and access to supporting services.

class Context(NamedTuple):
  did: str
  delta_file_name: str
  data_source: str
  flow_name: str
  flow_id: str
  action_name: str
  action_id: str
  action_version: str
  hostname: str
  system_name: str
  content_service: ContentService
  join: dict = None
  joined_dids: List[str] = None
  memo: str = None
  logger: Logger = None

Go actions receive a *Deltafile which provides direct access to context fields about where the action is running and the DeltaFile being processed.

// the did is the DeltaFile's id
did := df.Did
// the name of the DeltaFile, typically the original filename
deltaFileName := df.Name
// the original dataSource of this DeltaFile
dataSource := df.DataSource
// the name of the flow in which the action is being invoked
flowName := df.FlowName
// the id of the flow in which the action is being invoked
flowID := df.FlowID
// name of the Action as configured in a flow
actionName := df.ActionName
// the optional memo field used to pass bookmarking info to timed ingress actions
memo := df.Memo

C++ actions receive an ActionContext by reference, which gives access to information about where the action is running and the DeltaFile(s) being processed.

// the did is the DeltaFile's id
auto& did = context.did;
// the name of the DeltaFile, typically the original filename
auto& deltaFileName = context.delta_file_name;
// the original dataSource of this DeltaFile
auto& dataSource = context.data_source;
// the name of the flow in which the action is being invoked
auto& flowName = context.flow_name;
// the id of the flow in which the action is being invoked
auto& flowID = context.flow_id;
// name of the Action as configured in a flow
auto& actionName = context.action_name;
// hostname where the Action is running
auto& hostname = context.hostname;
// version of Core Actions or plugin containing the Action
auto& actionVersion = context.action_version;
// when the Action began execution
auto startTime = context.start_time;
// system name from DeltaFi System Properties
auto& systemName = context.system_name;
// the optional memo field used to pass bookmarking info to timed ingress actions
auto& memo = context.memo;

Input

Each Action has a specific Input class passed to its execution method. For example, a Transform Action receives the TransformInput in the transform() method. Each Input class is unique for each Action type, typically containing:

List<ActionContent> content;
Map<String, String> metadata;

Parameters

Actions can be configured with custom parameters by extending the ActionParameters class:

package org.deltafi.parameters;

import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonPropertyDescription;
import lombok.*;
import org.deltafi.actionkit.action.parameters.ActionParameters;


@Data
@ToString(callSuper = true)
@EqualsAndHashCode(callSuper = true)
@AllArgsConstructor
@NoArgsConstructor
public class DecompressionTransformParameters extends ActionParameters {
    @JsonProperty(required = true)
    @JsonPropertyDescription("Description goes here")
    public String myParameter;
}

from pydantic import BaseModel, Field

class DecompressionTransformParameters(BaseModel):
    myParameter: str = Field(description="Description goes here")

type MyTransformParams struct {
    MyParameter string `json:"myParameter" description:"Description goes here" required:"true"`
}

struct MyTransformParams {
    std::string my_parameter;

    DELTAFI_PARAMS(MyTransformParams,
        FIELD(my_parameter, "Description goes here"))
};

Content Storage

Actions are passed or can create ActionContent that serve as pointers to content data that is stored on disk.

To retrieve content as a byte array, string, or from a stream:

byte[] byteArray = content.loadBytes();
String string = content.loadString();
String encodedString = content.loadString(Charset.forName(encoding));
InputStream inputStream = content.loadInputStream();

To retrieve content as a string or byte array in a Python action execution:

    def transform(self, context: Context, params: BaseModel, transform_input: TransformInput):
        string = transform_input.content[0].load_str()
        bytes = transform_input.content[0].load_bytes()

To retrieve content as a byte array or string in a Go action execution:

func (a *MyAction) Transform(df *actionkit.Deltafile, params MyParams) *actionkit.Deltafile {
    content, _ := df.FirstContent()
    data, err := content.LoadString()
    bytes, err := content.LoadBytes()
}

To retrieve content as a byte array, string, or stream in a C++ action execution:

auto bytes = input.first_content().load_bytes();     // vector<uint8_t>
auto str = input.first_content().load_string();       // string
auto reader = input.first_content().load_reader();    // unique_ptr<istream>

To store content from a byte array or a stream and add to a Result:

// create a result of the type appropriate for your Action
TransformResult transformResult = new TransformResult();
transformResult.saveContent(byteArray, fileName, MediaType.APPLICATION_JSON);
transformResult.saveContent(inputStream, fileName, MediaType.APPLICATION_JSON);

// you can modify existing content and add it to the Result without having to save new Content to disk:
List<ActionContent> existingContentList = input.content();
transformResult.addContent(existingContentList);

// you can also manipulate existing content to store new content on disk
ActionContent copyOfFirstContent = existingContent.copy();
copyOfFirstContent.setName("new-name.txt");
// get the first 50 bytes
ActionContent partOfSecondContent = anotherContent.subcontent(0, 50);
copyOfFirstContent.append(partOfSecondContent);
// store the pointers to the stitched-together content without writing to disk
transformResult.addContent(copyOfFirstContent);

Or in Python save content to a Result:

result.save_content(data, content_name, media_type)

Or in Go save content via the Deltafile:

// Save content directly through the Deltafile (stores to MinIO and adds to output)
df.SaveStringContent(fileName, "text/plain", data)

// You can also manipulate existing content without saving to disk:
copied := existingContent.Copy()
copied.SetName("new-name.txt")
partial := anotherContent.Subcontent(0, 50)
copied.Append(partial)
df.AddContent(copied)

Or in C++ save content to a Result:

// Save directly via the result (stores to S3 and adds content)
result.save_content(data_string, "output.txt", "text/plain");

// Stream large content into a writer callback (never buffers the whole object in memory):
result.save_content_from_writer(
    [](std::ostream& out) { write_records(out); }, "output.csv", "text/csv");

// Manipulate existing content without saving to disk:
auto copied = existing_content.copy();
copied.set_name("new-name.txt");
auto partial = another_content.subcontent(0, 50);
copied.append(partial);
result.add_content(copied);

load_reader (above) and save_content_from_writer stream content in bounded-memory chunks, so large objects never fully materialize in memory — the store uses a multipart upload above its part size.

Results

Each Action returns a specific Result class from its execution method. The Result contains some combination of content, metadata, and annotations produced by the execution of that Action.

Actions may return an ErrorResult if something goes wrong. Errors terminate the flow and raise the error cause to an operator's attention for possible retry.

// return with a custom error message
if (somethingWentWrong) {
    return new ErrorResult(context, "Description of why the Action failed");
}

try {
    // something bad happens
} catch (SomeException e) {
    // return with Exception details
    return new ErrorResult(context, e.getMessage(), e.getCause());
}

Sometimes you want to halt a flow but not raise an error. In this case use a FilterResult:

return new FilterResult(context, "Description of why this DeltaFile was filtered");

or

return new FilterResult(context, "Common summary reason of why this DeltaFile was filtered", "Detailed reason");

In Go, use the Deltafile's error and filter methods:

// return an error
return df.Errorf("Description of failure").SetErrorContext("additional context")

// return a filter
return df.Filterf("Description of why filtered")

In C++, result types are returned as variants:

// return an error
return deltafi::ErrorResult(context, "Description of failure")
    .set_context("additional context");

// return a filter
return deltafi::FilterResult(context, "Description of why filtered");

Join

Transform actions may be configured to join multiple DeltaFiles before executing the transform method. When a transform action is configured for joining, DeltaFi will collect a batch of DeltaFiles until the join criteria is met. Once the criteria is met, the whole batch is sent to the Transform action as one execution, with one input per collected DeltaFile. Only transform actions can be joined; a data sink or data source that configures a join is rejected as an invalid flow.

Most actions do not need to support joining. Put the JoinDeltaFiles action at the point in the flow where the batch should be collected and configure the join on it: it collapses the collected DeltaFiles into one, keeping all of their content and combining their metadata. Every action after it sees a single ordinary DeltaFile, so it needs no join support of its own. Use Merge instead when the collected content should also be concatenated into a single content item.

- name: JoinDeltaFiles
  type: org.deltafi.core.action.join.JoinDeltaFiles
  join:
    maxAge: PT10M
    maxNum: 100
  parameters:
    metadataMerge: DISTINCT
- name: JoltTransform      # runs once on the whole batch, and knows nothing about joins
  type: org.deltafi.core.action.jolt.JoltTransform

An action implements joining itself only when it needs to see the collected DeltaFiles individually — because it combines their content in a way of its own, or produces a result that depends on which DeltaFile each piece came from.

In Java, an action that may be joined extends JoinTransform instead of TransformAction. Its transform method always takes a list of inputs, one per joined DeltaFile in join order. An execution that is not joined receives a list holding its single input, so a join action has exactly one transform method to write and it reads the same either way.

Most actions want the inputs combined into one, which is what JoinMerger.merge does. It keeps all content in join order and merges all metadata, with the last value found for a key winning:

@Component
public class MyJoiningAction extends JoinTransform<MyParams> {
    public MyJoiningAction() {
        super("Joins DeltaFiles");
    }

    @Override
    public TransformResultType transform(@NotNull ActionContext context, @NotNull MyParams params,
            @NotNull List<TransformInput> transformInputs) {
        TransformInput input = JoinMerger.merge(transformInputs);
        // the combined content and metadata
    }
}

JoinMerger resolves a repeated metadata key by MetadataMerge: LAST (the default), FIRST, joining every value (ALL) or the unique values (DISTINCT) with a delimiter, or DROP to keep no metadata at all. A value containing the delimiter cannot be told apart from two values, so choose one the values will not contain:

        TransformInput input = JoinMerger.merge(transformInputs, MetadataMerge.DISTINCT, params.getDelimiter());

JoinMerger.mergeMetadata returns only the merged metadata, for an action that combines the content itself:

        return TransformInput.builder()
                .content(myContentCombiner(transformInputs))
                .metadata(JoinMerger.mergeMetadata(transformInputs, MetadataMerge.ALL))
                .build();

An action does not have to combine the inputs at all. Working with them individually lets it produce a result the combined input could not describe, such as one child DeltaFile per joined input:

    @Override
    public TransformResultType transform(@NotNull ActionContext context, @NotNull MyParams params,
            @NotNull List<TransformInput> transformInputs) {
        TransformResults results = new TransformResults(context);
        for (TransformInput transformInput : transformInputs) {
            results.add(new ChildTransformResult(context, transformInput.getContent()));
        }
        return results;
    }

The inputs arrive in join order, which is the order DeltaFiles were collected rather than the order they were created. context.getJoinedDids() is positionally aligned with the inputs, so an action that needs creation order can sort by those ids.

To end the execution without producing a transform result, throw an ErrorResultException or a FilterResultException. The action kit converts either one into the matching result, preserving any annotations and metrics on the exception:

    @Override
    public TransformResultType transform(@NotNull ActionContext context, @NotNull MyParams params,
            @NotNull List<TransformInput> transformInputs) {
        if (transformInputs.size() < params.getRequired()) {
            throw new FilterResultException("Not enough DeltaFiles to process");
        }
        // ...
    }

Actions that implement the deprecated Join interface keep working: their inputs are combined by join and the result is passed to the transform method. Join is removed in 3.0, so extend JoinTransform instead.

Other kits provide the join as a separate method that combines the inputs before the transform method runs:

    def join(self, transform_inputs: List[TransformInput]):
        all_content = []
        for transform_input in transform_inputs:
            all_content += transform_input.content
        return TransformInput(content=all_content)

func (a *MyJoiningAction) Join(inputs []*actionkit.Deltafile, joined *actionkit.Deltafile, params MyParams) *actionkit.Deltafile {
    for _, input := range inputs {
        joined.AddContentList(input.GetContentList())
        joined.AddMetadataMap(input.GetMetadataMap())
    }
    return joined
}

deltafi::TransformInput join(std::vector<deltafi::TransformInput> inputs) {
    std::vector<deltafi::ActionContent> all_content;
    for (auto& input : inputs) {
        all_content.insert(all_content.end(),
            std::make_move_iterator(input.content.begin()),
            std::make_move_iterator(input.content.end()));
    }
    return deltafi::TransformInput{std::move(all_content), {}};
}

The join configuration that defines the criteria for collecting DeltaFiles that will be joined includes the following fields:

• maxAge (required) the maximum duration (ISO 8601) to wait after the first DeltaFile is received for a collection before the action is executed
• minNum the minimum number of DeltaFiles to collect within maxAge. If this number is not reached, all collected DeltaFiles will have the action marked in error.
• maxNum the maximum number of DeltaFiles to collect before the action is executed
• metadataKey an optional metadata key used to get the value to group collections by (defaults to collecting all)

SSL Setup

Java SSLContext

The deltafi-action-kit will autoconfigure a SslContextProvider bean which is available for injection. The provider will contain a populated SslBundle when the certs directory is fully populated (see Plugins SSL Config). The action-kit also provides a HttpClient bean that is preconfigured with an SSLContext when the SslContextProvider is configured.

To get a new SSlContext backed by the files in /cert, inject the SslContextProvider bean and call the createSslContext method. For more advanced use cases the SslContextProvider provides direct access to the SslBundle (getSslBundle) and private key (getPrivateKey).

Python

Python actions can access the files necessary to configure SSL in the /certs directory.

Action Pages

• Timed Ingress Action
• Transform Action
• Egress Action