Flink agents to S3 Tables

The AI SDK app writes chat and agent runs to PicoMQ streams under /examples/agents/ai-sdk/. Managed Flink discovers those streams by prefix, consumes them over the Kafka listener of the AWS harness, and writes two Iceberg tables in S3 Tables. The app never changes and never sees Iceberg.

Source: examples/connectors/flink-agents-s3-tables.

Streams	Table	Columns
chat/{id}	agents.conversations	stream, seq, role, content, ts
agent/run-{id}	agents.agent_events	stream, seq, type, step_index, finish_reason, total_tokens, tools, ts

stream is the PicoMQ stream name, seq the record position, ts the record timestamp. multi/* streams are discovered and ignored.

Run

The harness is applied and the AI SDK app is running against it from your laptop, see examples/agents/ai-sdk/harness.md.

cd examples/connectors/flink-agents-s3-tables
export AWS_PROFILE=picomq-support AWS_REGION=us-east-1

mvn -f flink-job -q -DskipTests package
cp terraform/backend.hcl.example terraform/backend.hcl
terraform -chdir=terraform init -backend-config=backend.hcl
terraform -chdir=terraform apply -auto-approve

Use the chat and agent pages, then:

./verify.sh

Stack

The example ships its own Terraform. It reads the harness state from the same S3 bucket and never modifies it.

Read from harness	Used for
vpc_id, private_subnet_ids	Flink security group and vpc_configuration
endpoints["1"]	PICO_HTTP for prefix discovery through the ALB
kafka_bootstrap	KAFKA_BOOTSTRAP, kafka.picomq.internal:9092
bootstrap_secret_arn	PICO_TOKEN_SECRET_ARN, fetched by the job at start

Created	Setting
S3 Tables bucket, namespace agents, two tables	schemas above, created before the job starts
Versioned code bucket and jar object	rebuilt jar redeploys the app on the next apply
Managed Flink application	FLINK-2_3, parallelism 1, snapshots on, private subnets
IAM role	jar read, s3tables:* on the bucket, GetSecretValue on the bootstrap secret, logs, ENI

Job

flink-job/src/main/java/picomq/example/AgentsJob.java:

Piece	Setting
Discovery	HttpKafkaMetadataService polls GET /?prefix=/examples/agents/ai-sdk/ with the bootstrap token, 10s
Source	DynamicKafkaSource, committed group offsets, earliest on a cold start
Routing	topic examples.agents.ai-sdk.chat.* to conversations, agent.* to agent_events
Sink	IcebergSink per table, REST catalog https://s3tables.<region>.amazonaws.com/iceberg, sigv4
Checkpoint	60s, one Iceberg commit per checkpoint
Config	MSF property group picomq, env fallback for local runs

Query

sql/duckdb.sql through verify.sh, attached with ENDPOINT_TYPE s3_tables:

SELECT stream AS run,
       count(*) FILTER (type = 'step') AS steps,
       string_agg(tools, ',') FILTER (tools IS NOT NULL) AS tools,
       max(total_tokens) AS total_tokens
FROM lake.agents.agent_events
GROUP BY stream;

Athena needs the S3 Tables catalog mounted once per account and region:

aws glue create-catalog --name s3tablescatalog --catalog-input '{
  "FederatedCatalog": {"Identifier": "arn:aws:s3tables:<region>:<account>:bucket/*", "ConnectionName": "aws:s3tables"},
  "CreateDatabaseDefaultPermissions": [{"Principal": {"DataLakePrincipalIdentifier": "IAM_ALLOWED_PRINCIPALS"}, "Permissions": ["ALL"]}],
  "CreateTableDefaultPermissions": [{"Principal": {"DataLakePrincipalIdentifier": "IAM_ALLOWED_PRINCIPALS"}, "Permissions": ["ALL"]}],
  "AllowFullTableExternalDataAccess": "True"
}'

SELECT * FROM "s3tablescatalog/<table bucket>"."agents"."conversations" ORDER BY stream, seq

Restarts

Snapshots are enabled on the application and the source starts from committed group offsets, so a redeploy or a new jar resumes where it left off. No rows are re-appended.

Teardown

terraform -chdir=terraform destroy -auto-approve

Then the harness. PicoMQ also has an Iceberg sink connector, see Fleet telematics to Iceberg. This example goes through the Kafka protocol into a third-party engine on purpose.