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 39041a1bcd4e1984966a9ae61b3f7c5df55809c2 Author: Claus Ibsen <[email protected]> AuthorDate: Mon Nov 2 12:25:19 2026 +0100 ai-benchmark: connect-aws-s3-sqs in the new set, S3 and SQS on the AWS emulator of camel infra run aws s3 camel infra run aws s3 starts Floci (mirror.gcr.io/floci/floci:2.4.0), an AWS emulator with S3, SQS and more in one container on localhost:4566. The project starts with a route that logs the order files of orders/. 1. Connect (endpoint override, region and keys in the request), upload each order file to the bucket orders with its file name as the key, and log it. Restart, as jms step 1: the log, and the bucket lists the object. 2. A route new-objects consumes the bucket, reads the order id from the object (a stream) into a variable, and sends the event {"orderId", "object"} to the SQS queue order-events. Restart: the log, and the queue holds the event, with both fields. 3. A route order-events consumes the queue and logs the event's fields. 4. Archive: the consumer moves each object to the bucket orders-archive (moveAfterRead): the archive bucket lists the objects. The probes read the bucket and the queue over the emulator's HTTP API, which answers requests without a signature: ListObjectsV2, and a ReceiveMessage with VisibilityTimeout 0, which leaves the messages on the queue. The checkpoint purges the queue when it resets a step (checkpoint_requests). Reference pass: 4 of 4, twice. Negative pass: 0 of 4. Model run (k=1, qwen3.6:35b-mlx, before the queue purge): steps 1 and 2 passed. Steps 3 and 4 had the right end state, but failed on errors: their consumer read messages that were not valid JSON (JsonParseException), left on the queue by the model's own first tries in step 2. The checkpoint purge is for that case. Co-Authored-By: Claude Opus 5.5 <[email protected]> Claude-Session: https://claude.ai/code/session_016Y4dDi5iQNgtHrDdz2NpwF --- ai-benchmark/gen_stepwise.py | 69 ++++++++++++++++++++++ .../seed/connect-aws-s3-sqs/orders/order-1001.json | 2 + .../seed/connect-aws-s3-sqs/orders/order-1002.json | 2 + .../seed/connect-aws-s3-sqs/orders/order-1003.json | 2 + 4 files changed, 75 insertions(+) diff --git a/ai-benchmark/gen_stepwise.py b/ai-benchmark/gen_stepwise.py index 4aec1fe..9b9da6d 100644 --- a/ai-benchmark/gen_stepwise.py +++ b/ai-benchmark/gen_stepwise.py @@ -1189,6 +1189,75 @@ EXAMPLES.append({ ]}) +# ---------------------------------------------------------------- connect-aws-s3-sqs (S3 and SQS on the emulator of camel infra run aws) +# Claus, Nov 2: the order files go to an S3 bucket; a route takes each new object and sends an event with the order id +# to an SQS queue; another route consumes the queue; and the processed objects are archived. `camel infra run aws s3` +# starts Floci, an AWS emulator with S3, SQS and more in one container on http://localhost:4566. The probes read the +# bucket and the queue over its HTTP API, which answers requests without a signature: ListObjectsV2, and a +# ReceiveMessage with VisibilityTimeout 0, which leaves the messages on the queue. An S3 object arrives as a stream. +AS = "aws-orders.camel.yaml" +AS_PROPS = ("# The AWS services started by: camel infra run aws s3\n" + "camel.component.aws2-s3.access-key=accesskey\ncamel.component.aws2-s3.secret-key=secretkey\n" + "camel.component.aws2-s3.region=us-east-1\ncamel.component.aws2-s3.override-endpoint=true\n" + "camel.component.aws2-s3.uri-endpoint-override=http://localhost:4566\n";) +AS_PROPS_SQS = AS_PROPS + AS_PROPS.split("\n", 1)[1].replace("aws2-s3", "aws2-sqs") +AS_UPLOAD = (" - to:\n uri: aws2-s3\n parameters:\n bucketNameOrArn: orders\n" + " autoCreateBucket: true\n keyName: \"${file:name}\"\n") +def as_new_objects(archive): + frm = " uri: aws2-s3\n parameters:\n bucketNameOrArn: orders\n" + ( + " moveAfterRead: true\n destinationBucket: orders-archive\n autoCreateBucket: true\n" if archive else "") + return route("new-objects", frm, + " - setVariable:\n name: orderId\n expression:\n jsonpath:\n" + " expression: \"$.orderId\"\n" + " - setBody:\n expression:\n simple:\n" + " expression: '{\"orderId\": \"${variable.orderId}\", \"object\": \"${header.CamelAwsS3Key}\"}'\n" + " - to:\n uri: aws2-sqs\n parameters:\n queueNameOrArn: order-events\n" + " autoCreateQueue: true\n" + + log("Event for ${variable.orderId} sent to the queue order-events")) +AS_EVENTS = route("order-events", " uri: aws2-sqs\n parameters:\n queueNameOrArn: order-events\n autoCreateQueue: true\n", + UNMARSHAL + log("Event received for ${body[orderId]} (object ${body[object]})")) +as_s0 = route("incoming-orders", FILE_ORDERS, log("Order file ${file:name} received")) +as_s1 = route("incoming-orders", FILE_ORDERS, AS_UPLOAD + log("Uploaded ${file:name} to the bucket orders")) +as_s2 = as_s1 + "\n" + as_new_objects(False) +as_s3 = as_s2 + "\n" + AS_EVENTS +as_s4 = as_s1 + "\n" + as_new_objects(True) + "\n" + AS_EVENTS +AWS = "http://localhost:4566"; +S3_LIST = lambda bucket, key: {"method": "GET", "url": AWS + "/" + bucket + "?list-type=2", "expect_status": 200, + "body_regex": "<Key>" + key.replace(".", "\\.") + "</Key>"} +# the event as the queue holds it: the message body is a JSON string inside the JSON answer, its quotes escaped +SQS_PEEK = lambda rx: {"method": "POST", "url": AWS + "/", "expect_status": 200, "body_regex": rx, + "headers": {"Content-Type": "application/x-amz-json-1.0", "X-Amz-Target": "AmazonSQS.ReceiveMessage"}, + "body": json.dumps({"QueueUrl": AWS + "/000000000000/order-events", "MaxNumberOfMessages": 10, + "VisibilityTimeout": 0})} +# the checkpoint empties the queue: messages that a model's earlier try could not write as JSON would make the next +# step's consumer fail on them, again after every visibility timeout (awsm1, steps 3 and 4) +SQS_PURGE = {"method": "POST", "url": AWS + "/", "note": "(purge the queue order-events)", + "headers": {"Content-Type": "application/x-amz-json-1.0", "X-Amz-Target": "AmazonSQS.PurgeQueue"}, + "body": json.dumps({"QueueUrl": AWS + "/000000000000/order-events"})} +EXAMPLES.append({ + "name": "connect-aws-s3-sqs", "route_file": AS, "props_file": "application.properties", "wait_seconds": 15, + "seed": "connect-aws-s3-sqs", "infra": ["aws s3"], "checkpoint_requests": [SQS_PURGE], "steps_dir": NEW, + "initial": {AS: as_s0, "application.properties": "# application properties\n"}, + "steps": [ + {"request": "Connect to the AWS services that camel infra started: endpoint override http://localhost:4566, region us-east-1, access key accesskey, secret key secretkey. Upload each order file to the S3 bucket orders, with the file name as the object key, and create the bucket if it does not exist. Log \"Uploaded ${file:name} to the bucket orders\" instead of received. The integration is restarted after your change, so new connection settings apply.", + # a reload does not apply camel.component.* properties to a component that exists already: a first try + # without them would keep the component without its endpoint, so the step restarts, as jms step 1 does + "check": {"file_regex": "aws2-s3", "log_regex": ["Uploaded order-1001\\.json to the bucket orders", "Uploaded order-1003\\.json to the bucket orders"], + "probes": [S3_LIST("orders", "order-1001.json")]}, + "reference": {AS: as_s1, "application.properties": AS_PROPS}, "restart": True}, + {"request": "Add a route new-objects that consumes the new objects of the bucket orders; for now the consumer may delete each object after reading it. Read the order id from the JSON of the object into a variable orderId, send the event {\"orderId\": \"<order id>\", \"object\": \"<object key>\"} as JSON text to the SQS queue order-events (create the queue if it does not exist), and log \"Event for ${variable.orderId} sent to the queue order-events\". The integration is restarted [...] + "check": {"file_regex": "aws2-sqs", "log_regex": ["Event for ORD-1001 sent to the queue order-events", "Event for ORD-1003 sent to the queue order-events"], + "probes": [SQS_PEEK(r'\\"orderId\\":\s*\\"ORD-100[123]\\"'), SQS_PEEK(r'\\"object\\":\s*\\"order-100[123]\.json\\"')]}, + "reference": {AS: as_s2, "application.properties": AS_PROPS_SQS}, "restart": True}, + {"request": "Add a route order-events that consumes the queue order-events, unmarshals the event from JSON and logs \"Event received for ${body[orderId]} (object ${body[object]})\".", + "check": {"log_regex": ["Event received for ORD-1001 \\(object order-1001\\.json\\)", "Event received for ORD-1003 \\(object order-1003\\.json\\)"]}, + "reference": {AS: as_s3}}, + {"request": "Keep the processed orders: instead of only deleting each object of the bucket orders after it is read, move it to the bucket orders-archive, and create that bucket if it does not exist.", + "check": {"file_regex": "orders-archive", "probes": [S3_LIST("orders-archive", "order-1001.json"), S3_LIST("orders-archive", "order-1003.json")]}, + "reference": {AS: as_s4}}, + ]}) + + def write_project(ex): d = os.path.join(PROJECTS, ex["name"]) shutil.rmtree(d, ignore_errors=True) diff --git a/ai-benchmark/seed/connect-aws-s3-sqs/orders/order-1001.json b/ai-benchmark/seed/connect-aws-s3-sqs/orders/order-1001.json new file mode 100644 index 0000000..3eb5f03 --- /dev/null +++ b/ai-benchmark/seed/connect-aws-s3-sqs/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-aws-s3-sqs/orders/order-1002.json b/ai-benchmark/seed/connect-aws-s3-sqs/orders/order-1002.json new file mode 100644 index 0000000..bb7aa0e --- /dev/null +++ b/ai-benchmark/seed/connect-aws-s3-sqs/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-aws-s3-sqs/orders/order-1003.json b/ai-benchmark/seed/connect-aws-s3-sqs/orders/order-1003.json new file mode 100644 index 0000000..9edab76 --- /dev/null +++ b/ai-benchmark/seed/connect-aws-s3-sqs/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}]}
