This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/camel-performance-tests.git
commit fa6f787f6afc83f9e4d95825899f9b617da3ba3b Author: Claus Ibsen <[email protected]> AuthorDate: Sun Oct 18 11:04:40 2026 +0200 ai-benchmark: a new set in steps-new/, with connect-service/kafka-orders as its first example The ladder's 71 steps stay in steps-ladder/ for comparison. The new examples (Oct 18: messaging, the EIPs the ladder never uses, code-first REST, AI) are written to steps-new/ through steps_dir, and run with BENCH_STEPS_DIR=steps-new run-stepwise-ladder.sh <tag> [k]. kafka-orders: Kafka from camel infra, the three README steps after the first: a validation route with choice and stop, a multicast to two more topics with a consumer each, and the order id as the record key (the dispatch log prints CamelKafkaKey, so the log proves it). Reference pass 3/3 in four runs. The project sets camel.component.kafka.pollTimeoutMs=500: a camel-kafka consumer stops only after its current poll, up to 5 s each and one route after another, and the 5 s shutdown timeout of camel run --dev is then exceeded on a reload with two or more Kafka routes; the forced shutdown logs ERRORs from the interrupted Kafka clients. Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]> Claude-Session: https://claude.ai/code/session_01STT6whBgK1AqsSsUKrnE8m --- ai-benchmark/.gitignore | 1 + ai-benchmark/gen_stepwise.py | 64 +++++++++++++++++++++- ai-benchmark/run-stepwise-ladder.sh | 7 ++- .../orders/order-1001.json | 2 + .../orders/order-1002.json | 2 + .../orders/order-1003.json | 2 + 6 files changed, 74 insertions(+), 4 deletions(-) diff --git a/ai-benchmark/.gitignore b/ai-benchmark/.gitignore index 46712c1..ef85329 100644 --- a/ai-benchmark/.gitignore +++ b/ai-benchmark/.gitignore @@ -8,6 +8,7 @@ local/ __pycache__/ # generated by gen_stepwise.py from its own step definitions: the steps files and the starting projects steps-ladder/ +steps-new/ stepwise-ladder/ stepwise-project/ # generated by gen_kamelet.py: the Kamelet side check diff --git a/ai-benchmark/gen_stepwise.py b/ai-benchmark/gen_stepwise.py index 34bc0f9..fd7ba4d 100644 --- a/ai-benchmark/gen_stepwise.py +++ b/ai-benchmark/gen_stepwise.py @@ -777,6 +777,66 @@ EXAMPLES.append({ ]}) +# ================================================================ the new set (2026-10-18), written to steps-new/ +# Claus, Oct 18: messaging, the EIPs the ladder never uses, code-first REST and AI. Kept out of steps-ladder/ so the +# nightly 71 steps stay comparable; run with BENCH_STEPS_DIR=steps-new run-stepwise-ladder.sh <tag> [k]. +NEW = "steps-new" + +# ---------------------------------------------------------------- connect-service/kafka-orders (Kafka started with camel infra) +KO = "kafka-orders.camel.yaml" +# pollTimeoutMs 500: a camel-kafka consumer stops only after its current poll returns (the poll loop holds the lock that +# stop() waits for before it wakes the consumer), up to the default 5000 ms per consumer and one route after another; +# the 5 s shutdown timeout of camel run --dev is then exceeded on a reload with two or more Kafka routes, and the +# forced shutdown logs ERRORs from the interrupted Kafka clients (kref1, Oct 18). Remove once CAMEL-25753 is fixed. +KO_PROPS = "camel.component.kafka.brokers=localhost:9092\ncamel.component.kafka.pollTimeoutMs=500\n" +def kafka_from(topic): + # earliest: every reload starts the consumers in a new random group, which then reads the topic from the start + return " uri: kafka\n parameters:\n topic: " + topic + "\n autoOffsetReset: earliest\n" +def kafka_to(topic, ind=" "): + return ind + "- to:\n" + ind + " uri: kafka\n" + ind + " parameters:\n" + ind + " topic: " + topic + "\n" +KO_VALIDATE = route("validate-order", DIRECT("validate-order"), + " - choice:\n when:\n - expression:\n simple:\n" + " expression: \"${header.status} != 'paid'\"\n steps:\n" + + log("Order ${header.orderId} is ${header.status}, not published", " ") + + " - stop: {}\n otherwise:\n steps:\n" + + log("Order ${header.orderId} validated", " ")) +def ko_incoming(validate, key): + return route("incoming-orders", FILE_ORDERS, + set_header("orderId", "jsonpath", "\"$.orderId\"") + set_header("status", "jsonpath", "\"$.status\"") + + (" - to:\n uri: direct:validate-order\n" if validate else "") + + (set_header("CamelKafkaKey", "simple", "\"${header.orderId}\"") if key else "") + + kafka_to("orders") + log("Published ${header.orderId} to the orders topic")) +def ko_dispatch(fan_out, key): + return route("dispatch", kafka_from("orders"), + (log("Dispatching ${header.CamelKafkaKey} from partition ${header.CamelKafkaPartition} offset ${header.CamelKafkaOffset}") + if key else log("Received ${header.orderId} from the orders topic")) + + (" - multicast:\n steps:\n" + kafka_to("fulfilment", " ") + + kafka_to("notifications", " ") if fan_out else "")) +KO_DEPARTMENTS = ("\n" + route("fulfilment", kafka_from("fulfilment"), + UNMARSHAL + log("Fulfilment: ${body[lines].size()} line(s) of ${body[orderId]} to the warehouse")) + + "\n" + route("notification", kafka_from("notifications"), + UNMARSHAL + log("Notification: email to customer ${body[customer]} about ${body[orderId]}"))) +ko_s1 = ko_incoming(False, False) + "\n" + ko_dispatch(False, False) +ko_s2 = ko_incoming(True, False) + "\n" + KO_VALIDATE + "\n" + ko_dispatch(False, False) +ko_s3 = ko_incoming(True, False) + "\n" + KO_VALIDATE + "\n" + ko_dispatch(True, False) + KO_DEPARTMENTS +ko_s4 = ko_incoming(True, True) + "\n" + KO_VALIDATE + "\n" + ko_dispatch(True, True) + KO_DEPARTMENTS +EXAMPLES.append({ + "name": "connect-service-kafka-orders", "route_file": KO, "props_file": "application.properties", "wait_seconds": 20, + "seed": "connect-service-kafka-orders", "infra": ["kafka"], "steps_dir": NEW, + "initial": {KO: ko_s1, "application.properties": KO_PROPS}, + "steps": [ + {"request": "Only paid orders may be published to Kafka. Add a route validate-order from direct:validate-order with a choice: when the status header is not paid, log \"Order ${header.orderId} is ${header.status}, not published\" and stop the message; otherwise log \"Order ${header.orderId} validated\". Call it from incoming-orders before the order is sent to the orders topic.", + "check": {"file_regex": "validate-order", "log_regex": ["Order ORD-1003 is pending, not published", "Order ORD-1001 validated", "Published ORD-1001 to the orders topic"], "log_not_regex": "Published ORD-1003"}, + "reference": {KO: ko_s2}}, + {"request": "Fan out the orders in dispatch: send each one with a multicast to two more topics, fulfilment and notifications. Add a route fulfilment that consumes the fulfilment topic from the earliest offset, unmarshals the JSON with Jackson and logs \"Fulfilment: ${body[lines].size()} line(s) of ${body[orderId]} to the warehouse\", and a route notification that consumes the notifications topic the same way and logs \"Notification: email to customer ${body[customer]} about ${bod [...] + "check": {"file_regex": "multicast", "log_regex": ["Fulfilment: 2 line\\(s\\) of ORD-1001 to the warehouse", "Notification: email to customer C-482 about ORD-1001"]}, + "reference": {KO: ko_s3}}, + {"request": "Publish each order with its order id as the Kafka record key, and change the log of dispatch to \"Dispatching ${header.CamelKafkaKey} from partition ${header.CamelKafkaPartition} offset ${header.CamelKafkaOffset}\".", + "check": {"log_regex": "Dispatching ORD-1001 from partition \\d+ offset \\d+"}, + "reference": {KO: ko_s4}}, + ]}) + + def write_project(ex): d = os.path.join(PROJECTS, ex["name"]) shutil.rmtree(d, ignore_errors=True) @@ -814,7 +874,9 @@ def main(): pd = os.path.join(HERE, ex["peer"]["project"]) shutil.rmtree(pd, ignore_errors=True) shutil.copytree(os.path.join(SEEDS, ex["peer"]["seed"]), pd) - with open(os.path.join(STEPS_DIR, ex["name"] + ".json"), "w") as f: + steps_dir = os.path.join(HERE, ex["steps_dir"]) if ex.get("steps_dir") else STEPS_DIR + os.makedirs(steps_dir, exist_ok=True) + with open(os.path.join(steps_dir, ex["name"] + ".json"), "w") as f: json.dump(cfg, f, indent=1) write_project(ex) print(f"{ex['name']}: {len(steps)} steps, project stepwise-ladder/{ex['name']}") diff --git a/ai-benchmark/run-stepwise-ladder.sh b/ai-benchmark/run-stepwise-ladder.sh index 5eba143..849c90a 100755 --- a/ai-benchmark/run-stepwise-ladder.sh +++ b/ai-benchmark/run-stepwise-ladder.sh @@ -3,6 +3,7 @@ # Usage: run-stepwise-ladder.sh <tag> [k] Results: stepwise/<tag>-<i>/<name>/ (results.json, run.log, traces) # BENCH_REFERENCE=1 applies the reference files instead of asking the model (the reference pass of the steps files). # BENCH_ONLY="<name> [<name> ...]" runs only those examples. +# BENCH_STEPS_DIR=steps-new runs the new set (2026-10-18) instead of the ladder in steps-ladder/. set -u TAG="${1:?usage: run-stepwise-ladder.sh <tag> [k]}" K="${2:-1}" @@ -12,7 +13,7 @@ command -v caffeinate > /dev/null && caffeinate -i -s -w $$ & for i in $(seq 1 "$K"); do if (( K > 1 )); then T="$TAG-$i"; else T="$TAG"; fi : > "stepwise-$T.log" # a fresh log per run, so a later wait on its DONE line cannot see an earlier run's - for f in steps-ladder/*.json; do + for f in ${BENCH_STEPS_DIR:-steps-ladder}/*.json; do name=$(basename "$f" .json) if [[ -n "${BENCH_ONLY:-}" && " $BENCH_ONLY " != *" $name "* ]]; then continue; fi python3 gen_stepwise.py "$name" > /dev/null @@ -30,9 +31,9 @@ for i in $(seq 1 "$K"); do done # the services an example needed stay up across its passes; stop them once the run is over for svc in $(python3 -c " -import glob, json +import glob, json, os out = set() -for f in glob.glob('steps-ladder/*.json'): +for f in glob.glob(os.environ.get('BENCH_STEPS_DIR', 'steps-ladder') + '/*.json'): try: out.update(json.load(open(f)).get('infra') or []) except Exception: pass print(' '.join(sorted(out)))"); do camel infra stop "$svc" > /dev/null 2>&1 || true; done diff --git a/ai-benchmark/seed/connect-service-kafka-orders/orders/order-1001.json b/ai-benchmark/seed/connect-service-kafka-orders/orders/order-1001.json new file mode 100644 index 0000000..3eb5f03 --- /dev/null +++ b/ai-benchmark/seed/connect-service-kafka-orders/orders/order-1001.json @@ -0,0 +1,2 @@ +{"orderId": "ORD-1001", "customer": "C-482", "country": "DK", "status": "paid", + "lines": [{"sku": "CAMEL-TSHIRT", "qty": 2, "price": 19.95}, {"sku": "CAMEL-MUG", "qty": 1, "price": 9.50}]} diff --git a/ai-benchmark/seed/connect-service-kafka-orders/orders/order-1002.json b/ai-benchmark/seed/connect-service-kafka-orders/orders/order-1002.json new file mode 100644 index 0000000..bb7aa0e --- /dev/null +++ b/ai-benchmark/seed/connect-service-kafka-orders/orders/order-1002.json @@ -0,0 +1,2 @@ +{"orderId": "ORD-1002", "customer": "C-207", "country": "DE", "status": "paid", + "lines": [{"sku": "CAMEL-MUG", "qty": 3, "price": 9.50}]} diff --git a/ai-benchmark/seed/connect-service-kafka-orders/orders/order-1003.json b/ai-benchmark/seed/connect-service-kafka-orders/orders/order-1003.json new file mode 100644 index 0000000..9edab76 --- /dev/null +++ b/ai-benchmark/seed/connect-service-kafka-orders/orders/order-1003.json @@ -0,0 +1,2 @@ +{"orderId": "ORD-1003", "customer": "C-134", "country": "US", "status": "pending", + "lines": [{"sku": "CAMEL-TSHIRT", "qty": 1, "price": 19.95}, {"sku": "CAMEL-CAP", "qty": 1, "price": 14.00}, {"sku": "CAMEL-MUG", "qty": 2, "price": 9.50}]}
