Async Request in YTsaurus Flow (Go)

An example of a pipeline that makes an asynchronous call to an external service: one computation turns events into requests and accumulates the responses in an external state, and another one serves the requests without any state. This is a Go implementation of the same scenario as the C++ example.

Source code

Structure

The companion serves two computations; injector stays a native source declared in the spec:

state (stateKeeper) — a stateful computation grouped by key, which:
- takes events from the event stream and produces a request in the request stream with a random request_id;
- takes responses from the response stream and adds up their total length (total_length) in the external state /state.
processor (requestProcessor) — a stateless computation grouped by request_id: it takes requests from the request stream and immediately answers with the length of the request string in the response stream.

The event → request → response → state cycle closes between the two computations. An event is answered with a request rather than with the result right away, so the serving side never holds up processing: the response arrives later, as a separate message, and only then does the state of the key move on.

`main.go`

The entry point: creating the pipeline and registering both computations.

Source code: main.go


type eventMessage struct {
	flow.YSONMessage
	Key  uint64 `yson:"key"`
	Data string `yson:"data"`
}

type requestMessage struct {
	flow.YSONMessage
	RequestID uint64 `yson:"request_id"`
	Key       uint64 `yson:"key"`
	Request   string `yson:"request"`
}

type responseMessage struct {
	flow.YSONMessage
	RequestID uint64 `yson:"request_id"`
	Key       uint64 `yson:"key"`
	Length    int64  `yson:"length"`
}

func main() {
	pipeline := flow.NewPipeline()
	pipeline.AddStreams(
		flow.NewYSONStream[eventMessage](eventStreamID),
		flow.NewYSONStream[requestMessage](requestStreamID),
		flow.NewYSONStream[responseMessage](responseStreamID),
	)
	pipeline.Add(
		flow.NewRowComputation("state", &stateKeeper{}),
		flow.NewRowComputation("processor", &requestProcessor{}),
	)

	if err := pipeline.Run(); err != nil {
		fmt.Fprintf(os.Stderr, "async_request: %v\n", err)
		os.Exit(1)
	}
}

`state_keeper.go`

Routing of the input streams (event / response) and work with the external state.


type totalLengthState struct {
	TotalLength int64 `yson:"total_length"`
}

type stateKeeper struct{}

var _ flow.RowFunction = (*stateKeeper)(nil)

func (k *stateKeeper) OnMessage(
	ctx context.Context,
	rt flow.Runtime,
	msg flow.ExtendedMessage,
	out flow.OutputCollector,
) error {
	switch msg.StreamID {
	case eventStreamID:
		return k.sendRequest(rt, msg, out)
	case responseStreamID:
		return k.accumulate(rt, msg)
	default:
		return fmt.Errorf("unhandled stream %q", msg.StreamID)
	}
}

func (*stateKeeper) sendRequest(rt flow.Runtime, msg flow.ExtendedMessage, out flow.OutputCollector) error {
	var event eventMessage
	if err := msg.ConvertTo(&event); err != nil {
		return err
	}

	request := flow.NewYSONMessage[requestMessage](requestStreamID)
	request.RequestID = rand.Uint64()
	request.Key = event.Key
	request.Request = event.Data

	encoded, err := flow.ConvertFrom(rt, request)
	if err != nil {
		return err
	}
	out.AddMessage(encoded)
	return nil
}

func (*stateKeeper) accumulate(rt flow.Runtime, msg flow.ExtendedMessage) error {
	var response responseMessage
	if err := msg.ConvertTo(&response); err != nil {
		return err
	}

	state, err := flow.OpenExternalState(rt, totalLengthStateName, msg)
	if err != nil {
		return err
	}

	var total totalLengthState
	if _, err := state.ConvertTo(&total); err != nil {
		return err
	}
	total.TotalLength += response.Length
	return state.ConvertFrom(&total)
}

`request_processor.go`

The stateless request handler: it computes the length of the request string and returns a response.


type requestProcessor struct{}

var _ flow.RowFunction = (*requestProcessor)(nil)

func (*requestProcessor) OnMessage(
	ctx context.Context,
	rt flow.Runtime,
	msg flow.ExtendedMessage,
	out flow.OutputCollector,
) error {
	var request requestMessage
	if err := msg.ConvertTo(&request); err != nil {
		return err
	}

	response := flow.NewYSONMessage[responseMessage](responseStreamID)
	response.RequestID = request.RequestID
	response.Key = request.Key
	response.Length = int64(len(request.Request))

	encoded, err := flow.ConvertFrom(rt, response)
	if err != nil {
		return err
	}
	out.AddMessage(encoded)
	return nil
}

Key patterns

Routing by msg.StreamID: a switch on the input stream identifier lets one computation handle several inputs with different logic. An unknown stream is an error rather than a silent skip.
A random request_id: rand.Uint64() links the request to the response. The request carries the key of the original event, so the response, partitioned by request_id, comes back to the state the call belongs to.
External state through flow.OpenExternalState(rt, "/state", msg): the row is turned into totalLengthState by ConvertTo, changed as a structure, and saved through ConvertFrom.
A stateless computation: requestProcessor uses no state and is grouped by request_id rather than by the event key, so the requests of one key spread over all partitions and scale independently.
Stream dependency: streams_dependency in the spec declares that request is produced from event — the worker takes this into account when advancing watermarks.

Async Request in YTsaurus Flow (Go)

Structure

main.go

state_keeper.go

request_processor.go

Key patterns

See also

`main.go`

`state_keeper.go`

`request_processor.go`