Testing with TestComputationHarness in YTsaurus Flow (Java)

The flow-test-utils module provides utilities for unit testing Flow pipeline components in Java and Kotlin without running a real gRPC server or C++ workers. The central class is TestComputationHarness, which emulates doProcess calls at the companion level.

Note

The TestComputationHarness class is designed for unit testing individual ProcessFunction instances without launching a real pipeline. For integration testing of the full pipeline (with real C++ workers, queues, and streams), use FlowTestJavaBase — see the section below.

General testing architecture

In production, the C++ worker sends gRPC requests to the companion, passing messages, timers, state, and watermarks. The Companion calls Computation.doProcess(), which delegates processing to the ProcessFunction.

In tests, TestComputationHarness replaces the gRPC layer: it accepts TestDoProcessRequest, converts it to protobuf format, calls CompanionRequestProcessor.processBatch(), and returns TestDoProcessResponse with deserialized results.

Dependencies

To use the test utilities, you need to add a dependency on flow-test-utils:

PEERDIR(
    yt/java/flow/flow-test-utils
)

All examples below use JUnit version 5 or later.

Setting up TestComputationHarness

To create a TestComputationHarness, you need:

PipelineContext — the pipeline context with registered Computation objects and streams.
Pipeline spec — the static pipeline specification in YSON format (the pipeline.yson file).
External state schemas (optional) — schemas for external states if the computation uses ExternalStateAccessor.

Builder API

Java

Kotlin

TestComputationHarness harness = TestComputationHarness.builder()
        .setPipelineContext(pipelineContext)       // required
        .setPipelineSpec(txtSpec)                  // required: String, YTreeNode, or InputStream
        .addExternalStateSchema("state-name", schema) // optional
        .setJobContext(jobContext)                 // optional, default timeout=10 minutes
        .build();

val harness = TestComputationHarness.builder()
        .setPipelineContext(pipelineContext)       // required
        .setPipelineSpec(txtSpec)                  // required: String, YTreeNode, or InputStream
        .addExternalStateSchema("state-name", schema) // optional
        .setJobContext(jobContext)                 // optional, default timeout=10 minutes
        .build()

When you call build(), the harness automatically:

Extracts stream information from the pipeline spec.
Registers missing streams as untyped (via FlowStreams.raw) in the PipelineContext.
Creates an internal CompanionRequestProcessor.

Building test requests

TestDoProcessRequest

TestDoProcessRequest is a request to process a batch of messages and/or timers. You create it using a builder:

Java

Kotlin

var request = TestDoProcessRequest.builder("computation-id")
        .setMessages(messages)           // List<ExtendedMessage>
        .setTimers(timers)               // List<Timer>
        .setWatermarks(watermarks)       // Map<String, Long>
        .setExternalState("name", stateMap)  // Map<Payload, ExternalState>
        .setInternalState("name", stateMap)  // Map<Payload, InternalState>
        .build();

val request = TestDoProcessRequest.builder("computation-id")
        .setMessages(messages)           // List<ExtendedMessage>
        .setTimers(timers)               // List<Timer>
        .setWatermarks(watermarks)       // Map<String, Long>
        .setExternalState("name", stateMap)  // Map<Payload, ExternalState>
        .setInternalState("name", stateMap)  // Map<Payload, InternalState>
        .build()

Parameters:

computationId — the ID of the computation that the request targets.
messages — a list of input messages of type ExtendedMessage.
timers — a list of triggered timers of type Timer.
watermarks — a map of watermarks by stream (streamId → timestamp).
externalStates — pre-filled external state.
internalStates — pre-filled internal state.

Building messages

With typed streams

If you register streams as typed (via FlowStreams.typed(...)), you can pass the payload as a POJO object:

Java

Kotlin

ExtendedMessage msg = ExtendedMessage.builder()
        .setStreamId("hit")
        .setKey(key)
        .setPayload(new Hit("hit-1", 1000L, "payload-data"))
        .setEventTimestamp(1000L)
        .build();

val msg = ExtendedMessage.builder()
        .setStreamId("hit")
        .setKey(key)
        .setPayload(Hit("hit-1", 1000L, "payload-data"))
        .setEventTimestamp(1000L)
        .build()

With untyped streams (PayloadBuilder)

If streams aren’t typed, you build the payload using PayloadBuilder:

Java

Kotlin

var hitSchema = testHarness.getStream("hit").getSchema();
var payload = new PayloadBuilder(hitSchema)
        .set("hit_id", "hit-1")
        .set("hit_time", 1000L)
        .set("hit_payload", "payload-data")
        .finish();

ExtendedMessage msg = ExtendedMessage.builder()
        .setStreamId("hit")
        .setKey(key)
        .setPayload(payload)
        .setEventTimestamp(1000L)
        .build();

val hitSchema = testHarness.getStream("hit").getSchema()
val payload = PayloadBuilder(hitSchema)
        .set("hit_id", "hit-1")
        .set("hit_time", 1000L)
        .set("hit_payload", "payload-data")
        .finish()

val msg = ExtendedMessage.builder()
        .setStreamId("hit")
        .setKey(key)
        .setPayload(payload)
        .setEventTimestamp(1000L)
        .build()

Note

The testHarness.getStream(streamId) method returns a FlowStream<?> with the schema extracted from the pipeline spec. This is convenient for getting the stream schema without defining it manually.

Building keys

You build the key (Payload) according to the group_by_schema of the Computation you’re testing, from the pipeline spec.

Note

The testHarness.getGroupBySchema(computationId) method returns a TableSchema based on the group_by_schema extracted from the pipeline spec for the Computation you’re testing.

Java

Kotlin

TableSchema keySchema = testHarness.getGroupBySchema("join");

Payload key = new PayloadBuilder(keySchema)
        .set("hash", 0L)
        .set("hit_id", "hit-1")
        .set("hit_time", 1000L)
        .finish();

val keySchema = testHarness.getGroupBySchema("join")

val key = PayloadBuilder(keySchema)
        .set("hash", 0L)
        .set("hit_id", "hit-1")
        .set("hit_time", 1000L)
        .finish()

Tip

The hash field (farm_hash) is calculated on the C++ worker side. In tests, you can set it to 0L.

Building timers

You create a timer using a builder:

Java

Kotlin

Timer timer = Timer.builder()
        .setStreamId("timer")
        .setKey(key)
        .setTriggerTimestamp(triggerTimestamp)
        .build();

val timer = Timer.builder()
        .setStreamId("timer")
        .setKey(key)
        .setTriggerTimestamp(triggerTimestamp)
        .build()

Configuring watermarks

Watermarks define which messages are considered late. A message with eventTimestamp < watermark will be discarded if the ProcessFunction implements the corresponding check via ctx.getEpochInputEventWatermark(). See an example of this check in Wait Click Join in Java.

Java

Kotlin

Map<String, Long> watermarks = Map.of(
        "hit", 0L,
        "action", 0L
);

val watermarks = mapOf(
        "hit" to 0L,
        "action" to 0L
)

Setting the watermark to 0L means that all messages with eventTimestamp >= 0 won’t be considered late.

Pre-populating state

External State

Java

Kotlin

TableSchema stateSchema = TableSchema.builder()
        .addValue("hit_payload", ColumnValueType.STRING)
        .addValue("show_time", ColumnValueType.UINT64)
        .addValue("click_time", ColumnValueType.UINT64)
        .build();

ExternalState preState = new ExternalState(
        new PayloadBuilder(stateSchema)
                .set("hit_payload", "some-payload")
                .set("show_time", showTime)
                .set("click_time", clickTime)
                .finish()
);

var request = TestDoProcessRequest.builder("join")
        .setTimers(List.of(timer))
        .setExternalState("join-state", Map.of(key, preState))
        .build();

val stateSchema = TableSchema.builder()
        .addValue("hit_payload", ColumnValueType.STRING)
        .addValue("show_time", ColumnValueType.UINT64)
        .addValue("click_time", ColumnValueType.UINT64)
        .build()

val preState = ExternalState(
        PayloadBuilder(stateSchema)
                .set("hit_payload", "some-payload")
                .set("show_time", showTime)
                .set("click_time", clickTime)
                .finish()
)

val request = TestDoProcessRequest.builder("join")
        .setTimers(listOf(timer))
        .setExternalState("join-state", mapOf(key to preState))
        .build()

Internal State (Proto)

For internal state based on protobuf:

Java

Kotlin

TJoinState protoState = TJoinState.newBuilder()
        .setHitPayload("some-payload")
        .setShowTime(showTime)
        .setClickTime(clickTime)
        .build();

InternalState internalState = new InternalState(protoState.toByteArray());

var request = TestDoProcessRequest.builder("join")
        .setTimers(List.of(timer))
        .setInternalState("join-state", Map.of(key, internalState))
        .build();

val protoState = TJoinState.newBuilder()
        .setHitPayload("some-payload")
        .setShowTime(showTime)
        .setClickTime(clickTime)
        .build()

val internalState = InternalState(protoState.toByteArray())

val request = TestDoProcessRequest.builder("join")
        .setTimers(listOf(timer))
        .setInternalState("join-state", mapOf(key to internalState))
        .build()

Analyzing the results

TestDoProcessResponse

TestDoProcessResponse provides methods to verify processing results.

Method	Description
`getOutputMessagesFlatten()`	All output messages from all transform groups
`getOutputTimersFlatten()`	All set timers from all transform groups
`getTransformResults()`	A list of `TransformResult` grouped by parent IDs

Note

TestDoProcessResponse provides two sets of methods to access the state:

Full view (getExternalState(s), getInternalState(s)) — all states after processing: the keys passed in the request, with computation changes applied on top. Use this when you need to check the final state regardless of whether computation changed it.
Only modified (getModifiedExternalState(s), getModifiedInternalState(s)) — only what computation actually changed. These are the states sent back to the worker and saved. Use this to verify that the state for a key hasn’t changed, or when you need to check only the state changes.

All state access methods are null-safe: an unknown state name or key returns an empty map or null, not an exception.

Full view of the state (loaded states with computation changes applied):

Method	Description
`getExternalStateNames()` / `getInternalStateNames()`	Names of all states (loaded and/or modified)
`getExternalStates()` / `getInternalStates()`	All states grouped by name
`getExternalStates(name)` / `getInternalStates(name)`	All state records by name (empty map if none)
`getExternalStateSize(name)` / `getInternalStateSize(name)`	Number of records in the state (0 if none)
`getExternalState(name, key)` / `getInternalState(name, key)`	Specific state record (`null` if none)

Only states modified by computation:

Method	Description
`getModifiedExternalStateNames()` / `getModifiedInternalStateNames()`	Names of modified states
`getModifiedExternalStates()` / `getModifiedInternalStates()`	Modified states grouped by name
`getModifiedExternalStates(name)` / `getModifiedInternalStates(name)`	Modified state records by name (empty map if none)
`getModifiedExternalStateSize(name)` / `getModifiedInternalStateSize(name)`	Number of modified records (0 if none)
`getModifiedExternalState(name, key)` / `getModifiedInternalState(name, key)`	Specific modified record (`null` if not modified)

Checking output messages

Java

Kotlin

var response = testHarness.doProcess(request);

// Check the number of output messages
assertEquals(1, response.getOutputMessagesFlatten().size());

// Get the message and check the stream
var msg = response.getOutputMessagesFlatten().get(0);
assertEquals("joined_action", msg.getStreamId());

// For typed streams — cast payload to POJO
JoinedAction result = (JoinedAction) msg.getPayload();
assertEquals("hit-1", result.getHitId());

// For untyped streams — read fields via get()
byte[] data = msg.get("data", byte[].class);

val response = testHarness.doProcess(request)

// Check the number of output messages
assertEquals(1, response.getOutputMessagesFlatten().size)

// Get the message and check the stream
val msg = response.getOutputMessagesFlatten()[0]
assertEquals("joined_action", msg.getStreamId())

// For typed streams — cast payload to POJO
val result = msg.getPayload() as JoinedAction
assertEquals("hit-1", result.getHitId())

// For untyped streams — read fields via get()
val data = msg.get("data", ByteArray::class.java)

Checking timers

Java

Kotlin

assertEquals(1, response.getOutputTimersFlatten().size());
var timer = response.getOutputTimersFlatten().get(0);
assertEquals(expectedTriggerTimestamp, timer.getTriggerTimestamp());

assertEquals(1, response.getOutputTimersFlatten().size)
val timer = response.getOutputTimersFlatten()[0]
assertEquals(expectedTriggerTimestamp, timer.getTriggerTimestamp())

Checking the state

Java

Kotlin

// External state
assertEquals(1, response.getExternalStateSize("join-state"));
var state = response.getExternalState("join-state", key);
assertFalse(state.isReset());
assertEquals("payload-data", state.getValue().get("hit_payload", String.class));

// Check state reset
var stateAfterTimer = response.getExternalState("join-state", key);
assertTrue(stateAfterTimer.isReset());

// External state
assertEquals(1, response.getExternalStateSize("join-state"))
val state = response.getExternalState("join-state", key)!!
assertFalse(state.isReset)
assertEquals("payload-data", state.value.get("hit_payload", String::class.java))

// Check state reset
val stateAfterTimer = response.getExternalState("join-state", key)!!
assertTrue(stateAfterTimer.isReset)

Java

Kotlin

// Internal state (Proto)
var internalState = response.getInternalState("join-state", key);
assertFalse(internalState.isReset());
TJoinState joinState = ProtoUtils.parseBytes(internalState.getValue(), TJoinState.class);
assertEquals("payload-data", joinState.getHitPayload());

// Internal state (Proto)
val internalState = response.getInternalState("join-state", key)!!
assertFalse(internalState.isReset)
val joinState = ProtoUtils.parseBytes(internalState.value, TJoinState::class.java)
assertEquals("payload-data", joinState.getHitPayload())

Full view vs. modified states

The getExternalState / getInternalState methods return the full view: the pre-filled state is available even if computation didn’t change it. To check what computation actually changed, use the getModified* methods.

Java

Kotlin

// Full view: the loaded state is visible even if it wasn’t changed
assertEquals(1, response.getExternalStateSize("join-state"));
assertNotNull(response.getExternalState("join-state", key));

// Only modified states: computation wrote nothing for this key
assertEquals(0, response.getModifiedExternalStateSize("join-state"));
assertNull(response.getModifiedExternalState("join-state", key));

// Unknown name/key doesn’t throw an exception
assertNull(response.getExternalState("unknown", key));
assertTrue(response.getExternalStates("unknown").isEmpty());

// Full view: the loaded state is visible even if it wasn’t changed
assertEquals(1, response.getExternalStateSize("join-state"))
assertNotNull(response.getExternalState("join-state", key))

// Only modified states: computation wrote nothing for this key
assertEquals(0, response.getModifiedExternalStateSize("join-state"))
assertNull(response.getModifiedExternalState("join-state", key))

// Unknown name/key doesn’t throw an exception
assertNull(response.getExternalState("unknown", key))
assertTrue(response.getExternalStates("unknown").isEmpty())

Example: a test without Spring

This is a complete unit test example for JoinProcessFunction from the wait_click_join project. In this variant, you create PipelineContext manually, without a Spring container.

Java

Kotlin

public class JoinProcessFunctionTest {

    private static final long WAIT_SECONDS = 10L;
    private static final long BASE_HIT_TIME = 1000L;
    private static final long WATERMARK = 0L;

    private TestComputationHarness testHarness;
    private TableSchema keySchema;
    private TableSchema joinStateSchema;

    @BeforeEach
    void init() throws IOException {
        // 1. You create PipelineContext manually
        var pipelineContext = new PipelineContext();

        // 2. Register the computation with the process function
        Computation join = Computation.builder()
                .setComputationId("join")
                .setProcessFunction(new JoinProcessFunction())
                .build();
        pipelineContext.registerComputation(join);

        // 3. Register typed streams
        pipelineContext.registerStream(FlowStreams.typed("hit", Hit.class));
        pipelineContext.registerStream(FlowStreams.typed("action", Action.class));
        pipelineContext.registerStream(FlowStreams.typed("joined_action", JoinedAction.class));

        // 4. Define the external state schema
        this.joinStateSchema = TableSchema.builder()
                .addValue("hit_payload", ColumnValueType.STRING)
                .addValue("show_time", ColumnValueType.UINT64)
                .addValue("click_time", ColumnValueType.UINT64)
                .build();

        // 5. Read the pipeline spec and create the harness
        var specPath = Paths.getSourcePath(
                "yt/yt/flow/examples/java/wait_click_join/test/pipeline.yson");
        var txtSpec = Files.readString(Path.of(specPath));

        this.testHarness = TestComputationHarness.builder()
                .setPipelineContext(pipelineContext)
                .setPipelineSpec(txtSpec)
                .addExternalStateSchema("join-state", joinStateSchema)
                .build();

        // 6. Get the key schema for the join computation (group_by_schema from pipeline.yson)
        this.keySchema = testHarness.getGroupBySchema("join");
    }

    // — Helper methods —

    private Payload buildKey(String hitId, long hitTime) {
        return new PayloadBuilder(keySchema)
                .set("hash", 0L)
                .set("hit_id", hitId)
                .set("hit_time", hitTime)
                .finish();
    }

    private ExtendedMessage buildHitMessage(String hitId, long hitTime, String hitPayload) {
        return ExtendedMessage.builder()
                .setStreamId("hit")
                .setKey(buildKey(hitId, hitTime))
                .setPayload(new Hit(hitId, hitTime, hitPayload))
                .setEventTimestamp(hitTime)
                .build();
    }

    private Map<String, Long> defaultWatermarks() {
        return Map.of("hit", WATERMARK, "action", WATERMARK);
    }

    // — Tests —

    @Test
    void testHitMessageStoresHitPayloadInState() {
        var messages = List.of(buildHitMessage("hit-1", BASE_HIT_TIME, "payload-data"));

        var request = TestDoProcessRequest.builder("join")
                .setMessages(messages)
                .setWatermarks(defaultWatermarks())
                .build();

        var response = testHarness.doProcess(request);

        // There are no output messages — the result will appear when the timer fires
        assertTrue(response.getOutputMessagesFlatten().isEmpty());

        // The timer must be set
        assertEquals(1, response.getOutputTimersFlatten().size());
        assertEquals(BASE_HIT_TIME + WAIT_SECONDS,
                response.getOutputTimersFlatten().get(0).getTriggerTimestamp());

        // The state must contain hit_payload
        Payload key = buildKey("hit-1", BASE_HIT_TIME);
        var state = response.getExternalState("join-state", key);
        assertFalse(state.isReset());
        assertEquals("payload-data", state.getValue().get("hit_payload", String.class));
    }

    @Test
    void testTimerEmitsJoinedAction() {
        Payload key = buildKey("hit-10", BASE_HIT_TIME);

        // Pre-populate the state
        ExternalState preState = new ExternalState(
                new PayloadBuilder(joinStateSchema)
                        .set("hit_payload", "some-payload")
                        .set("show_time", BASE_HIT_TIME + 3L)
                        .set("click_time", BASE_HIT_TIME + 7L)
                        .finish()
        );

        Timer timer = Timer.builder()
                .setStreamId("timer")
                .setKey(key)
                .setTriggerTimestamp(BASE_HIT_TIME + WAIT_SECONDS)
                .build();

        var request = TestDoProcessRequest.builder("join")
                .setTimers(List.of(timer))
                .setExternalState("join-state", Map.of(key, preState))
                .build();

        var response = testHarness.doProcess(request);

        // There must be one output message
        assertEquals(1, response.getOutputMessagesFlatten().size());
        var msg = response.getOutputMessagesFlatten().get(0);
        assertEquals("joined_action", msg.getStreamId());

        JoinedAction result = (JoinedAction) msg.getPayload();
        assertEquals("hit-10", result.getHitId());
        assertTrue(result.getClick());

        // The state must be reset
        assertTrue(response.getExternalState("join-state", key).isReset());
    }

    @Test
    void testLateMessageIsDropped() {
        long watermark = BASE_HIT_TIME + 5L;
        long eventTimestamp = BASE_HIT_TIME + 3L; // < watermark → late

        var messages = List.of(buildActionMessage("hit-5", BASE_HIT_TIME,
                BASE_HIT_TIME + 2L, false, eventTimestamp));

        var request = TestDoProcessRequest.builder("join")
                .setMessages(messages)
                .setWatermarks(Map.of("hit", watermark, "action", watermark))
                .build();

        var response = testHarness.doProcess(request);

        // The late message is dropped
        assertTrue(response.getOutputMessagesFlatten().isEmpty());
        assertTrue(response.getOutputTimersFlatten().isEmpty());
        assertEquals(0, response.getExternalStateSize("join-state"));
    }
}

class JoinProcessFunctionTest {

    companion object {
        private const val WAIT_SECONDS = 10L
        private const val BASE_HIT_TIME = 1000L
        private const val WATERMARK = 0L
    }

    private lateinit var testHarness: TestComputationHarness
    private lateinit var keySchema: TableSchema
    private lateinit var joinStateSchema: TableSchema

    @BeforeEach
    fun init() {
        // 1. You create PipelineContext manually
        val pipelineContext = PipelineContext()

        // 2. Register the computation with the process function
        val join = Computation.builder()
                .setComputationId("join")
                .setProcessFunction(JoinProcessFunction())
                .build()
        pipelineContext.registerComputation(join)

        // 3. Register typed streams
        pipelineContext.registerStream(FlowStreams.typed("hit", Hit::class.java))
        pipelineContext.registerStream(FlowStreams.typed("action", Action::class.java))
        pipelineContext.registerStream(FlowStreams.typed("joined_action", JoinedAction::class.java))

        // 4. Define the external state schema
        joinStateSchema = TableSchema.builder()
                .addValue("hit_payload", ColumnValueType.STRING)
                .addValue("show_time", ColumnValueType.UINT64)
                .addValue("click_time", ColumnValueType.UINT64)
                .build()

        // 5. Read the pipeline spec and create the harness
        val specPath = Paths.getSourcePath(
                "yt/yt/flow/examples/java/wait_click_join/test/pipeline.yson")
        val txtSpec = Files.readString(Path.of(specPath))

        testHarness = TestComputationHarness.builder()
                .setPipelineContext(pipelineContext)
                .setPipelineSpec(txtSpec)
                .addExternalStateSchema("join-state", joinStateSchema)
                .build()

        // 6. Get the key schema for the join computation (group_by_schema from pipeline.yson)
        keySchema = testHarness.getGroupBySchema("join")
    }

    // — Helper methods —

    private fun buildKey(hitId: String, hitTime: Long): Payload =
        PayloadBuilder(keySchema)
                .set("hash", 0L)
                .set("hit_id", hitId)
                .set("hit_time", hitTime)
                .finish()

    private fun buildHitMessage(hitId: String, hitTime: Long, hitPayload: String): ExtendedMessage =
        ExtendedMessage.builder()
                .setStreamId("hit")
                .setKey(buildKey(hitId, hitTime))
                .setPayload(Hit(hitId, hitTime, hitPayload))
                .setEventTimestamp(hitTime)
                .build()

    private fun defaultWatermarks() = mapOf("hit" to WATERMARK, "action" to WATERMARK)

    // — Tests —

    @Test
    fun testHitMessageStoresHitPayloadInState() {
        val messages = listOf(buildHitMessage("hit-1", BASE_HIT_TIME, "payload-data"))

        val request = TestDoProcessRequest.builder("join")
                .setMessages(messages)
                .setWatermarks(defaultWatermarks())
                .build()

        val response = testHarness.doProcess(request)

        // There are no output messages — the result will appear when the timer fires
        assertTrue(response.getOutputMessagesFlatten().isEmpty())

        // The timer must be set
        assertEquals(1, response.getOutputTimersFlatten().size)
        assertEquals(BASE_HIT_TIME + WAIT_SECONDS,
                response.getOutputTimersFlatten()[0].getTriggerTimestamp())

        // The state must contain hit_payload
        val key = buildKey("hit-1", BASE_HIT_TIME)
        val state = response.getExternalState("join-state", key)!!
        assertFalse(state.isReset)
        assertEquals("payload-data", state.value.get("hit_payload", String::class.java))
    }

    @Test
    fun testTimerEmitsJoinedAction() {
        val key = buildKey("hit-10", BASE_HIT_TIME)

        // Pre-populate the state
        val preState = ExternalState(
                PayloadBuilder(joinStateSchema)
                        .set("hit_payload", "some-payload")
                        .set("show_time", BASE_HIT_TIME + 3L)
                        .set("click_time", BASE_HIT_TIME + 7L)
                        .finish()
        )

        val timer = Timer.builder()
                .setStreamId("timer")
                .setKey(key)
                .setTriggerTimestamp(BASE_HIT_TIME + WAIT_SECONDS)
                .build()

        val request = TestDoProcessRequest.builder("join")
                .setTimers(listOf(timer))
                .setExternalState("join-state", mapOf(key to preState))
                .build()

        val response = testHarness.doProcess(request)

        // There must be one output message
        assertEquals(1, response.getOutputMessagesFlatten().size)
        val msg = response.getOutputMessagesFlatten()[0]
        assertEquals("joined_action", msg.getStreamId())

        val result = msg.getPayload() as JoinedAction
        assertEquals("hit-10", result.getHitId())
        assertTrue(result.getClick())

        // The state must be reset
        assertTrue(response.getExternalState("join-state", key)!!.isReset)
    }

    @Test
    fun testLateMessageIsDropped() {
        val watermark = BASE_HIT_TIME + 5L
        val eventTimestamp = BASE_HIT_TIME + 3L // < watermark → late

        val messages = listOf(buildActionMessage("hit-5", BASE_HIT_TIME,
                BASE_HIT_TIME + 2L, false, eventTimestamp))

        val request = TestDoProcessRequest.builder("join")
                .setMessages(messages)
                .setWatermarks(mapOf("hit" to watermark, "action" to watermark))
                .build()

        val response = testHarness.doProcess(request)

        // The late message is dropped
        assertTrue(response.getOutputMessagesFlatten().isEmpty())
        assertTrue(response.getOutputTimersFlatten().isEmpty())
        assertEquals(0, response.getExternalStateSize("join-state"))
    }
}

Multi-step test

To test the full processing flow (hit → show → click → timer), you manually pass the state between steps:

Java

Kotlin

@Test
void testFullJoinFlow() {
    String hitId = "hit-20";
    long hitTime = BASE_HIT_TIME;

    // Step 1: process the hit message
    var hitRequest = TestDoProcessRequest.builder("join")
            .setMessages(List.of(buildHitMessage(hitId, hitTime, "full-payload")))
            .setWatermarks(defaultWatermarks())
            .build();
    var hitResponse = testHarness.doProcess(hitRequest);

    Payload key = buildKey(hitId, hitTime);
    var stateAfterHit = hitResponse.getExternalState("join-state", key);

    // Step 2: process the show message, pass the state from step 1
    var showRequest = TestDoProcessRequest.builder("join")
            .setMessages(List.of(buildActionMessage(hitId, hitTime, hitTime + 3L, false, hitTime + 1L)))
            .setExternalState("join-state", Map.of(key, stateAfterHit))
            .setWatermarks(defaultWatermarks())
            .build();
    var showResponse = testHarness.doProcess(showRequest);

    var stateAfterShow = showResponse.getExternalState("join-state", key);

    // Step 3: process the click message
    var clickRequest = TestDoProcessRequest.builder("join")
            .setMessages(List.of(buildActionMessage(hitId, hitTime, hitTime + 7L, true, hitTime + 2L)))
            .setExternalState("join-state", Map.of(key, stateAfterShow))
            .setWatermarks(defaultWatermarks())
            .build();
    var clickResponse = testHarness.doProcess(clickRequest);

    var stateAfterClick = clickResponse.getExternalState("join-state", key);

    // Step 4: trigger the timer
    Timer timer = Timer.builder()
            .setStreamId("timer")
            .setKey(key)
            .setTriggerTimestamp(hitTime + WAIT_SECONDS)
            .build();
    var timerRequest = TestDoProcessRequest.builder("join")
            .setTimers(List.of(timer))
            .setExternalState("join-state", Map.of(key, stateAfterClick))
            .build();
    var timerResponse = testHarness.doProcess(timerRequest);

    // Check the final result
    assertEquals(1, timerResponse.getOutputMessagesFlatten().size());
    JoinedAction result = timerResponse.getOutputMessagesFlatten().get(0).getPayload();
    assertEquals(hitId, result.getHitId());
    assertTrue(result.getClick());

    assertTrue(timerResponse.getExternalState("join-state", key).isReset());
}

@Test
fun testFullJoinFlow() {
    val hitId = "hit-20"
    val hitTime = BASE_HIT_TIME

    // Step 1: process the hit message
    val hitRequest = TestDoProcessRequest.builder("join")
            .setMessages(listOf(buildHitMessage(hitId, hitTime, "full-payload")))
            .setWatermarks(defaultWatermarks())
            .build()
    val hitResponse = testHarness.doProcess(hitRequest)

    val key = buildKey(hitId, hitTime)
    val stateAfterHit = hitResponse.getExternalState("join-state", key)!!

    // Step 2: process the show message, pass the state from step 1
    val showRequest = TestDoProcessRequest.builder("join")
            .setMessages(listOf(buildActionMessage(hitId, hitTime, hitTime + 3L, false, hitTime + 1L)))
            .setExternalState("join-state", mapOf(key to stateAfterHit))
            .setWatermarks(defaultWatermarks())
            .build()
    val showResponse = testHarness.doProcess(showRequest)

    val stateAfterShow = showResponse.getExternalState("join-state", key)!!

    // Step 3: process the click message
    val clickRequest = TestDoProcessRequest.builder("join")
            .setMessages(listOf(buildActionMessage(hitId, hitTime, hitTime + 7L, true, hitTime + 2L)))
            .setExternalState("join-state", mapOf(key to stateAfterShow))
            .setWatermarks(defaultWatermarks())
            .build()
    val clickResponse = testHarness.doProcess(clickRequest)

    val stateAfterClick = clickResponse.getExternalState("join-state", key)!!

    // Step 4: trigger the timer
    val timer = Timer.builder()
            .setStreamId("timer")
            .setKey(key)
            .setTriggerTimestamp(hitTime + WAIT_SECONDS)
            .build()
    val timerRequest = TestDoProcessRequest.builder("join")
            .setTimers(listOf(timer))
            .setExternalState("join-state", mapOf(key to stateAfterClick))
            .build()
    val timerResponse = testHarness.doProcess(timerRequest)

    // Check the final result
    assertEquals(1, timerResponse.getOutputMessagesFlatten().size)
    val result = timerResponse.getOutputMessagesFlatten()[0].getPayload() as JoinedAction
    assertEquals(hitId, result.getHitId())
    assertTrue(result.getClick())

    assertTrue(timerResponse.getExternalState("join-state", key)!!.isReset)
}

Warning

TestComputationHarness doesn't store state between doProcess() calls. Each call is an independent batch. To emulate multi-step processing, you must manually pass the state from the previous step's response to the next step's request.

Example: a test with Spring

When you use Spring Boot Starter, testing becomes simpler: the PipelineContext is created automatically via FlowAutoConfiguration based on beans annotated with @FlowComputation or @FlowSourceComputation.

Test configuration

For tests, you need to replace GrpcServerExecution with NoServerTestExecution so you don’t start a real gRPC server:

Java

Kotlin

@TestConfiguration
public class MyTestConfiguration {

    @Bean
    public CompanionExecutionConfig companionExecutionConfig() {
        return new CompanionExecutionConfig(0, new MockEnvironmentReader().worker());
    }

    @Bean
    public GrpcServerExecution grpcServerExecution(
            PipelineContext pipelineContext,
            CompanionExecutionConfig companionExecutionConfig
    ) {
        return new NoServerTestExecution(pipelineContext, companionExecutionConfig);
    }
}

@TestConfiguration
class MyTestConfiguration {

    @Bean
    fun companionExecutionConfig(): CompanionExecutionConfig =
        CompanionExecutionConfig(0, MockEnvironmentReader().worker())

    @Bean
    fun grpcServerExecution(
        pipelineContext: PipelineContext,
        companionExecutionConfig: CompanionExecutionConfig
    ): GrpcServerExecution = NoServerTestExecution(pipelineContext, companionExecutionConfig)
}

Key points:

MockEnvironmentReader replaces environment variable reading. The .worker() method sets YT_FLOW_MODE=Worker.
NoServerTestExecution is a stub for GrpcServerExecution that doesn’t start a real gRPC server.
CompanionExecutionConfig(0, ...) — a port value of 0 means no real port is allocated.

Test class

Java

Kotlin

@SpringBootTest(classes = MyTestConfiguration.class)
class JoinFunctionTest {

    @Autowired
    private PipelineContext pipelineContext;

    private TestComputationHarness testHarness;
    private TableSchema keySchema;

    @BeforeEach
    void init() throws IOException {
        this.keySchema = TableSchema.builder()
                .addValue("hash", ColumnValueType.UINT64)
                .addValue("hit_id", ColumnValueType.STRING)
                .addValue("hit_time", ColumnValueType.UINT64)
                .build();

        var specPath = Paths.getSourcePath(
                "yt/yt/flow/examples/java/lb_wait_click_join/test/pipeline.yson");
        var txtSpec = Files.readString(Path.of(specPath));

        this.testHarness = TestComputationHarness.builder()
                .setPipelineContext(pipelineContext)  // injected via Spring
                .setPipelineSpec(txtSpec)
                .build();
    }

    @Test
    void testHitMessageStoresState() {
        var messages = List.of(buildHitMessage("hit-1", 1000L, "payload-data"));

        var request = TestDoProcessRequest.builder("join")
                .setMessages(messages)
                .setWatermarks(Map.of("hit", 0L, "action", 0L))
                .build();

        var response = testHarness.doProcess(request);

        assertTrue(response.getOutputMessagesFlatten().isEmpty());
        assertEquals(1, response.getOutputTimersFlatten().size());
        assertEquals(1, response.getInternalStateSize("join-state"));
    }

    // ... helper methods and other tests
}

@SpringBootTest(classes = [MyTestConfiguration::class])
class JoinFunctionTest {

    @Autowired
    private lateinit var pipelineContext: PipelineContext

    private lateinit var testHarness: TestComputationHarness
    private lateinit var keySchema: TableSchema

    @BeforeEach
    fun init() {
        keySchema = TableSchema.builder()
                .addValue("hash", ColumnValueType.UINT64)
                .addValue("hit_id", ColumnValueType.STRING)
                .addValue("hit_time", ColumnValueType.UINT64)
                .build()

        val specPath = Paths.getSourcePath(
                "yt/yt/flow/examples/java/lb_wait_click_join/test/pipeline.yson")
        val txtSpec = Files.readString(Path.of(specPath))

        testHarness = TestComputationHarness.builder()
                .setPipelineContext(pipelineContext)  // injected via Spring
                .setPipelineSpec(txtSpec)
                .build()
    }

    @Test
    fun testHitMessageStoresState() {
        val messages = listOf(buildHitMessage("hit-1", 1000L, "payload-data"))

        val request = TestDoProcessRequest.builder("join")
                .setMessages(messages)
                .setWatermarks(mapOf("hit" to 0L, "action" to 0L))
                .build()

        val response = testHarness.doProcess(request)

        assertTrue(response.getOutputMessagesFlatten().isEmpty())
        assertEquals(1, response.getOutputTimersFlatten().size)
        assertEquals(1, response.getInternalStateSize("join-state"))
    }

    // ... helper methods and other tests
}

Differences from a test without Spring

Aspect	Without Spring	With Spring
Creating `PipelineContext`	Manually: `new PipelineContext()`	Automatically via `FlowAutoConfiguration`
Registering Computation	`pipelineContext.registerComputation(...)`	Via the `@FlowComputation` or `@FlowSourceComputation` annotation
Registering streams	`pipelineContext.registerStream(...)`	Via a `FlowStream<?>` bean or `ComputationProvider.getStreams()`
Injecting dependencies into ProcessFunction	Manually via the constructor	Automatically via `@Autowired`
Replacing the gRPC server	Not needed (no server)	`NoServerTestExecution` + `MockEnvironmentReader`

Integration testing with FlowTestJavaBase

To run full integration testing of the pipeline (with real C++ workers, queues, and streams), use the FlowTestJavaBase base class.

Dependencies

PEERDIR(
    yt/yt/flow/library/python/integration_test_base
)

Setup

Your test inherits from FlowTestJavaBase and defines two required attributes:

from yt.yt.flow.library.python.integration_test_base.yt_flow_java_base import FlowTestJavaBase
import yatest.common

class TestWordCount(FlowTestJavaBase):
    JAVA_RUNNER_BINARY_DIR = yatest.common.binary_path(
        "yt/yt/flow/examples/java/word_count/wordcount/"
    )
    JAVA_MAIN_CLASS = "tech.ytsaurus.flow.examples.wordcount.WordCountApplication"

Attribute	Description
`JAVA_RUNNER_BINARY_DIR`	Path to the directory with the Java runner binary (contains `run.sh`)
`JAVA_MAIN_CLASS`	Full name of the pipeline entry-point class: it starts the pipeline and serves as the companion

Integration test examples (Java)

Warning

Integration tests require a deployed YTsaurus cluster and are run via ya make -ttt. For fast iteration, use unit tests with TestComputationHarness as described above.

Use these principles when writing integration tests for pipelines:

Test end-to-end scenarios. That is:
- Write input/source data to the local YT.
- In the pipeline spec, mark the sources as finite=%true.
- For a Key Visitor stream that should run during the test, set finite=%false.
- Run the pipeline.
- When the Key Visitor should finish, call the integration test base method self.ask_key_visitor_to_complete("<computation_id>", "<stream_id>"); it switches that stream's dynamic finite parameter to %true.
- Wait for the pipeline to finish.
- Check the output data.
Prepare the environment with the same code you use in production:
- Test the same pipeline binary that will run in production.
- Generate pipeline specs with the same code that generates them for production.
- If input/output data require non-trivial serialization/parsing, implement this logic with code shared with production.
Use the shared test framework (a README.md is available).
Implement a failover test.
- Many errors surface when workers fail and other workers take over their tasks. So, run a test with problems=True and more than one worker.
Write stable tests.
- Remember that in CI, any part of a test might run unexpectedly long.
  - The ideal runtime for a single test is 20 seconds for any build type.
  - For sanitizer builds, reduce the amount of input data.
  - Set all local timeouts with a multiple safety margin.
- The test execution logic shouldn’t depend significantly on the current time. For example, a test shouldn’t fail if it starts before midnight and finishes after.

Debugging tests

Logs

After a test finishes, you can find its logs in test-results/py3test/testing_out_stuff in the test directory. Key logs include:

run.log — Python test logs.
<test_class_name>/<test_name>/Controller_<number>... — Controller logs (.err is the process’s stderr; .log contains regular logs written via YT_LOG_...).
<test_class_name>/<test_name>/Worker_<number>... — Worker logs.
<test_class_name>/<test_name>/Runner... — Runner logs.

If something isn’t working, check errors in all these logs. The approach to reviewing them is the same as in production — see Logs.

You can also view logs before the test finishes. To do this, locate the temporary directory where the test runs. The simplest way is to run the test with the --keep-temps flag: ya make --keep-temps -ttt <target>. In this case, ya make won’t delete the temporary directory after the test finishes and will print a link to it in the output.

You can also use these bash aliases:

alias curtestdir="ps -f -u $USER | python3 -c \"import sys, re; drs = set(e for e in re.findall(r'[\s=](/\S*testing' + r'_out_stuff)\b', sys.stdin.read())); print('' if len(drs) == 1 else 'Select first from ' + repr(drs), file=sys.stderr); print(list(drs)[0])\""
alias cdcurtestdir='cd $(curtestdir)'

# Get the link to the local YTsaurus UI from the logs.
alias curlocalyt='cat $(curtestdir)/stderr 2>/dev/null | grep YT'

Test framework

You can read about how to set up the environment for better test framework performance and how to influence test parameters in the framework’s README.md.

You configure the behavior of integration tests using --test-param NAME=VAL:

Parameter	Default	Values	Action
`RUNNER_LOG_LEVEL`		`Error`, `Info`, `Debug`, …	Sets the logging level for the process that runs the pipeline (runner).
`PAUSE_BEFORE_FLOW_PROCESS_FEDERATION_TEARDOWN`	`0`	`0`, `1`	Pauses the test before stopping Flow processes. Combined with `--test-disable-timeout`, this lets you keep the local YTsaurus and the Flow process federation running for a long time so you can study them at your own pace via the UI.
`EXTERNAL_YT_CONFIG`	(not set)	YSON — see below	Runs the pipeline on real external YTsaurus clusters instead of the local recipe.

Examples:

ya make -A --test-param RUNNER_LOG_LEVEL=Debug
ya make -A --test-disable-timeout --test-param PAUSE_BEFORE_FLOW_PROCESS_FEDERATION_TEARDOWN=1

`EXTERNAL_YT_CONFIG`

The local YTsaurus still starts via the recipe, but the test ignores it.

Required fields that are common for all clusters:

path — base directory.
tablet_cell_bundle — bundle for the dynamic tables that are created.
proxy_role — RPC proxy role.

Optional: primary_medium (default "default").

The clusters list — the first element is primary. Record fields: cluster_name (required), proxy_url (defaults to cluster_name).

Authorization: when using an external YTsaurus, YT_TOKEN/YT_USER from the local recipe are cleared; yt-wrapper picks up the token from ~/.yt/token.

Isolation: work_yt_path = path/<local-username>/<test_name>; the path/<username> directory is deleted and recreated once per class in setup_class.

Example:

ya make -A --test-param 'EXTERNAL_YT_CONFIG={path="//tmp/yt_flow";tablet_cell_bundle=default;proxy_role=default;clusters=[{cluster_name=<cluster-name>};];}'

Testing with TestComputationHarness in YTsaurus Flow (Java)

General testing architecture

Dependencies

Setting up TestComputationHarness

Builder API

Building test requests

TestDoProcessRequest

Building messages

With typed streams

With untyped streams (PayloadBuilder)

Building keys

Building timers

Configuring watermarks

Pre-populating state

External State

Internal State (Proto)

Analyzing the results

TestDoProcessResponse

Checking output messages

Checking timers

Checking the state

Full view vs. modified states

Example: a test without Spring

Multi-step test

Example: a test with Spring

Test configuration

Test class

Differences from a test without Spring

Integration testing with FlowTestJavaBase

Dependencies

Setup

Debugging tests

Logs

Test framework

EXTERNAL_YT_CONFIG

See also

`EXTERNAL_YT_CONFIG`