starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-18 16:02:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: add auto_commit option to Kafka triggers with advanced UI badges (#8317)

Browse Source 
* feat: add auto_commit option to Kafka triggers with manual commit API

Add ability to disable auto-commit on Kafka triggers so users can
manually commit offsets after processing messages. This prevents
message loss when processing fails.

Changes:
- Add `auto_commit` column to kafka_trigger table (default true)
- Add POST /kafka_triggers/commit_offsets/{path} endpoint using
  BaseConsumer with manual assign() to avoid rebalance
- Enrich trigger_info payload with partition and offset fields
- Conditionally commit based on auto_commit setting
- Add auto-commit toggle to frontend Kafka trigger config
- Add commitKafkaOffsets helpers to Python and TypeScript SDKs
- Add integration tests for auto_commit DB defaults

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* feat: use DB-based pending commits for kafka manual offset commit

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* feat: pass trigger_path to all v2 preprocessors, secure commit_offsets endpoint, fix commit semantics

- Add trigger_path to v2 preprocessor event for all trigger types (kafka, nats, sqs, mqtt, gcp, postgres, websocket, http, email)
- Secure commit_offsets endpoint: infer trigger from job token (OptJobAuthed) instead of requiring trigger path parameter
- Fix auto_commit: only commit offset after successful job push
- Fix pending commits: commit offset+1 (Kafka semantics) and use CommitMode::Sync
- Update TS/Python clients and frontend preprocessor templates

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* feat: add advanced section badges and reorganize kafka trigger settings

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: remove dead wm_trigger assertions from kafka e2e test

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* sqlx

* refactor: remove unused advancedCollapsed state from all trigger editors

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* update ref

* chore: update ee-repo-ref to ed2c9d360e6fab866b9744cc79f50038d1fc7152

This commit updates the EE repository reference after PR #452 was merged in windmill-ee-private.

Previous ee-repo-ref: 5b31116a1d5a042c6a780732901cfd89584d1773

New ee-repo-ref: ed2c9d360e6fab866b9744cc79f50038d1fc7152

Automated by sync-ee-ref workflow.

* fix: use path-based auth for kafka commit_offsets endpoint

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* chore: update ee-repo-ref to fcd3ea52b0cc94fbe1159baf662a38da947456de

This commit updates the EE repository reference after PR #457 was merged in windmill-ee-private.

Previous ee-repo-ref: b3a5c33c92cb1b2caf7a65986d71da291ff72a35

New ee-repo-ref: fcd3ea52b0cc94fbe1159baf662a38da947456de

Automated by sync-ee-ref workflow.

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
Co-authored-by: windmill-internal-app[bot] <windmill-internal-app[bot]@users.noreply.github.com>
Co-authored-by: Ruben Fiszel <ruben@windmill.dev>
This commit is contained in:
hugocasa
2026-03-12 15:00:30 +01:00
committed by GitHub
parent d2b9799ac4
commit ec20d76216
43 changed files with 677 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-038d2fde90fa9e99e30d15161777fa3ab402e33edfca46daa95b52e525424586.json
Generated
+41
View File
@@ -0,0 +1,41 @@
{
"db_name": "PostgreSQL",
"query": "SELECT id, topic, partition, \"offset\" FROM kafka_pending_commits\n WHERE workspace_id = $1 AND kafka_trigger_path = $2\n ORDER BY id",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Int8"
},
{
"ordinal": 1,
"name": "topic",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "partition",
"type_info": "Int4"
},
{
"ordinal": 3,
"name": "offset",
"type_info": "Int8"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
false,
false,
false,
false
]
},
"hash": "038d2fde90fa9e99e30d15161777fa3ab402e33edfca46daa95b52e525424586"
}
backend/.sqlx/query-12631fecee6aa11a45cf5c8d101c0dd8de50ac9b57e68198f637038d344fdd46.json → backend/.sqlx/query-072e5ab78f929c6b7264f98c1588cb24cc635836276ee6faa2438f494bfbce04.json
Generated
+3 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE kafka_trigger\n SET\n kafka_resource_path = $1,\n group_id = $2,\n topics = $3,\n filters = $4,\n auto_offset_reset = $5,\n script_path = $6,\n path = $7,\n is_flow = $8,\n edited_by = $9,\n email = $10,\n edited_at = now(),\n server_id = NULL,\n error = NULL,\n error_handler_path = $13,\n error_handler_args = $14,\n retry = $15\n WHERE\n workspace_id = $11 AND path = $12\n ",
"query": "\n UPDATE kafka_trigger\n SET\n kafka_resource_path = $1,\n group_id = $2,\n topics = $3,\n filters = $4,\n auto_offset_reset = $5,\n auto_commit = $6,\n script_path = $7,\n path = $8,\n is_flow = $9,\n edited_by = $10,\n email = $11,\n edited_at = now(),\n server_id = NULL,\n error = NULL,\n error_handler_path = $14,\n error_handler_args = $15,\n retry = $16\n WHERE\n workspace_id = $12 AND path = $13\n ",
"describe": {
"columns": [],
"parameters": {
@@ -10,6 +10,7 @@
"VarcharArray",
"JsonbArray",
"Varchar",
"Bool",
"Varchar",
"Varchar",
"Bool",
@@ -24,5 +25,5 @@
},
"nullable": []
},
"hash": "12631fecee6aa11a45cf5c8d101c0dd8de50ac9b57e68198f637038d344fdd46"
"hash": "072e5ab78f929c6b7264f98c1588cb24cc635836276ee6faa2438f494bfbce04"
}
backend/.sqlx/query-1df610a583e86edb70c374fd66c68554a6a4291426c09dd5b04fd832f9d31208.json
Generated
+23
View File
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "SELECT reset_offset FROM kafka_trigger WHERE workspace_id = $1 AND path = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "reset_offset",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
false
]
},
"hash": "1df610a583e86edb70c374fd66c68554a6a4291426c09dd5b04fd832f9d31208"
}
backend/.sqlx/query-45fc21026fa76e5d69f00a68a7be81abb3ec627578f2d14f0ce33896dc6ab4cf.json
Generated
+23
View File
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO kafka_trigger (\n path, kafka_resource_path, topics, group_id, script_path,\n is_flow, workspace_id, edited_by, email, auto_commit\n )\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"VarcharArray",
"Varchar",
"Varchar",
"Bool",
"Varchar",
"Varchar",
"Varchar",
"Bool"
]
},
"nullable": []
},
"hash": "45fc21026fa76e5d69f00a68a7be81abb3ec627578f2d14f0ce33896dc6ab4cf"
}
backend/.sqlx/query-4b2a29b3ef7ec4802d81ec4b706623b991c938e40d0db25290b03dc0577c2740.json
Generated
+23
View File
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "SELECT auto_commit FROM kafka_trigger WHERE workspace_id = $1 AND path = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "auto_commit",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
false
]
},
"hash": "4b2a29b3ef7ec4802d81ec4b706623b991c938e40d0db25290b03dc0577c2740"
}
backend/.sqlx/query-7e3bfb33fb771aec39b43a7550091ce7c9b1261b52d10f4a7f3273fed3c916df.json → backend/.sqlx/query-4cf4be7a981173d3f242887d9313c7e60d23e9827f23c0de5b546ed56697d54a.json
Generated
+20 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT kafka_resource_path, topics, group_id, mode AS \"mode: String\"\n FROM kafka_trigger\n WHERE workspace_id = $1 AND path = $2\n ",
"query": "\n SELECT kafka_resource_path, topics, group_id, mode AS \"mode: String\",\n auto_offset_reset, auto_commit, reset_offset\n FROM kafka_trigger\n WHERE workspace_id = $1 AND path = $2\n ",
"describe": {
"columns": [
{
@@ -33,6 +33,21 @@
}
}
}
},
{
"ordinal": 4,
"name": "auto_offset_reset",
"type_info": "Varchar"
},
{
"ordinal": 5,
"name": "auto_commit",
"type_info": "Bool"
},
{
"ordinal": 6,
"name": "reset_offset",
"type_info": "Bool"
}
],
"parameters": {
@@ -42,11 +57,14 @@
]
},
"nullable": [
false,
false,
false,
false,
false,
false,
false
]
},
"hash": "7e3bfb33fb771aec39b43a7550091ce7c9b1261b52d10f4a7f3273fed3c916df"
"hash": "4cf4be7a981173d3f242887d9313c7e60d23e9827f23c0de5b546ed56697d54a"
}
backend/.sqlx/query-50807b807bb901a380926798be655c13a18dfd26e237a8218d3006e2898b5aa3.json
Generated
+23
View File
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT auto_commit\n FROM kafka_trigger\n WHERE workspace_id = $1 AND path = $2\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "auto_commit",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
false
]
},
"hash": "50807b807bb901a380926798be655c13a18dfd26e237a8218d3006e2898b5aa3"
}
backend/.sqlx/query-4b5a711986017654bdd495893a16ddc6ab09c98cd8723865cbd341404bc6a02f.json → backend/.sqlx/query-5dd6315ec270c268e905262e4b0a920837354d91a0ae16b1236c1267da71765f.json
Generated
+3 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO kafka_trigger (\n workspace_id,\n path,\n kafka_resource_path,\n group_id,\n topics,\n filters,\n auto_offset_reset,\n script_path,\n is_flow,\n mode,\n edited_by,\n email,\n edited_at,\n error_handler_path,\n error_handler_args,\n retry\n ) VALUES (\n $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, now(), $13, $14, $15\n )\n ",
"query": "\n INSERT INTO kafka_trigger (\n workspace_id,\n path,\n kafka_resource_path,\n group_id,\n topics,\n filters,\n auto_offset_reset,\n auto_commit,\n script_path,\n is_flow,\n mode,\n edited_by,\n email,\n edited_at,\n error_handler_path,\n error_handler_args,\n retry\n ) VALUES (\n $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, now(), $14, $15, $16\n )\n ",
"describe": {
"columns": [],
"parameters": {
@@ -12,6 +12,7 @@
"VarcharArray",
"JsonbArray",
"Varchar",
"Bool",
"Varchar",
"Bool",
{
@@ -35,5 +36,5 @@
},
"nullable": []
},
"hash": "4b5a711986017654bdd495893a16ddc6ab09c98cd8723865cbd341404bc6a02f"
"hash": "5dd6315ec270c268e905262e4b0a920837354d91a0ae16b1236c1267da71765f"
}
backend/.sqlx/query-80bad96cbec6b5eca57a6380e7515565490a271050dcc4b5aac2b730ae3a55b9.json
Generated
+14
View File
@@ -0,0 +1,14 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM kafka_pending_commits WHERE id = ANY($1)",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Int8Array"
]
},
"nullable": []
},
"hash": "80bad96cbec6b5eca57a6380e7515565490a271050dcc4b5aac2b730ae3a55b9"
}
backend/.sqlx/query-c2f38c9e09aac73d10e8f327715927c07832badb2c9145d5996b829163bdf7d9.json
Generated
+23
View File
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE kafka_trigger SET reset_offset = true, server_id = NULL WHERE workspace_id = $1 AND path = $2 RETURNING true",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
null
]
},
"hash": "c2f38c9e09aac73d10e8f327715927c07832badb2c9145d5996b829163bdf7d9"
}
backend/.sqlx/query-ef15599f532fab2cbb487542ffec047cf3b7ce22ce868db1b1a63e6c10d0d12b.json
Generated
+15
View File
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE kafka_trigger SET reset_offset = false WHERE workspace_id = $1 AND path = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": []
},
"hash": "ef15599f532fab2cbb487542ffec047cf3b7ce22ce868db1b1a63e6c10d0d12b"
}
backend/.sqlx/query-f67e5c96eb9cb35953d4c3e83e0fcbb5b647737e0366529a2f418218b1a74679.json
Generated
+18
View File
@@ -0,0 +1,18 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO kafka_pending_commits (workspace_id, kafka_trigger_path, topic, partition, \"offset\")\n VALUES ($1, $2, $3, $4, $5)",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"Varchar",
"Int4",
"Int8"
]
},
"nullable": []
},
"hash": "f67e5c96eb9cb35953d4c3e83e0fcbb5b647737e0366529a2f418218b1a74679"
}
backend/Cargo.lock
Generated
+3 -3
View File
@@ -16870,7 +16870,7 @@ dependencies = [
[[package]]
name = "windmill-parser-py-asset"
version = "1.653.0"
version = "1.654.0"
dependencies = [
"anyhow",
"rustpython-ast",
@@ -16950,7 +16950,7 @@ dependencies = [
[[package]]
name = "windmill-parser-sql-asset"
version = "1.653.0"
version = "1.654.0"
dependencies = [
"anyhow",
"serde",
@@ -16980,7 +16980,7 @@ dependencies = [
[[package]]
name = "windmill-parser-ts-asset"
version = "1.653.0"
version = "1.654.0"
dependencies = [
"anyhow",
"serde-wasm-bindgen",
backend/ee-repo-ref.txt
+1 -1
View File
@@ -1 +1 @@
85bcbfc8e952842c555156d1050b2a024e7e37a3
fcd3ea52b0cc94fbe1159baf662a38da947456de
backend/migrations/20260312000000_kafka_auto_commit.down.sql
+2
View File
@@ -0,0 +1,2 @@
DROP TABLE IF EXISTS kafka_pending_commits;
ALTER TABLE kafka_trigger DROP COLUMN auto_commit;
backend/migrations/20260312000000_kafka_auto_commit.up.sql
+14
View File
@@ -0,0 +1,14 @@
ALTER TABLE kafka_trigger ADD COLUMN auto_commit BOOLEAN NOT NULL DEFAULT TRUE;
CREATE TABLE kafka_pending_commits (
id BIGSERIAL PRIMARY KEY,
workspace_id VARCHAR(50) NOT NULL,
kafka_trigger_path VARCHAR(255) NOT NULL,
topic VARCHAR(255) NOT NULL,
partition INTEGER NOT NULL,
"offset" BIGINT NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
FOREIGN KEY (workspace_id, kafka_trigger_path) REFERENCES kafka_trigger(workspace_id, path) ON DELETE CASCADE
);
CREATE INDEX idx_kafka_pending_commits_trigger ON kafka_pending_commits (workspace_id, kafka_trigger_path);
backend/summarized_schema.txt
+3 -1
View File
@@ -109,7 +109,9 @@ job_result_stream_v2: job_id(uuid), workspace_id(text), stream(text), idx(int)
job_settings: job_id(uuid), runnable_settings(bigint)
job_stats: workspace_id(char), job_id(uuid), metric_id(char), metric_name(char), metric_kind(metric_kind), scalar_int(int), scalar_float(float), timestamps(ts), timeseries_int(int[]), timeseries_float(float[])
FK: (workspace_id) -> workspace(id)
kafka_trigger: path(char), kafka_resource_path(char), topics(char), group_id(char), script_path(char), is_flow(bool), workspace_id(char), edited_by(char), email(char), edited_at(ts), extra_perms(jsonb), server_id(char), last_server_ping(ts), error(text), error_handler_path(char), error_handler_args(jsonb), retry(jsonb), mode(trigger_mode), filters(jsonb[])
kafka_pending_commits: id(bigint), workspace_id(char), kafka_trigger_path(char), topic(char), partition(int), offset(bigint), created_at(ts)
FK: (workspace_id, kafka_trigger_path) -> kafka_trigger(workspace_id, path)
kafka_trigger: path(char), kafka_resource_path(char), topics(char), group_id(char), script_path(char), is_flow(bool), workspace_id(char), edited_by(char), email(char), edited_at(ts), extra_perms(jsonb), server_id(char), last_server_ping(ts), error(text), error_handler_path(char), error_handler_args(jsonb), retry(jsonb), mode(trigger_mode), filters(jsonb[]), auto_commit(bool)
log_file: hostname(char), log_ts(ts), ok_lines(bigint), err_lines(bigint), mode(log_mode), worker_group(char), file_path(char), json_fmt(bool)
magic_link: email(char), token(char), expiration(ts)
mcp_oauth_client: mcp_server_url(text), client_id(text), client_secret(text), client_secret_expires_at(ts), token_endpoint(text), created_at(ts)
backend/windmill-api-integration-tests/tests/trigger_e2e.rs
+10 -8
View File
@@ -251,7 +251,8 @@ async fn test_websocket_e2e(db: Pool<Postgres>) -> anyhow::Result<()> {
"test-workspace",
"test-user",
"test@windmill.dev",
&[json!({"type": "RawMessage", "content": "hello from e2e test"})] as &[serde_json::Value],
&[json!({"type": "RawMessage", "content": "hello from e2e test"})]
as &[serde_json::Value],
)
.execute(&db)
.await?;
@@ -302,9 +303,11 @@ async fn test_postgres_e2e(db: Pool<Postgres>) -> anyhow::Result<()> {
sqlx::query("CREATE TABLE test_trigger_table (id serial PRIMARY KEY, data text)")
.execute(&db)
.await?;
sqlx::query(&format!("CREATE PUBLICATION {pub_name} FOR TABLE test_trigger_table"))
.execute(&db)
.await?;
sqlx::query(&format!(
"CREATE PUBLICATION {pub_name} FOR TABLE test_trigger_table"
))
.execute(&db)
.await?;
sqlx::query(&format!(
"SELECT pg_create_logical_replication_slot('{slot_name}', 'pgoutput')"
))
@@ -313,10 +316,9 @@ async fn test_postgres_e2e(db: Pool<Postgres>) -> anyhow::Result<()> {
// Extract the test DB name from the pool so the resource points here,
// not at the main windmill database.
let test_db_name: String =
sqlx::query_scalar("SELECT current_database()")
.fetch_one(&db)
.await?;
let test_db_name: String = sqlx::query_scalar("SELECT current_database()")
.fetch_one(&db)
.await?;
insert_resource(
&db,
backend/windmill-api-integration-tests/tests/triggers.rs
+56 -14
View File
@@ -214,12 +214,9 @@ async fn test_capture_delete(db: Pool<Postgres>) -> anyhow::Result<()> {
.execute(&db)
.await?;
let count = sqlx::query_scalar!(
"SELECT COUNT(*) FROM capture WHERE id = $1",
id,
)
.fetch_one(&db)
.await?;
let count = sqlx::query_scalar!("SELECT COUNT(*) FROM capture WHERE id = $1", id,)
.fetch_one(&db)
.await?;
assert_eq!(count, Some(0));
@@ -385,7 +382,10 @@ async fn test_capture_api_list_captures(db: Pool<Postgres>) -> anyhow::Result<()
.send()
.await?;
assert!(response.status().is_success(), "list captures should succeed");
assert!(
response.status().is_success(),
"list captures should succeed"
);
let captures: Vec<CaptureResponse> = response.json().await?;
assert_eq!(captures.len(), 3);
@@ -480,12 +480,9 @@ async fn test_capture_api_delete(db: Pool<Postgres>) -> anyhow::Result<()> {
assert!(response.status().is_success(), "delete should succeed");
let count = sqlx::query_scalar!(
"SELECT COUNT(*) FROM capture WHERE id = $1",
id,
)
.fetch_one(&db)
.await?;
let count = sqlx::query_scalar!("SELECT COUNT(*) FROM capture WHERE id = $1", id,)
.fetch_one(&db)
.await?;
assert_eq!(count, Some(0));
@@ -933,7 +930,8 @@ async fn test_kafka_trigger_insert(db: Pool<Postgres>) -> anyhow::Result<()> {
let trigger = sqlx::query!(
r#"
SELECT kafka_resource_path, topics, group_id, mode AS "mode: String"
SELECT kafka_resource_path, topics, group_id, mode AS "mode: String",
auto_offset_reset, auto_commit, reset_offset
FROM kafka_trigger
WHERE workspace_id = $1 AND path = $2
"#,
@@ -947,6 +945,50 @@ async fn test_kafka_trigger_insert(db: Pool<Postgres>) -> anyhow::Result<()> {
assert_eq!(trigger.topics, vec!["topic-a", "topic-b"]);
assert_eq!(trigger.group_id, "my-consumer-group");
assert_eq!(trigger.mode, "enabled");
assert_eq!(trigger.auto_offset_reset, "latest");
assert_eq!(trigger.auto_commit, true);
assert_eq!(trigger.reset_offset, false);
Ok(())
}
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
async fn test_kafka_trigger_insert_auto_commit_disabled(db: Pool<Postgres>) -> anyhow::Result<()> {
sqlx::query!(
r#"
INSERT INTO kafka_trigger (
path, kafka_resource_path, topics, group_id, script_path,
is_flow, workspace_id, edited_by, email, auto_commit
)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
"#,
"f/test/kafka_trigger_no_commit",
"u/admin/kafka_resource",
&["topic-c"] as &[&str],
"my-consumer-group-2",
"f/test/kafka_handler",
false,
"test-workspace",
"test-user",
"test@windmill.dev",
false,
)
.execute(&db)
.await?;
let trigger = sqlx::query!(
r#"
SELECT auto_commit
FROM kafka_trigger
WHERE workspace_id = $1 AND path = $2
"#,
"test-workspace",
"f/test/kafka_trigger_no_commit",
)
.fetch_one(&db)
.await?;
assert_eq!(trigger.auto_commit, false);
Ok(())
}
backend/windmill-api/openapi.yaml
+45
View File
@@ -12017,6 +12017,39 @@ paths:
"200":
description: kafka trigger offsets reset successfully
/w/{workspace}/kafka_triggers/commit_offsets/{path}:
post:
summary: commit kafka offsets for a specific trigger
operationId: commitKafkaOffsets
tags:
- kafka_trigger
parameters:
- $ref: "#/components/parameters/WorkspaceId"
- $ref: "#/components/parameters/Path"
requestBody:
description: offsets to commit
required: true
content:
application/json:
schema:
type: object
properties:
topic:
type: string
partition:
type: integer
format: int32
offset:
type: integer
format: int64
required:
- topic
- partition
- offset
responses:
"200":
description: kafka offsets committed successfully
/w/{workspace}/nats_triggers/create:
post:
summary: create nats trigger
@@ -22098,6 +22131,10 @@ components:
- earliest
default: latest
description: "Initial offset behavior when consumer group has no committed offset. 'latest' starts from new messages only, 'earliest' starts from the beginning."
auto_commit:
type: boolean
default: true
description: "When true (default), offsets are committed automatically after receiving each message. When false, you must manually commit offsets using the commit_offsets endpoint."
server_id:
type: string
description: ID of the server currently handling this trigger (internal)
@@ -22165,6 +22202,10 @@ components:
- earliest
default: latest
description: "Initial offset behavior when consumer group has no committed offset."
auto_commit:
type: boolean
default: true
description: "When true (default), offsets are committed automatically after receiving each message. When false, you must manually commit offsets using the commit_offsets endpoint."
mode:
$ref: "#/components/schemas/TriggerMode"
error_handler_path:
@@ -22224,6 +22265,10 @@ components:
- earliest
default: latest
description: "Initial offset behavior when consumer group has no committed offset."
auto_commit:
type: boolean
default: true
description: "When true (default), offsets are committed automatically after receiving each message. When false, you must manually commit offsets using the commit_offsets endpoint."
path:
type: string
description: The unique path identifier for this trigger
backend/windmill-api/src/capture.rs
+1
View File
@@ -1012,6 +1012,7 @@ async fn http_payload(
.to_v2_preprocessor_args(
&http_trigger_config.route_path,
&route_path,
"",
&params,
headers,
query,
backend/windmill-api/src/triggers/http/handler.rs
+1
View File
@@ -468,6 +468,7 @@ async fn route_job(
.to_args_from_format(
&trigger.route_path,
&called_path,
&trigger.path,
&params,
runnable_format,
trigger.wrap_body,
backend/windmill-api/src/triggers/http/http_trigger_args.rs
+12 -1
View File
@@ -68,6 +68,7 @@ struct HttpTriggerPreprocessorEvent<'a> {
kind: String,
route: &'a str,
path: &'a str,
trigger_path: &'a str,
body: Box<RawValue>,
raw_string: Option<String>,
params: &'a HashMap<String, String>,
@@ -117,6 +118,7 @@ impl HttpTriggerArgs {
self,
route_path: &str,
called_path: &str,
trigger_path: &str,
params: &HashMap<String, String>,
format: RunnableFormat,
wrap_body: bool,
@@ -126,7 +128,14 @@ impl HttpTriggerArgs {
match format {
RunnableFormat { has_preprocessor: true, version: RunnableFormatVersion::V2 } => {
// we don't care about wrap_body in v2
self.to_v2_preprocessor_args(route_path, called_path, params, headers, query)
self.to_v2_preprocessor_args(
route_path,
called_path,
trigger_path,
params,
headers,
query,
)
}
RunnableFormat { has_preprocessor: true, version: RunnableFormatVersion::V1 } => self
.to_v1_preprocessor_args(
@@ -177,6 +186,7 @@ impl HttpTriggerArgs {
self,
route_path: &str,
called_path: &str,
trigger_path: &str,
params: &HashMap<String, String>,
headers: HashMap<String, Box<RawValue>>,
query: HashMap<String, Box<RawValue>>,
@@ -193,6 +203,7 @@ impl HttpTriggerArgs {
method: (&self.0.metadata.method).try_into()?,
route: route_path,
path: called_path,
trigger_path,
params,
}),
);
backend/windmill-trigger-websocket/src/listener.rs
+2 -1
View File
@@ -322,7 +322,7 @@ impl Listener for WebsocketTrigger {
db: &DB,
listening_trigger: &ListeningTrigger<Self::TriggerConfig>,
payload: Self::Payload,
trigger_info: HashMap<String, Box<RawValue>>,
mut trigger_info: HashMap<String, Box<RawValue>>,
extra: Option<Self::Extra>,
) -> Result<()> {
let ListeningTrigger {
@@ -338,6 +338,7 @@ impl Listener for WebsocketTrigger {
let WebsocketConfig { url, .. } = trigger_config;
trigger_info.insert("trigger_path".to_string(), to_raw_value(path));
let args = WebsocketTrigger::build_job_args(
&script_path,
*is_flow,
backend/windmill-trigger/src/listener.rs
+11 -1
View File
@@ -21,6 +21,7 @@ use windmill_common::{
jobs::JobTriggerKind,
triggers::{TriggerKind, TriggerMetadata},
utils::report_critical_error,
worker::to_raw_value,
DB, INSTANCE_NAME,
};
@@ -467,9 +468,13 @@ pub trait Listener: TriggerCrud + TriggerJobArgs {
db: &DB,
listening_trigger: &ListeningTrigger<Self::TriggerConfig>,
payload: Self::Payload,
trigger_info: HashMap<String, Box<RawValue>>,
mut trigger_info: HashMap<String, Box<RawValue>>,
_extra: Option<Self::Extra>,
) -> Result<()> {
trigger_info.insert(
"trigger_path".to_string(),
to_raw_value(&listening_trigger.path),
);
let args = Self::build_job_args(
&listening_trigger.script_path,
listening_trigger.is_flow,
@@ -552,6 +557,11 @@ pub trait Listener: TriggerCrud + TriggerJobArgs {
return Ok(());
}
let mut trigger_info = trigger_info;
trigger_info.insert(
"trigger_path".to_string(),
to_raw_value(&listening_trigger.path),
);
let (main_args, preprocessor_args) = Self::build_capture_payloads(&payload, trigger_info);
if let Err(err) = insert_capture_payload(
db,
frontend/src/lib/components/triggers/TriggerAdvancedBadges.svelte
+28
View File
@@ -0,0 +1,28 @@
<script lang="ts">
import Badge from '$lib/components/Badge.svelte'
import type { Retry } from '$lib/gen'
interface Props {
error_handler_path?: string | undefined
retry?: Retry | undefined
extraBadges?: { name: string; active: boolean }[]
}
let { error_handler_path = undefined, retry = undefined, extraBadges = [] }: Props = $props()
let allBadges = $derived(
[
{ name: 'Error Handler', active: !!error_handler_path },
{ name: 'Retries', active: !!retry },
...extraBadges
].filter((b) => b.active)
)
</script>
{#if allBadges.length > 0}
<div class="flex grow min-w-0 w-full flex-wrap gap-1 ps-2">
{#each allBadges as badge}
<Badge twBgColor="bg-surface-sunken" twTextColor="text-primary">{badge.name}</Badge>
{/each}
</div>
{/if}
frontend/src/lib/components/triggers/email/EmailTriggerEditorInner.svelte
+6 -1
View File
@@ -25,6 +25,7 @@
import Tabs from '$lib/components/common/tabs/Tabs.svelte'
import Tab from '$lib/components/common/tabs/Tab.svelte'
import TriggerRetriesAndErrorHandler from '../TriggerRetriesAndErrorHandler.svelte'
import TriggerAdvancedBadges from '../TriggerAdvancedBadges.svelte'
import { saveEmailTriggerFromCfg } from './utils'
import { deepEqual } from 'fast-equals'
import TriggerSuspendedJobsAlert from '../TriggerSuspendedJobsAlert.svelte'
@@ -370,7 +371,10 @@
/>
<Section label="Advanced" collapsable>
<div class="flex flex-col gap-4">
{#snippet header()}
<TriggerAdvancedBadges {error_handler_path} {retry} />
{/snippet}
<div class="flex flex-col gap-6">
<div class="min-h-96">
<Tabs bind:selected={optionTabSelected}>
<Tab value="error_handler" label="Error Handler" />
@@ -390,6 +394,7 @@
</div>
</div>
</Section>
<div class="pb-8" />
</div>
{/if}
{/snippet}
frontend/src/lib/components/triggers/gcp/GcpTriggerEditorInner.svelte
+8 -1
View File
@@ -31,6 +31,7 @@
import Tabs from '$lib/components/common/tabs/Tabs.svelte'
import Tab from '$lib/components/common/tabs/Tab.svelte'
import TriggerRetriesAndErrorHandler from '../TriggerRetriesAndErrorHandler.svelte'
import TriggerAdvancedBadges from '../TriggerAdvancedBadges.svelte'
import Subsection from '$lib/components/Subsection.svelte'
import Toggle from '$lib/components/Toggle.svelte'
@@ -452,7 +453,12 @@
/>
<Section label="Advanced" collapsable>
<div class="flex flex-col gap-4">
{#snippet header()}
<TriggerAdvancedBadges {error_handler_path} {retry} extraBadges={[
{ name: 'Manual Ack', active: !auto_acknowledge_msg }
]} />
{/snippet}
<div class="flex flex-col gap-6">
<div class="min-h-96">
<Tabs bind:selected={optionTabSelected}>
<Tab value="settings" label="Settings" />
@@ -520,6 +526,7 @@
</div>
</div>
</Section>
<div class="pb-8" />
</div>
{/if}
{/snippet}
frontend/src/lib/components/triggers/http/RouteEditorInner.svelte
+9
View File
@@ -52,6 +52,7 @@
import Tabs from '$lib/components/common/tabs/Tabs.svelte'
import Tab from '$lib/components/common/tabs/Tab.svelte'
import TriggerRetriesAndErrorHandler from '../TriggerRetriesAndErrorHandler.svelte'
import TriggerAdvancedBadges from '../TriggerAdvancedBadges.svelte'
import { deepEqual } from 'fast-equals'
import TriggerSuspendedJobsAlert from '../TriggerSuspendedJobsAlert.svelte'
import TriggerSuspendedJobsModal from '../TriggerSuspendedJobsModal.svelte'
@@ -695,6 +696,13 @@
{#if !is_static_website}
<Section label="Advanced" collapsable>
{#snippet header()}
<TriggerAdvancedBadges {error_handler_path} {retry} extraBadges={[
{ name: 'Async', active: request_type === 'async' },
{ name: 'SSE', active: request_type === 'sync_sse' },
{ name: 'Authentication', active: authentication_method !== 'none' }
]} />
{/snippet}
<div class="min-h-96">
<Tabs bind:selected={optionTabSelected}>
<Tab value="request_options" label="Request Options" />
@@ -909,6 +917,7 @@
</div>
</div>
</Section>
<div class="pb-8" />
{/if}
</div>
{/if}
frontend/src/lib/components/triggers/kafka/KafkaTriggerEditorInner.svelte
+82 -30
View File
@@ -6,12 +6,7 @@
import Path from '$lib/components/Path.svelte'
import Required from '$lib/components/Required.svelte'
import ScriptPicker from '$lib/components/ScriptPicker.svelte'
import {
KafkaTriggerService,
type ErrorHandler,
type Retry,
type TriggerMode
} from '$lib/gen'
import { KafkaTriggerService, type ErrorHandler, type Retry, type TriggerMode } from '$lib/gen'
import { usedTriggerKinds, userStore, workspaceStore } from '$lib/stores'
import { canWrite, capitalize, emptyString, sendUserToast } from '$lib/utils'
import Section from '$lib/components/Section.svelte'
@@ -25,10 +20,13 @@
import Tabs from '$lib/components/common/tabs/Tabs.svelte'
import Tab from '$lib/components/common/tabs/Tab.svelte'
import TriggerRetriesAndErrorHandler from '../TriggerRetriesAndErrorHandler.svelte'
import TriggerAdvancedBadges from '../TriggerAdvancedBadges.svelte'
import { deepEqual } from 'fast-equals'
import TriggerSuspendedJobsAlert from '../TriggerSuspendedJobsAlert.svelte'
import TriggerSuspendedJobsModal from '../TriggerSuspendedJobsModal.svelte'
import TriggerFilters from '../TriggerFilters.svelte'
import Select from '$lib/components/select/Select.svelte'
import Toggle from '$lib/components/Toggle.svelte'
interface Props {
useDrawer?: boolean
@@ -87,6 +85,7 @@
let kafkaResourcePath = $state('')
let kafkaCfg: Record<string, any> = $state({})
let autoOffsetReset = $state('latest')
let autoCommit = $state(true)
let deploymentLoading = $state(false)
let resetLoading = $state(false)
let optionTabSelected: 'error_handler' | 'retries' = $state('error_handler')
@@ -176,6 +175,7 @@
topics: nDefaultValues?.topics ?? ['']
}
autoOffsetReset = nDefaultValues?.auto_offset_reset ?? 'latest'
autoCommit = nDefaultValues?.auto_commit ?? true
initialScriptPath = ''
fixedScriptPath = fixedScriptPath_ ?? ''
script_path = fixedScriptPath
@@ -207,6 +207,7 @@
topics: cfg?.topics
}
autoOffsetReset = cfg?.auto_offset_reset ?? 'latest'
autoCommit = cfg?.auto_commit ?? true
mode = cfg?.mode ?? 'enabled'
extra_perms = cfg?.extra_perms
can_write = canWrite(path, cfg?.extra_perms, $userStore)
@@ -240,6 +241,7 @@
topics: kafkaCfg.topics,
filters,
auto_offset_reset: autoOffsetReset,
auto_commit: autoCommit,
mode,
extra_perms: extra_perms,
error_handler_path,
@@ -481,36 +483,85 @@
bind:kafkaCfgValid
bind:kafkaResourcePath
bind:kafkaCfg
bind:autoOffsetReset
{path}
{can_write}
showTestingBadge={isEditor}
/>
{#if edit && can_write}
<Label label="Consumer offset">
{#snippet header()}
<span class="text-2xs text-tertiary ml-2">
Force re-read all messages from the beginning
</span>
{/snippet}
<Button
variant="default"
size="xs"
startIcon={{ icon: RotateCcw }}
disabled={resetLoading}
loading={resetLoading}
onclick={() => (resetConfirmOpen = true)}
>
Reset offset to earliest
</Button>
</Label>
{/if}
<TriggerFilters bind:filters disabled={!can_write} />
<Section label="Advanced" collapsable>
<div class="flex flex-col gap-4">
{#snippet header()}
<TriggerAdvancedBadges
{error_handler_path}
{retry}
extraBadges={[
{ name: 'Earliest offset', active: autoOffsetReset !== 'latest' },
{ name: 'Manual commit', active: !autoCommit },
{ name: 'Filters', active: filters.length > 0 }
]}
/>
{/snippet}
<div class="flex flex-col gap-6">
<Label label="Initial offset">
{#snippet header()}
<span class="text-2xs text-tertiary ml-2">
Only applies when no committed offset exists
</span>
{/snippet}
<Select
items={[
{ label: 'Latest (new messages only)', value: 'latest' },
{ label: 'Earliest (from beginning)', value: 'earliest' }
]}
bind:value={autoOffsetReset}
disabled={!can_write}
/>
</Label>
<div class="flex flex-col gap-2">
<Label label="Auto-commit offsets">
{#snippet header()}
<span class="text-2xs text-tertiary ml-2">
Automatically commit offsets after receiving each message
</span>
{/snippet}
<Toggle bind:checked={autoCommit} disabled={!can_write} />
</Label>
{#if !autoCommit}
<Alert title="Manual commit mode" type="info" size="xs">
Offsets will not be committed automatically. Use <code
>wmill.commit_kafka_offsets(trigger_path, topic, partition, offset)</code
>
in Python or <code
>wmill.commitKafkaOffsets(triggerPath, topic, partition, offset)</code
> in TypeScript with the values from the event payload. The consumer collects
all pending commits and commits the highest offset for each topic/partition
pair.
</Alert>
{/if}
</div>
{#if edit && can_write}
<Label label="Consumer offset">
{#snippet header()}
<span class="text-2xs text-tertiary ml-2">
Force re-read all messages from the beginning
</span>
{/snippet}
<Button
variant="default"
size="xs"
startIcon={{ icon: RotateCcw }}
disabled={resetLoading}
loading={resetLoading}
onclick={() => (resetConfirmOpen = true)}
>
Reset offset to earliest
</Button>
</Label>
{/if}
<TriggerFilters bind:filters disabled={!can_write} />
<div class="min-h-96">
<Tabs bind:selected={optionTabSelected}>
<Tab value="error_handler" label="Error Handler" />
@@ -530,6 +581,7 @@
</div>
</div>
</Section>
<div class="pb-8" />
</div>
{/if}
{/snippet}
frontend/src/lib/components/triggers/kafka/KafkaTriggersConfigSection.svelte
-23
View File
@@ -3,8 +3,6 @@
import Section from '$lib/components/Section.svelte'
import Subsection from '$lib/components/Subsection.svelte'
import SchemaForm from '../../SchemaForm.svelte'
import Select from '$lib/components/select/Select.svelte'
import Label from '$lib/components/Label.svelte'
import { workspaceStore } from '$lib/stores'
import TestTriggerConnection from '../TestTriggerConnection.svelte'
import TestingBadge from '../testingBadge.svelte'
@@ -15,7 +13,6 @@
kafkaCfgValid?: boolean
kafkaResourcePath?: string
kafkaCfg?: Record<string, any>
autoOffsetReset?: string
can_write?: boolean
showTestingBadge?: boolean
}
@@ -25,16 +22,10 @@
kafkaCfgValid = $bindable(false),
kafkaResourcePath = $bindable(''),
kafkaCfg = $bindable({}),
autoOffsetReset = $bindable('latest'),
can_write = true,
showTestingBadge = false
}: Props = $props()
const offsetResetOptions = [
{ label: 'Latest (new messages only)', value: 'latest' },
{ label: 'Earliest (from beginning)', value: 'earliest' }
]
const kafkaConfigSchema = {
$schema: 'http://json-schema.org/draft-07/schema#',
type: 'object',
@@ -109,20 +100,6 @@
</Subsection>
</div>
<div class="block grow w-full">
<Label label="Initial offset">
{#snippet header()}
<span class="text-2xs text-tertiary ml-2">
Only applies when no committed offset exists
</span>
{/snippet}
<Select
items={offsetResetOptions}
bind:value={autoOffsetReset}
disabled={!can_write}
/>
</Label>
</div>
</div>
</Section>
</div>
frontend/src/lib/components/triggers/kafka/utils.ts
+1
View File
@@ -25,6 +25,7 @@ export async function saveKafkaTriggerFromCfg(
topics: cfg.topics,
filters: cfg.filters ?? [],
auto_offset_reset: cfg.auto_offset_reset ?? 'latest',
auto_commit: cfg.auto_commit ?? true,
...errorHandlerAndRetries
}
try {
frontend/src/lib/components/triggers/mqtt/MqttTriggerEditorInner.svelte
+9 -1
View File
@@ -29,6 +29,7 @@
import Tabs from '$lib/components/common/tabs/Tabs.svelte'
import Tab from '$lib/components/common/tabs/Tab.svelte'
import TriggerRetriesAndErrorHandler from '../TriggerRetriesAndErrorHandler.svelte'
import TriggerAdvancedBadges from '../TriggerAdvancedBadges.svelte'
import Toggle from '$lib/components/Toggle.svelte'
import ToggleButtonGroup from '$lib/components/common/toggleButton-v2/ToggleButtonGroup.svelte'
import ToggleButton from '$lib/components/common/toggleButton-v2/ToggleButton.svelte'
@@ -469,7 +470,13 @@
/>
<Section label="Advanced" collapsable>
<div class="flex flex-col gap-4">
{#snippet header()}
<TriggerAdvancedBadges {error_handler_path} {retry} extraBadges={[
{ name: 'Custom Client ID', active: !!client_id },
{ name: 'MQTT v3', active: client_version === 'v3' }
]} />
{/snippet}
<div class="flex flex-col gap-6">
<div class="min-h-96">
<Tabs bind:selected={optionTabSelected}>
<Tab value="connection_options" label="Connection Options" />
@@ -606,6 +613,7 @@
</div>
</div>
</Section>
<div class="pb-8" />
</div>
{/if}
{/snippet}
frontend/src/lib/components/triggers/nats/NatsTriggerEditorInner.svelte
+6 -1
View File
@@ -19,6 +19,7 @@
import Tabs from '$lib/components/common/tabs/Tabs.svelte'
import Tab from '$lib/components/common/tabs/Tab.svelte'
import TriggerRetriesAndErrorHandler from '../TriggerRetriesAndErrorHandler.svelte'
import TriggerAdvancedBadges from '../TriggerAdvancedBadges.svelte'
import { deepEqual } from 'fast-equals'
import TriggerSuspendedJobsAlert from '../TriggerSuspendedJobsAlert.svelte'
import TriggerSuspendedJobsModal from '../TriggerSuspendedJobsModal.svelte'
@@ -452,7 +453,10 @@
/>
<Section label="Advanced" collapsable>
<div class="flex flex-col gap-4">
{#snippet header()}
<TriggerAdvancedBadges {error_handler_path} {retry} />
{/snippet}
<div class="flex flex-col gap-6">
<div class="min-h-96">
<Tabs bind:selected={optionTabSelected}>
<Tab value="error_handler" label="Error Handler" />
@@ -472,6 +476,7 @@
</div>
</div>
</Section>
<div class="pb-8" />
</div>
{/if}
{/snippet}
frontend/src/lib/components/triggers/postgres/PostgresTriggerEditorInner.svelte
+6 -1
View File
@@ -36,6 +36,7 @@
import TestingBadge from '../testingBadge.svelte'
import { getHandlerType, handleConfigChange, type Trigger } from '../utils'
import TriggerRetriesAndErrorHandler from '../TriggerRetriesAndErrorHandler.svelte'
import TriggerAdvancedBadges from '../TriggerAdvancedBadges.svelte'
import { fade } from 'svelte/transition'
import MultiSelect from '$lib/components/select/MultiSelect.svelte'
import { safeSelectItems } from '$lib/components/select/utils.svelte'
@@ -854,7 +855,10 @@
</Section>
<Section label="Advanced" collapsable>
<div class="flex flex-col gap-4">
{#snippet header()}
<TriggerAdvancedBadges {error_handler_path} {retry} />
{/snippet}
<div class="flex flex-col gap-6">
<div class="min-h-96">
<Tabs bind:selected={optionTabSelected}>
<Tab value="error_handler" label="Error Handler" />
@@ -874,6 +878,7 @@
</div>
</div>
</Section>
<div class="pb-8" />
</div>
{/if}
{/snippet}
frontend/src/lib/components/triggers/schedules/ScheduleEditorInner.svelte
+10
View File
@@ -1,5 +1,6 @@
<script lang="ts">
import { Alert, Badge, Button, ButtonType, Tab, Tabs } from '$lib/components/common'
import TriggerAdvancedBadges from '../TriggerAdvancedBadges.svelte'
import Drawer from '$lib/components/common/drawer/Drawer.svelte'
import DrawerContent from '$lib/components/common/drawer/DrawerContent.svelte'
import CronInput from '$lib/components/CronInput.svelte'
@@ -911,8 +912,17 @@
</Section>
<Section label="Advanced" collapsable>
{#snippet header()}
<TriggerAdvancedBadges error_handler_path={errorHandlerPath} {retry} extraBadges={[
{ name: 'Recovery Handler', active: !!recoveryHandlerPath },
{ name: 'Success Handler', active: !!successHandlerPath },
{ name: 'Dynamic Skip', active: !!dynamicSkipPath },
{ name: 'Custom Tag', active: !!tag }
]} />
{/snippet}
{@render errorHandler()}
</Section>
<div class="pb-8" />
</div>
{/if}
{/snippet}
frontend/src/lib/components/triggers/sqs/SqsTriggerEditorInner.svelte
+6 -1
View File
@@ -25,6 +25,7 @@
import Tabs from '$lib/components/common/tabs/Tabs.svelte'
import Tab from '$lib/components/common/tabs/Tab.svelte'
import TriggerRetriesAndErrorHandler from '../TriggerRetriesAndErrorHandler.svelte'
import TriggerAdvancedBadges from '../TriggerAdvancedBadges.svelte'
import TriggerSuspendedJobsAlert from '../TriggerSuspendedJobsAlert.svelte'
import TriggerSuspendedJobsModal from '../TriggerSuspendedJobsModal.svelte'
import { deepEqual } from 'fast-equals'
@@ -440,7 +441,10 @@
/>
<Section label="Advanced" collapsable>
<div class="flex flex-col gap-4">
{#snippet header()}
<TriggerAdvancedBadges {error_handler_path} {retry} />
{/snippet}
<div class="flex flex-col gap-6">
<div class="min-h-96">
<Tabs bind:selected={optionTabSelected}>
<Tab value="error_handler" label="Error Handler" />
@@ -460,6 +464,7 @@
</div>
</div>
</Section>
<div class="pb-8" />
</div>
{/if}
{/snippet}
frontend/src/lib/components/triggers/websocket/WebsocketTriggerEditorInner.svelte
+9 -3
View File
@@ -36,6 +36,7 @@
import Tabs from '$lib/components/common/tabs/Tabs.svelte'
import Tab from '$lib/components/common/tabs/Tab.svelte'
import TriggerRetriesAndErrorHandler from '../TriggerRetriesAndErrorHandler.svelte'
import TriggerAdvancedBadges from '../TriggerAdvancedBadges.svelte'
import { deepEqual } from 'fast-equals'
import TriggerSuspendedJobsAlert from '../TriggerSuspendedJobsAlert.svelte'
import TriggerSuspendedJobsModal from '../TriggerSuspendedJobsModal.svelte'
@@ -685,10 +686,14 @@
</div>
</Section>
<TriggerFilters bind:filters disabled={!can_write} />
<Section label="Advanced" collapsable>
<div class="flex flex-col gap-4">
{#snippet header()}
<TriggerAdvancedBadges {error_handler_path} {retry} extraBadges={[
{ name: 'Filters', active: filters.length > 0 }
]} />
{/snippet}
<div class="flex flex-col gap-6">
<TriggerFilters bind:filters disabled={!can_write} />
<div class="min-h-96">
<Tabs bind:selected={optionTabSelected}>
<Tab value="error_handler" label="Error Handler" />
@@ -708,6 +713,7 @@
</div>
</div>
</Section>
<div class="pb-8" />
</div>
{/if}
{/snippet}
frontend/src/lib/script_helpers.ts
+44 -20
View File
@@ -721,6 +721,7 @@ type TriggerEvent =
}
| {
kind: "http";
trigger_path: string;
body: any;
raw_string: string | null;
route: string;
@@ -732,20 +733,25 @@ type TriggerEvent =
}
| {
kind: "email";
trigger_path: string;
parsed_email: any;
raw_email: string;
email_extra_args?: Record<string, string>;
}
| { kind: "websocket"; msg: string; url: string }
| { kind: "websocket"; trigger_path: string; msg: string; url: string }
| {
kind: "kafka";
trigger_path: string;
payload: string;
brokers: string[];
topic: string;
partition: number;
offset: number;
group_id: string;
}
| {
kind: "nats";
trigger_path: string;
payload: string;
servers: string[];
subject: string;
@@ -756,6 +762,7 @@ type TriggerEvent =
}
| {
kind: "sqs";
trigger_path: string;
msg: string;
queue_url: string;
message_id?: string;
@@ -768,6 +775,7 @@ type TriggerEvent =
}
| {
kind: "mqtt";
trigger_path: string;
payload: string;
topic: string;
retain: boolean;
@@ -785,6 +793,7 @@ type TriggerEvent =
}
| {
kind: "gcp";
trigger_path: string;
payload: string;
message_id: string;
subscription: string;
@@ -797,6 +806,7 @@ type TriggerEvent =
}
| {
kind: "postgres";
trigger_path: string;
transaction_type: "insert" | "update" | "delete";
schema_name: string;
table_name: string;
@@ -869,6 +879,7 @@ class WebhookEvent(TypedDict):
class HttpEvent(TypedDict):
kind: Literal["http"]
trigger_path: str
body: dict
raw_string: Optional[str]
route: str
@@ -881,6 +892,7 @@ class HttpEvent(TypedDict):
class EmailEvent(TypedDict):
kind: Literal["email"]
trigger_path: str
parsed_email: dict
raw_email: str
email_extra_args: Optional[dict[str, str]]
@@ -888,20 +900,25 @@ class EmailEvent(TypedDict):
class WebsocketEvent(TypedDict):
kind: Literal["websocket"]
trigger_path: str
msg: str
url: str
class KafkaEvent(TypedDict):
kind: Literal["kafka"]
trigger_path: str
payload: str
brokers: list[str]
topic: str
partition: int
offset: int
group_id: str
class NatsEvent(TypedDict):
kind: Literal["nats"]
trigger_path: str
payload: str
servers: list[str]
subject: str
@@ -918,6 +935,7 @@ class MessageAttribute(TypedDict):
class SqsEvent(TypedDict):
kind: Literal["sqs"]
trigger_path: str
msg: str
queue_url: str
message_id: Optional[str]
@@ -938,6 +956,7 @@ class MqttV5Properties(TypedDict, total=False):
class MqttEvent(TypedDict):
kind: Literal["mqtt"]
trigger_path: str
payload: str
topic: str
retain: bool
@@ -948,6 +967,7 @@ class MqttEvent(TypedDict):
class GcpEvent(TypedDict):
kind: Literal["gcp"]
trigger_path: str
payload: str
message_id: str
subscription: str
@@ -961,6 +981,7 @@ class GcpEvent(TypedDict):
class PostgresEvent(TypedDict):
kind: Literal["postgres"]
trigger_path: str
transaction_type: Literal["insert", "update", "delete"]
schema_name: str
table_name: str
@@ -1025,41 +1046,44 @@ export const PHP_PREPROCESSOR_FLOW_INTRO = `<?php
export const PHP_PREPROCESSOR_MODULE_CODE = `function preprocessor(object $event) {
// $event can be one of the following types:
//
// All events (except webhook) include 'trigger_path' => '...' (the path of the trigger in Windmill)
//
// Webhook event:
// ['kind' => 'webhook', 'body' => [...], 'raw_string' => '...', 'query' => [...], 'headers' => [...]]
//
//
// HTTP event:
// ['kind' => 'http', 'body' => [...], 'raw_string' => '...', 'route' => '...', 'path' => '...',
// ['kind' => 'http', 'trigger_path' => '...', 'body' => [...], 'raw_string' => '...', 'route' => '...', 'path' => '...',
// 'method' => '...', 'params' => [...], 'query' => [...], 'headers' => [...]]
//
//
// Email event:
// ['kind' => 'email', 'parsed_email' => [...], 'raw_email' => '...', 'email_extra_args' => [...]]
//
// ['kind' => 'email', 'trigger_path' => '...', 'parsed_email' => [...], 'raw_email' => '...', 'email_extra_args' => [...]]
//
// WebSocket event:
// ['kind' => 'websocket', 'msg' => '...', 'url' => '...']
//
// ['kind' => 'websocket', 'trigger_path' => '...', 'msg' => '...', 'url' => '...']
//
// Kafka event:
// ['kind' => 'kafka', 'payload' => '...', 'brokers' => [...], 'topic' => '...', 'group_id' => '...']
//
// ['kind' => 'kafka', 'trigger_path' => '...', 'payload' => '...', 'brokers' => [...], 'topic' => '...',
// 'partition' => 0, 'offset' => 0, 'group_id' => '...']
//
// NATS event:
// ['kind' => 'nats', 'payload' => '...', 'servers' => [...], 'subject' => '...',
// ['kind' => 'nats', 'trigger_path' => '...', 'payload' => '...', 'servers' => [...], 'subject' => '...',
// 'headers' => [...], 'status' => 200, 'description' => '...', 'length' => 100]
//
//
// SQS event:
// ['kind' => 'sqs', 'msg' => '...', 'queue_url' => '...', 'message_id' => '...',
// ['kind' => 'sqs', 'trigger_path' => '...', 'msg' => '...', 'queue_url' => '...', 'message_id' => '...',
// 'receipt_handle' => '...', 'attributes' => [...], 'message_attributes' => [...]]
//
//
// MQTT event:
// ['kind' => 'mqtt', 'payload' => '...', 'topic' => '...', 'retain' => true, 'pkid' => 1,
// ['kind' => 'mqtt', 'trigger_path' => '...', 'payload' => '...', 'topic' => '...', 'retain' => true, 'pkid' => 1,
// 'qos' => 1, 'v5' => [...]]
//
//
// GCP event:
// ['kind' => 'gcp', 'payload' => '...', 'message_id' => '...', 'subscription' => '...',
// 'ordering_key' => '...', 'attributes' => [...], 'delivery_type' => 'push',
// ['kind' => 'gcp', 'trigger_path' => '...', 'payload' => '...', 'message_id' => '...', 'subscription' => '...',
// 'ordering_key' => '...', 'attributes' => [...], 'delivery_type' => 'push',
// 'headers' => [...], 'publish_time' => '...', 'ack_id' => '...']
//
//
// Postgres event:
// ['kind' => 'postgres', 'transaction_type' => 'insert', 'schema_name' => '...',
// ['kind' => 'postgres', 'trigger_path' => '...', 'transaction_type' => 'insert', 'schema_name' => '...',
// 'table_name' => '...', 'old_row' => [...], 'row' => [...]]
return [
python-client/wmill/wmill/client.py
+25
View File
@@ -2842,3 +2842,28 @@ def _run_workflow(func, checkpoint: dict, input_args: dict):
"""Synchronous wrapper that runs the workflow coroutine to completion
or until it suspends."""
return _asyncio.run(_run_workflow_async(func, checkpoint, input_args))
@init_global_client
def commit_kafka_offsets(
trigger_path: str,
topic: str,
partition: int,
offset: int,
) -> None:
"""Commit Kafka offsets for a trigger with auto_commit disabled.
Args:
trigger_path: Path to the Kafka trigger (from event['wm_trigger']['trigger_path'])
topic: Kafka topic name (from event['topic'])
partition: Partition number (from event['partition'])
offset: Message offset to commit (from event['offset'])
"""
_client.post(
f"/w/{_client.workspace}/kafka_triggers/commit_offsets/{trigger_path}",
json={
"topic": topic,
"partition": partition,
"offset": offset,
},
)
typescript-client/build.sh
+3 -1
View File
@@ -40,7 +40,7 @@ cp "${script_dirpath}/sqlUtils.ts" "${script_dirpath}/src/"
echo "" >> "${script_dirpath}/src/index.ts"
echo 'export type { DenoS3LightClientSettings } from "./s3Types";' >> "${script_dirpath}/src/index.ts"
echo "" >> "${script_dirpath}/src/index.ts"
echo 'export { type Base64, setClient, getVariable, setVariable, getResource, setResource, getResumeUrls, setState, setProgress, getProgress, getState, getIdToken, denoS3LightClientSettings, loadS3FileStream, loadS3File, writeS3File, signS3Objects, signS3Object, getPresignedS3PublicUrls, getPresignedS3PublicUrl, task, taskScript, taskFlow, workflow, step, sleep, parallel, waitForApproval, type TaskOptions, WorkflowCtx, _workflowCtx, setWorkflowCtx, StepSuspend, runScript, runScriptAsync, runScriptByPath, runScriptByHash, runScriptByPathAsync, runScriptByHashAsync, runFlow, runFlowAsync, waitJob, getRootJobId, setFlowUserState, getFlowUserState, usernameToEmail, requestInteractiveSlackApproval, type Sql, requestInteractiveTeamsApproval, appendToResultStream, streamResult, datatable, ducklake, type DatatableSqlTemplateFunction, type SqlTemplateFunction, type S3Object, type S3ObjectRecord, type S3ObjectURI } from "./client";' >> "${script_dirpath}/src/index.ts"
echo 'export { type Base64, setClient, getVariable, setVariable, getResource, setResource, getResumeUrls, setState, setProgress, getProgress, getState, getIdToken, denoS3LightClientSettings, loadS3FileStream, loadS3File, writeS3File, signS3Objects, signS3Object, getPresignedS3PublicUrls, getPresignedS3PublicUrl, task, taskScript, taskFlow, workflow, step, sleep, parallel, waitForApproval, type TaskOptions, WorkflowCtx, _workflowCtx, setWorkflowCtx, StepSuspend, runScript, runScriptAsync, runScriptByPath, runScriptByHash, runScriptByPathAsync, runScriptByHashAsync, runFlow, runFlowAsync, waitJob, getRootJobId, setFlowUserState, getFlowUserState, usernameToEmail, requestInteractiveSlackApproval, type Sql, requestInteractiveTeamsApproval, appendToResultStream, streamResult, datatable, ducklake, type DatatableSqlTemplateFunction, type SqlTemplateFunction, type S3Object, type S3ObjectRecord, type S3ObjectURI, commitKafkaOffsets } from "./client";' >> "${script_dirpath}/src/index.ts"
# Build default export by combining client utilities + services
# This preserves backward compatibility for `import wmill from "windmill-client"`
@@ -111,6 +111,7 @@ import {
base64ToUint8Array,
uint8ArrayToBase64,
parseS3Object,
commitKafkaOffsets,
} from "./client";
import {
@@ -195,6 +196,7 @@ const wmill = {
base64ToUint8Array,
uint8ArrayToBase64,
parseS3Object,
commitKafkaOffsets,
// Services
AdminService,
AuditService,
typescript-client/client.d.ts
Vendored
+13
View File
@@ -263,3 +263,16 @@ export declare function uint8ArrayToBase64(arrayBuffer: Uint8Array): string;
* @returns email address
*/
export declare function usernameToEmail(username: string): Promise<string>;
/**
* Commit Kafka offsets for a trigger with auto_commit disabled.
* @param triggerPath - Path to the Kafka trigger (from event.wm_trigger.trigger_path)
* @param topic - Kafka topic name (from event.topic)
* @param partition - Partition number (from event.partition)
* @param offset - Message offset to commit (from event.offset)
*/
export declare function commitKafkaOffsets(
triggerPath: string,
topic: string,
partition: number,
offset: number,
): Promise<void>;
typescript-client/client.ts
+22
View File
@@ -9,6 +9,7 @@ import {
MetricsService,
OidcService,
UserService,
KafkaTriggerService,
} from "./services.gen";
import { OpenAPI } from "./core/OpenAPI";
// import type { DenoS3LightClientSettings } from "./index";
@@ -1866,3 +1867,24 @@ export async function parallel<T, R>(
return results;
}
/**
* Commit Kafka offsets for a trigger with auto_commit disabled.
* @param triggerPath - Path to the Kafka trigger (from event.wm_trigger.trigger_path)
* @param topic - Kafka topic name (from event.topic)
* @param partition - Partition number (from event.partition)
* @param offset - Message offset to commit (from event.offset)
*/
export async function commitKafkaOffsets(
triggerPath: string,
topic: string,
partition: number,
offset: number,
): Promise<void> {
const workspace = getWorkspace();
await KafkaTriggerService.commitKafkaOffsets({
workspace,
path: triggerPath,
requestBody: { topic, partition, offset },
});
}