External State Join in YTsaurus Flow (Java and Kotlin)
This pipeline enriches queue events from a prebuilt dynamic reference table. Java and Kotlin use the same read-only joined-state API.
Pipeline structure
event_reader publishes the event stream. The lookup_join companion computation is grouped by key, reads /reference through external_state_joiners, and sends successful matches to enriched. The queue sink stores that output. The joiner's path may point to a Cypress link, allowing the reference table to be switched without restarting the pipeline.
See the Java pipeline spec or Kotlin pipeline spec.
Read the joined state
StateDescriptors.externalReadOnly("/reference") binds the process function to the joiner declared under the same name. The accessor selects the row by the current message key. A missing row returns no name, so the example emits nothing. The joined accessor does not provide a write path.
@Override
public void onMessage(ExtendedMessage message, OutputCollector output, RuntimeContext ctx) {
ReadOnlyExternalStateAccessor stateAccessor = ctx.getState(REFERENCE_STATE, message);
String name = stateAccessor.getOrDefault().get("name", String.class);
if (name == null) {
return;
}
output.addMessage(ctx.createMessageBuilder("enriched")
.set("key", message.get("key", Long.class))
.set("name", name)
.finish());
}
override fun onMessage(message: ExtendedMessage, output: OutputCollector, ctx: RuntimeContext) {
val stateAccessor = ctx.getState(REFERENCE_STATE, message)
val name = stateAccessor.orDefault.get("name", String::class.java) ?: return
output.addMessage(
ctx.createMessageBuilder("enriched")
.set("key", message.get("key", Long::class.java))
.set("name", name)
.finish()
)
}
The function is registered as lookup_join through @FlowComputation; the pipeline spec declares the event and enriched stream schemas.