The distribute flag in YTsaurus Flow (Java)

The distribute flag is a per-message flag that you set when you add an output message to SourceComputation. It controls whether the message is published further along the processing graph.

The distribute flag ensures:

• Correct watermark evaluation: messages with distribute=false are still accounted for by the watermark generator (unlike filtering in onMessage, which can break the watermark).
• Assignment of deterministic identifiers to messages.

When to use distribute=false

Use the distribute=false flag when:

• You need to filter some output messages at the source computation stage.
• Correct watermark evaluation is important.

If you don’t set the flag, it defaults to true, and the message is published further.

Usage

Move the filtering logic to the processing function: instead of a separate filtering step, emit the message with the required AddMessageOptions.

public class HitParsingFunction implements RowFunction {
    @Override
    public void onMessage(ExtendedMessage message, OutputCollector output, RuntimeContext ctx) {
        var hit = ProtoUtils.parseBytes(message.get("data", byte[].class), THit.class);
        // Duplicates are emitted but not published further.
        var distribute = !hit.getHitPayload().equals("duplicate_payload");
        output.addMessage(
                ctx.createMessageBuilder("hit")
                        .set("hit_id", hit.getHitId())
                        .set("hit_payload", hit.getHitPayload())
                        .finish(),
                AddMessageOptions.builder().setDistribute(distribute).build()
        );
    }
}

class HitParsingFunction : RowFunction {
    override fun onMessage(message: ExtendedMessage, output: OutputCollector, ctx: RuntimeContext) {
        val hit = ProtoUtils.parseBytes(message.get("data", ByteArray::class.java), THit::class.java)
        // Duplicates are emitted but not published further.
        val distribute = hit.hitPayload != "duplicate_payload"
        output.addMessage(
            ctx.createMessageBuilder("hit")
                .set("hit_id", hit.hitId)
                .set("hit_payload", hit.hitPayload)
                .finish(),
            AddMessageOptions.builder().setDistribute(distribute).build()
        )
    }
}

Registering a source computation

You create a source computation using SourceComputation.builder(). You no longer need a separate filtering parameter — the decision about publishing is made in the processing function.

var hitReader = SourceComputation.builder()
        .setComputationId("hit_reader")
        .setProcessFunction(new HitParsingFunction())
        .build();

See also

• Computation (Java)
• Watermarks
• The distribute flag (Python)