feat: add kafka trigger offset reset and auto.offset.reset config (#8283)

* fix: parallel branchall hang on bad stop_after_all_iters_if + results.x.length null

Two fixes:

1. When a parallel branchall/forloop has a `stop_after_all_iters_if` expression
   that fails (e.g. bad JS syntax), the error was propagated with `?`, causing
   the transaction to roll back the parallel index increment. Since all parallel
   jobs were already completed, nothing could ever increment the index again and
   the flow hung forever. Now the error is caught and converted to a stop-early
   failure so the transaction commits and the flow fails gracefully.

2. Expressions like `results.a.length` in step input transforms resolved to null
   because the `handle_full_regex` fast path intercepted them and used
   PostgreSQL's `#>` JSON path operator, which can't resolve JS runtime
   properties like `.length` on arrays. Now the fast path skips expressions
   ending with JS-only properties (like `length`), falling through to full
   QuickJS evaluation where they work correctly.

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

* feat: add kafka trigger offset reset and auto.offset.reset configuration

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

* chore: update ee-repo-ref for kafka offset reset

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

* chore: update ee-repo-ref for subscribe+seek approach

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

* chore: update ee-repo-ref for kafka offset reset fix

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

* fix: use ConfirmationModal instead of browser confirm() for kafka offset reset

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

* chore: update ee-repo-ref for offset commit fix

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

* sqlx update

* Update ee-repo-ref.txt

* update ee ref

* update sqlx

* update ee ref

* chore: update ee-repo-ref to a70d7db187aa78a7fbfd3bfaf92372160cff320a

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

Previous ee-repo-ref: 238c2c0a91f353126f349a5153173a6d16c9d652

New ee-repo-ref: a70d7db187aa78a7fbfd3bfaf92372160cff320a

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>

backend/ee-repo-ref.txt

@@ -1 +1 @@
-09dfb247f6f59c61b7f2431932c4557fb26c22d8
+a70d7db187aa78a7fbfd3bfaf92372160cff320a

backend/migrations/20260309000000_kafka_offset_reset.down.sql

@@ -0,0 +1,2 @@
+ALTER TABLE kafka_trigger DROP COLUMN auto_offset_reset;
+ALTER TABLE kafka_trigger DROP COLUMN reset_offset;

backend/migrations/20260309000000_kafka_offset_reset.up.sql

@@ -0,0 +1,2 @@
+ALTER TABLE kafka_trigger ADD COLUMN auto_offset_reset VARCHAR(10) NOT NULL DEFAULT 'latest';
+ALTER TABLE kafka_trigger ADD COLUMN reset_offset BOOLEAN NOT NULL DEFAULT FALSE;

backend/windmill-api/openapi.yaml

@@ -12004,6 +12004,19 @@ paths:
               schema:
                 type: string
 
+  /w/{workspace}/kafka_triggers/reset_offsets/{path}:
+    post:
+      summary: reset kafka trigger offsets to earliest
+      operationId: resetKafkaOffsets
+      tags:
+        - kafka_trigger
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+        - $ref: "#/components/parameters/Path"
+      responses:
+        "200":
+          description: kafka trigger offsets reset successfully
+
   /w/{workspace}/nats_triggers/create:
     post:
       summary: create nats trigger
@@ -22061,7 +22074,7 @@ components:
           description: Path to the Kafka resource containing connection configuration
         group_id:
           type: string
-          description: Kafka consumer group ID for this trigger 
+          description: Kafka consumer group ID for this trigger
         topics:
           type: array
           items:
@@ -22078,6 +22091,13 @@ components:
             required:
               - key
               - value
+        auto_offset_reset:
+          type: string
+          enum:
+            - latest
+            - 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."
         server_id:
           type: string
           description: ID of the server currently handling this trigger (internal)
@@ -22138,6 +22158,13 @@ components:
             required:
               - key
               - value
+        auto_offset_reset:
+          type: string
+          enum:
+            - latest
+            - earliest
+          default: latest
+          description: "Initial offset behavior when consumer group has no committed offset."
         mode:
           $ref: "#/components/schemas/TriggerMode"
         error_handler_path:
@@ -22190,6 +22217,13 @@ components:
             required:
               - key
               - value
+        auto_offset_reset:
+          type: string
+          enum:
+            - latest
+            - earliest
+          default: latest
+          description: "Initial offset behavior when consumer group has no committed offset."
         path:
           type: string
           description: The unique path identifier for this trigger

frontend/src/lib/components/triggers/kafka/KafkaTriggerEditorInner.svelte

@@ -1,15 +1,21 @@
 <script lang="ts">
 	import { Alert, Button } from '$lib/components/common'
+	import ConfirmationModal from '$lib/components/common/confirmationModal/ConfirmationModal.svelte'
 	import Drawer from '$lib/components/common/drawer/Drawer.svelte'
 	import DrawerContent from '$lib/components/common/drawer/DrawerContent.svelte'
 	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'
-	import { Loader2 } from 'lucide-svelte'
+	import { Loader2, RotateCcw } from 'lucide-svelte'
 	import Label from '$lib/components/Label.svelte'
 	import KafkaTriggersConfigSection from './KafkaTriggersConfigSection.svelte'
 	import { untrack, type Snippet } from 'svelte'
@@ -80,7 +86,9 @@
 	let kafkaCfgValid = $state(false)
 	let kafkaResourcePath = $state('')
 	let kafkaCfg: Record<string, any> = $state({})
+	let autoOffsetReset = $state('latest')
 	let deploymentLoading = $state(false)
+	let resetLoading = $state(false)
 	let optionTabSelected: 'error_handler' | 'retries' = $state('error_handler')
 	let errorHandlerSelected: ErrorHandler = $state('slack')
 	let error_handler_path: string | undefined = $state()
@@ -90,6 +98,7 @@
 
 	let suspendedJobsModal = $state<TriggerSuspendedJobsModal | null>(null)
 	let originalConfig = $state<Record<string, any> | undefined>(undefined)
+	let resetConfirmOpen = $state(false)
 
 	const isValid = $derived(
 		!!kafkaResourcePath &&
@@ -166,6 +175,7 @@
 				group_id: nDefaultValues?.group_id ?? '',
 				topics: nDefaultValues?.topics ?? ['']
 			}
+			autoOffsetReset = nDefaultValues?.auto_offset_reset ?? 'latest'
 			initialScriptPath = ''
 			fixedScriptPath = fixedScriptPath_ ?? ''
 			script_path = fixedScriptPath
@@ -196,6 +206,7 @@
 			group_id: cfg?.group_id,
 			topics: cfg?.topics
 		}
+		autoOffsetReset = cfg?.auto_offset_reset ?? 'latest'
 		mode = cfg?.mode ?? 'enabled'
 		extra_perms = cfg?.extra_perms
 		can_write = canWrite(path, cfg?.extra_perms, $userStore)
@@ -228,6 +239,7 @@
 			group_id: kafkaCfg.group_id,
 			topics: kafkaCfg.topics,
 			filters,
+			auto_offset_reset: autoOffsetReset,
 			mode,
 			extra_perms: extra_perms,
 			error_handler_path,
@@ -267,6 +279,24 @@
 		}
 	}
 
+	async function resetOffsets() {
+		resetLoading = true
+		try {
+			await KafkaTriggerService.resetKafkaOffsets({
+				workspace: $workspaceStore!,
+				path: initialPath
+			})
+			sendUserToast(
+				'Offset reset triggered. The consumer will restart and re-read from the beginning.'
+			)
+		} catch (error) {
+			sendUserToast(error.body || error.message, true)
+		} finally {
+			resetLoading = false
+			resetConfirmOpen = false
+		}
+	}
+
 	async function handleToggleMode(newMode: TriggerMode) {
 		mode = newMode
 		if (!trigger?.draftConfig) {
@@ -296,6 +326,18 @@
 	})
 </script>
 
+<ConfirmationModal
+	title="Reset consumer offset"
+	confirmationText="Reset"
+	open={resetConfirmOpen}
+	loading={resetLoading}
+	onConfirmed={resetOffsets}
+	onCanceled={() => (resetConfirmOpen = false)}
+>
+	This will re-process all messages from the beginning of the topic. The consumer will restart
+	automatically.
+</ConfirmationModal>
+
 {#if mode === 'suspended'}
 	<TriggerSuspendedJobsModal
 		bind:this={suspendedJobsModal}
@@ -439,11 +481,32 @@
 				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>

frontend/src/lib/components/triggers/kafka/KafkaTriggersConfigSection.svelte

@@ -3,6 +3,8 @@
 	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'
@@ -13,6 +15,7 @@
 		kafkaCfgValid?: boolean
 		kafkaResourcePath?: string
 		kafkaCfg?: Record<string, any>
+		autoOffsetReset?: string
 		can_write?: boolean
 		showTestingBadge?: boolean
 	}
@@ -22,10 +25,16 @@
 		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',
@@ -99,6 +108,21 @@
 					/>
 				</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

@@ -24,6 +24,7 @@ export async function saveKafkaTriggerFromCfg(
 		group_id: cfg.group_id,
 		topics: cfg.topics,
 		filters: cfg.filters ?? [],
+		auto_offset_reset: cfg.auto_offset_reset ?? 'latest',
 		...errorHandlerAndRetries
 	}
 	try {