DataStream API

6 h
requires FLK-01

From your world. a dbt staging model (filter + enrich + route), but as a continuous DataStream job

Enrich orders and route rejects to a side output

This exercise is a dbt staging model written again as a continuous DataStream job. The job filters orders to a valid-currency allowlist and enriches them with a dimension lookup. It sends the rows that fail validation to a dead-letter stream. All of this happens in one pass.

Task

Implement de.flk.EnrichOrders.run(List<String[]> orders, Map<String,String> companyNames).

Each order is [orderId, companyId, currency, amount]. Build a DataStream job that does this in a single pass:

Valid orders (currency in VALID = {USD, EUR, GBP, JPY}): emit them to the main stream, enriched with the company name. Render them as orderId|companyName|currency|amount. If the company id is not in companyNames, use "UNKNOWN" as the name. Do not drop or reject the order.
Invalid orders (currency not in the allowlist): route them to a side output (the dead-letter stream). Render them as orderId|companyId|currency.

Return a Result that holds the enriched list and the rejects list.

What the tests check

The enriched set equals the enriched set of an independent oracle. The test compares sets, so the order of the subtasks does not matter.
The rejects set equals the rejects of the oracle. Lowercase usd is not USD.
No order is lost or duplicated. Every input id appears exactly once across the two streams.
A valid order with an unknown company is enriched as UNKNOWN. It is never rejected.

Hints

Use a ProcessFunction<String[], String>. In processElement, call out.collect(..) for valid orders and ctx.output(rejectsTag, ..) for invalid orders.
Declare the tag as an anonymous subclass: new OutputTag<String>("rejects"){}.
Read the side stream with mainOperator.getSideOutput(rejectsTag). Collect the main and side streams with executeAndCollect().

Starter code

src/main/java/de/flk/EnrichOrders.java

package de.flk;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * FLK-02 starter — NOT implemented yet (the tests must fail: G1).
 *
 * Build a single-pass DataStream job that filters orders to the valid-currency
 * allowlist, enriches valid orders with their company name, and routes invalid
 * orders to a side output.
 */
public class EnrichOrders {

    public static final Set<String> VALID = Set.of("USD", "EUR", "GBP", "JPY");

    public static final class Result {
        public final List<String> enriched;
        public final List<String> rejects;

        public Result(List<String> enriched, List<String> rejects) {
            this.enriched = enriched;
            this.rejects = rejects;
        }
    }

    public static Result run(List<String[]> orders, Map<String, String> companyNames)
            throws Exception {
        // TODO: build a ProcessFunction that emits valid orders enriched with the
        //       company name to the main collector and invalid-currency orders to
        //       a side output (OutputTag). Collect both streams and return them.
        return new Result(new ArrayList<>(), new ArrayList<>());
    }
}

Hints

Hint 1

Build a ProcessFunction over the orders: emit valid orders (currency in the allowlist) to the main Collector and invalid ones to a side output via ctx.output(rejectsTag, order).

Hint 2

Enrich each valid order by looking up its company name from the provided map; keep the companyId too. Declare the OutputTag as an anonymous subclass: new OutputTag<...>("rejects"){}.

Hint 3

Collect the main stream and the getSideOutput(rejectsTag) stream separately with executeAndCollect(), and return both lists.

Run it locally

uv run de start FLK-02/01-enrich-and-reject
# edit workspace/FLK-02/01-enrich-and-reject/
uv run de test