Apache Kafka Trigger

Apache Kafka Trigger
Certified
Start a Flow on scheduled Kafka pulls

Polls Kafka on a fixed interval (default PT1M, pollDuration PT5S) to batch records into one Execution. In groupType: CONSUMER (default), behavior is classic consumer groups with manual offset commits and committed-only reads. In groupType: SHARE, behavior is queue semantics with share groups and explicit acknowledgements. Records are stored in internal storage at {{ trigger.uri }}; defaults use STRING deserializers. Use header filters to drop mismatching records or switch to RealtimeTrigger for one-execution-per-record.

yaml
type: io.kestra.plugin.kafka.Trigger

Examples

yaml
id: kafka_trigger
namespace: company.team

tasks:
  - id: log
    type: io.kestra.plugin.core.log.Log
    message: "{{ trigger.value ?? '' }}"

triggers:
  - id: trigger
    type: io.kestra.plugin.kafka.Trigger
    topic: test_kestra
    properties:
      bootstrap.servers: localhost:9092
    serdeProperties:
      schema.registry.url: http://localhost:8085
    keyDeserializer: STRING
    valueDeserializer: AVRO
    interval: PT30S
    maxRecords: 5
    groupId: kafkaConsumerGroupId

Use a Kafka share group for queue semantics

yaml
id: kafka_trigger_share_group
namespace: company.team

tasks:
  - id: log
    type: io.kestra.plugin.core.log.Log
    message: "{{ trigger.messagesCount ?? 0 }}"

triggers:
  - id: trigger
    type: io.kestra.plugin.kafka.Trigger
    topic: orders
    properties:
      bootstrap.servers: localhost:9092
    groupId: orders-share-group
    groupType: SHARE
    acknowledgeType: ACCEPT

Properties

groupId *string

Kafka consumer group ID

Determines offset management; required when using topicPattern and mandatory for groupType: SHARE (share group).

properties *object

Kafka client properties

Must include bootstrap.servers; accepts any Kafka consumer or producer config. Provide base64-encoded content for ssl.keystore.location and ssl.truststore.location when using SSL.

acknowledgeType string

DefaultACCEPT

Possible Values

ACCEPTRELEASEREJECTRENEW

Acknowledgement action for SHARE group type

Used only when groupType is SHARE. ACCEPT (default) acknowledges processed records, RELEASE returns records to the queue, REJECT negatively acknowledges records, and RENEW extends the acquisition lock timeout for the current delivery attempt without changing record state. Ignored when groupType is CONSUMER.

allowConcurrent boolean

Defaultfalse

Specifies whether a trigger is allowed to start a new execution even if a previous run is still in progress.

conditions

List of conditions in order to limit the flow trigger.

Definitions

Allow events only between two datetimes.

Compares a rendered date (defaults to {{ trigger.date }}) against optional after and before bounds expressed as ISO-8601 datetimes with zone.

You must provide at least one bound; if both are set the date must fall strictly between them. Missing both bounds triggers an evaluation error.

Example

Trigger condition to execute the flow only after the specific date.

yaml
id: schedule_condition_datetimebetween
namespace: company.team

tasks:
  - id: log_message
    type: io.kestra.plugin.core.log.Log
    message: "This flow will execute once every 5 minutes after the date 2025-12-31T23:59:59Z"

triggers:
  - id: schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "*/5 * * * *"
    conditions:
      - type: io.kestra.plugin.core.condition.DateTimeBetween
        date: "{{ trigger.date }}"
        after: "2025-12-31T23:59:59Z"

Trigger condition to execute the flow between two specific dates.

yaml
id: schedule_condition_datetimebetween
namespace: company.team

tasks:
  - id: log_message
    type: io.kestra.plugin.core.log.Log
    message: "This flow will be executed once every 5 minutes between the before and after dates"

triggers:
  - id: schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "*/5 * * * *"
    conditions:
      - type: io.kestra.plugin.core.condition.DateTimeBetween
        date: "{{ trigger.date }}"
        after: "2025-01-01T00:00:00Z"
        before: "2025-12-31T23:59:59Z"

type*object

Possible Values

io.kestra.plugin.core.condition.DateTimeBetweenio.kestra.core.models.conditions.types.DateTimeBetweenConditionio.kestra.plugin.core.condition.DateTimeBetweenCondition

afterstring

The date to test must be after this one.

Must be a valid ISO 8601 datetime with the zone identifier (use 'Z' for the default zone identifier).

beforestring

The date to test must be before this one.

Must be a valid ISO 8601 datetime with the zone identifier (use 'Z' for the default zone identifier).

datestring

Default{{ trigger.date }}

The date to test.

Can be any variable or any valid ISO 8601 datetime. By default, it will use the trigger date.

Allow events on a specific weekday.

Renders a date (defaults to the trigger timestamp) and checks its day-of-week against the provided dayOfWeek.

Accepts ISO-8601 date/time strings; the comparison uses the rendered timezone.

Example

Trigger condition to execute the flow only on a specific day of the week.

yaml
id: schedule_condition_dayweek
namespace: company.team

tasks:
  - id: log_message
    type: io.kestra.plugin.core.log.Log
    message: "This flow will execute only on Mondays at 11:00 am."

triggers:
  - id: schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 11 * * *"
    conditions:
      - type: io.kestra.plugin.core.condition.DayWeek
        dayOfWeek: "MONDAY"

dayOfWeek*string

Possible Values

MONDAYTUESDAYWEDNESDAYTHURSDAYFRIDAYSATURDAYSUNDAY

The day of week.

type*object

Possible Values

io.kestra.plugin.core.condition.DayWeekio.kestra.core.models.conditions.types.DayWeekConditionio.kestra.plugin.core.condition.DayWeekCondition

datestring

Default{{ trigger.date }}

The date to test.

Can be any variable or any valid ISO 8601 datetime. By default, it will use the trigger date.

Allow events on an nth weekday within the month.

Renders a date (defaults to the trigger timestamp) and checks whether it matches the requested weekday and position in the month (FIRST, SECOND, THIRD, FOURTH, or LAST).

Useful for patterns like “first Monday” or “last Friday”. Dates must be valid ISO-8601 strings.

Example

Trigger condition to execute the flow only on the first Monday of the month.

yaml
id: schedule_condition_dayweekinmonth
namespace: company.team

tasks:
  - id: log_message
    type: io.kestra.plugin.core.log.Log
    message: "This flow will execute only on the first Monday of the month at 11:00 am."

triggers:
  - id: schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 11 * * *"
    conditions:
      - type: io.kestra.plugin.core.condition.DayWeekInMonth
        dayOfWeek: "MONDAY"
        dayInMonth: FIRST

dayInMonth*string

Possible Values

FIRSTLASTSECONDTHIRDFOURTH

Are you looking for the first or the last day in the month?

dayOfWeek*string

Possible Values

MONDAYTUESDAYWEDNESDAYTHURSDAYFRIDAYSATURDAYSUNDAY

The day of week.

type*object

Possible Values

io.kestra.plugin.core.condition.DayWeekInMonthio.kestra.plugin.core.condition.DayWeekInMonthConditionio.kestra.core.models.conditions.types.DayWeekInMonthCondition

datestring

Default{{ trigger.date }}

The date to test.

Can be any variable or any valid ISO 8601 datetime. By default, it will use the trigger date.

Match events from a specific flow.

Passes only when the triggering execution belongs to the given Namespace and Flow ID.

Pair with the Flow trigger to react to a single upstream flow; for broader namespace matching use ExecutionNamespace.

Example

Trigger condition to execute the flow based on execution of another flow.

yaml
id: flow_condition_executionflow
namespace: company.team

tasks:
  - id: hello
    type: io.kestra.plugin.core.log.Log
    message: "This flow will execute when flow `flow_a` of namespace `company.team` enters RUNNING state."

triggers:
  - id: flow_trigger
    type: io.kestra.plugin.core.trigger.Flow
    conditions:
      - type: io.kestra.plugin.core.condition.ExecutionFlow
        flowId: flow_a
        namespace: company.team
    states:
      - RUNNING

flowId*string

The Flow ID

namespace*string

The Flow Namespace

type*object

Possible Values

io.kestra.plugin.core.condition.ExecutionFlowio.kestra.plugin.core.condition.ExecutionFlowConditionio.kestra.core.models.conditions.types.ExecutionFlowCondition

Match executions by label set.

Evaluates true only if the execution carries all provided labels. Labels can be declared as a map or list; system-reserved labels are rejected.

Useful for filtering Flow triggers to specific environments, owners, or any custom tagging strategy.

Example

Trigger condition to execute the flow based on execution of another flow(s) with certain execution labels.

yaml
id: flow_condition_executionlabels
namespace: company.team

tasks:
  - id: hello
    type: io.kestra.plugin.core.log.Log
    message: "This flow will execute when flow with specified labels enters FAILED state."

triggers:
  - id: flow_trigger
    type: io.kestra.plugin.core.trigger.Flow
    conditions:
      - type: io.kestra.plugin.core.condition.ExecutionLabels
        labels:
          owner: john.doe
          env: prod
    states:
      - FAILED

labels*arrayobject

List of labels to match in the execution.

type*object

Possible Values

io.kestra.plugin.core.condition.ExecutionLabelsio.kestra.core.models.conditions.types.ExecutionLabelsConditionio.kestra.plugin.core.condition.ExecutionLabelsCondition

Match executions by namespace.

Compares the triggering execution’s namespace against a target string using EQUALS, PREFIX, or SUFFIX. The prefix boolean is a shorthand for comparison: PREFIX.

If no comparison is set, it defaults to strict equality unless prefix is true.

Example

Trigger condition to execute the flow based on execution of another flow(s) belonging to certain namespace.

yaml
id: flow_condition_executionnamespace
namespace: company.team

tasks:
  - id: hello
    type: io.kestra.plugin.core.log.Log
    message: "This flow will execute when any flow within `company.engineering` namespace enters RUNNING state."

triggers:
  - id: flow_trigger
    type: io.kestra.plugin.core.trigger.Flow
    conditions:
      - type: io.kestra.plugin.core.condition.ExecutionNamespace
        namespace: company.engineering
        comparison: PREFIX
    states:
      - RUNNING

namespace*string

String against which to match the execution namespace depending on the provided comparison.

type*object

Possible Values

io.kestra.plugin.core.condition.ExecutionNamespaceio.kestra.core.models.conditions.types.ExecutionNamespaceConditionio.kestra.plugin.core.condition.ExecutionNamespaceCondition

comparisonstring

Possible Values

EQUALSPREFIXSUFFIX

Comparison to use when checking if namespace matches. If not provided, it will use EQUALS by default.

prefixbooleanstring

Defaultfalse

Whether to look at the flow namespace by prefix. Shortcut for comparison: PREFIX.

Only used when comparison is not set

Condition based on the outputs of an upstream execution.

Renders the provided boolean expression against the upstream execution outputs exposed under trigger.outputs.

If the execution exposes no outputs the condition is false and the expression is skipped. A rendered result of blank, space-only, or literal false is treated as false.

Example

The upstream flow_a must explicitly define its outputs to be used in the ExecutionOutputs condition.

id: flow_a
namespace: company.team

inputs: 
  - id: user_value
    type: STRING
    defaults: hello

tasks: 
  - id: hello
    type: io.kestra.plugin.core.debug.Return
    format: "{{ inputs.user_value }}"

outputs: 
  - id: flow_a_output
    type: STRING
    value: "{{ outputs.hello.value }}"

The flow_condition_executionoutputs will run whenever flow_a finishes successfully and returns an output matching the value 'hello':

yaml
id: flow_condition_executionoutputs
namespace: company.team

tasks:
  - id: upstream_outputs
    type: io.kestra.plugin.core.log.Log
    message: hello from a downstream flow

triggers:
  - id: condition_on_flow_execution_outputs
    type: io.kestra.plugin.core.trigger.Flow
    states:
      - SUCCESS
    conditions:
      - type: io.kestra.plugin.core.condition.ExecutionOutputs
        expression: "{{ trigger.outputs.flow_a_output == 'hello' }}"

expression*booleanstring

type*object

Possible Values

io.kestra.plugin.core.condition.ExecutionOutputsio.kestra.core.models.conditions.types.ExecutionOutputsConditionio.kestra.plugin.core.condition.ExecutionOutputsCondition

Match executions by status.

Passes when the triggering execution’s current state is included in in and not present in notIn.

If a list is empty it is ignored, so provide at least one of them to avoid unintentionally matching everything.

Example

Trigger condition to execute the flow based on execution status of another flow(s).

yaml
id: flow_condition_executionstatus
namespace: company.team

tasks:
  - id: hello
    type: io.kestra.plugin.core.log.Log
    message: "This flow will execute when any flow enters FAILED or KILLED state."

triggers:
  - id: flow_trigger
    type: io.kestra.plugin.core.trigger.Flow
    conditions:
      - type: io.kestra.plugin.core.condition.ExecutionStatus
        in:
          - FAILED
          - KILLED

type*object

Possible Values

io.kestra.plugin.core.condition.ExecutionStatusio.kestra.core.models.conditions.types.ExecutionStatusConditionio.kestra.plugin.core.condition.ExecutionStatusCondition

inarray

SubTypestring

Possible Values

CREATEDSUBMITTEDRUNNINGPAUSEDRESTARTEDKILLINGSUCCESSWARNINGFAILEDKILLEDCANCELLEDQUEUEDRETRYINGRETRIEDSKIPPEDBREAKPOINTRESUBMITTED

List of states that are authorized.

notInarray

SubTypestring

Possible Values

CREATEDSUBMITTEDRUNNINGPAUSEDRESTARTEDKILLINGSUCCESSWARNINGFAILEDKILLEDCANCELLEDQUEUEDRETRYINGRETRIEDSKIPPEDBREAKPOINTRESUBMITTED

List of states that aren't authorized.

Condition based on variable expression.

Renders a templated expression and treats the result as truthy/falsey to decide whether the condition passes.

Blank strings or the literal false (case-sensitive) evaluate to false; everything else is true. Expressions can reference any flow variables available at evaluation time, so make sure they resolve without errors.

Example

Trigger condition to execute the flow when the expression evaluates to true.

yaml
id: myflow
namespace: company.team

tasks:
  - id: hello
    type: io.kestra.plugin.core.log.Log
    message: Average value has gone below 10

triggers:
  - id: expression_trigger
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "*/1 * * * *"
    conditions:
      - type: io.kestra.plugin.core.condition.Expression
        expression: "{{ kv('average_value') < 10 }}"

expression*string

type*object

Possible Values

io.kestra.plugin.core.condition.Expressionio.kestra.plugin.core.condition.ExpressionConditionio.kestra.core.models.conditions.types.VariableCondition

Match executions where a task was retried.

Passes if any task run has more than one attempt and its state satisfies in / notIn filters.

Provide in states to avoid matching nothing (an empty in list fails all attempts). Use with Flow triggers to react when retries occur.

Example

Trigger condition when any flow task on retry enters the state specified in the in states under the HasRetryAttempt condition.

yaml
id: flow_condition_hasretryattempt
namespace: company.team

tasks:
  - id: log_message
    type: io.kestra.plugin.core.log.Log
    message: "This flow will execute when any flow task on retry enters a specific state(s)."

triggers:
  - id: flow_condition
    type: io.kestra.plugin.core.trigger.Flow
    conditions:
      - type: io.kestra.plugin.core.condition.HasRetryAttempt
        in:
          - FAILED

type*object

Possible Values

io.kestra.plugin.core.condition.HasRetryAttemptio.kestra.plugin.core.condition.HasRetryAttemptConditionio.kestra.core.models.conditions.types.HasRetryAttemptCondition

inarray

SubTypestring

Possible Values

CREATEDSUBMITTEDRUNNINGPAUSEDRESTARTEDKILLINGSUCCESSWARNINGFAILEDKILLEDCANCELLEDQUEUEDRETRYINGRETRIEDSKIPPEDBREAKPOINTRESUBMITTED

List of states that are authorized.

notInarray

SubTypestring

Possible Values

CREATEDSUBMITTEDRUNNINGPAUSEDRESTARTEDKILLINGSUCCESSWARNINGFAILEDKILLEDCANCELLEDQUEUEDRETRYINGRETRIEDSKIPPEDBREAKPOINTRESUBMITTED

List of states that aren't authorized.

Invert one or more conditions.

Fails the evaluation if any nested condition is true; otherwise passes.

Combine with other schedule conditions to express “not these dates/states” patterns.

Example

Trigger condition to execute the flow when the condition is not satisfied.

yaml
id: schedule_condition_not
namespace: company.team

tasks:
  - id: log_message
    type: io.kestra.plugin.core.log.Log
    message: "This flow will execute on all days except Sunday at 11am."

triggers:
  - id: schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 11 * * *"
    conditions:
      - type: io.kestra.plugin.core.condition.Not
        conditions:
          - type: io.kestra.plugin.core.condition.DayWeek
            dayOfWeek: "SUNDAY"

conditions*

Min items1

The list of conditions to exclude.

If any condition is true, it will prevent the event's execution.

type*object

Possible Values

io.kestra.plugin.core.condition.Notio.kestra.core.models.conditions.types.NotConditionio.kestra.plugin.core.condition.NotCondition

Pass when any condition is true.

Evaluates nested conditions and succeeds if at least one passes.

Useful for schedule scenarios like “weekends or holidays” without duplicating triggers.

Example

Trigger condition to execute the flow when any of the conditions are satisfied.

yaml
id: schedule_condition_or
namespace: company.team

tasks:
  - id: log_message
    type: io.kestra.plugin.core.log.Log
    message: "This flow will execute on Sundays and Mondays at 11am."

triggers:
  - id: schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 11 * * *"
    conditions:
      - type: io.kestra.plugin.core.condition.Or
        conditions:
          - type: io.kestra.plugin.core.condition.DayWeek
            dayOfWeek: "MONDAY"
          - type: io.kestra.plugin.core.condition.DayWeek
            dayOfWeek: "SUNDAY"

conditions*

Min items1

The list of conditions to validate.

If any condition is true, it will allow the event's execution.

type*object

Possible Values

io.kestra.plugin.core.condition.Orio.kestra.core.models.conditions.types.OrConditionio.kestra.plugin.core.condition.OrCondition

Allow events on public holidays.

Uses the Jollyday calendar to check whether a rendered date (defaults to {{ trigger.date }}) is a public holiday for the given country/sub-division.

If country is omitted, the default locale is used. Provide ISO 3166-1 (country) and optionally 3166-2 (region) codes.

Example

Trigger condition to execute the flow only on public holidays.

yaml
id: schedule_condition_public-holiday
namespace: company.team

tasks:
  - id: log_message
    type: io.kestra.plugin.core.log.Log
    message: "This flow will execute only on the public holidays of France at 11:00 am."

triggers:
  - id: schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 11 * * *"
    conditions:
      - type: io.kestra.plugin.core.condition.PublicHoliday
        country: FR

Trigger condition to execute the flow only on work days in France.

yaml
id: schedule-condition-work-days
namespace: company.team

tasks:
  - id: log_message
    type: io.kestra.plugin.core.log.Log
    message: "This flow will execute only on the work days of France at 11:00 am."

triggers:
  - id: schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 11 * * *"
    conditions:
      - type: io.kestra.plugin.core.condition.Not
        conditions:
          - type: io.kestra.plugin.core.condition.PublicHoliday
            country: FR
          - type: io.kestra.plugin.core.condition.Weekend

type*object

Possible Values

io.kestra.plugin.core.condition.PublicHolidayio.kestra.plugin.core.condition.PublicHolidayConditionio.kestra.core.models.conditions.types.PublicHolidayCondition

countrystring

ISO 3166-1 alpha-2 country code. If not set, it uses the country code from the default locale.

It uses the Jollyday library for public holiday calendar that supports more than 70 countries.

datestring

Default{{ trigger.date}}

The date to test.

Can be any variable or any valid ISO 8601 datetime. By default, it will use the trigger date.

subDivisionstring

ISO 3166-2 country subdivision (e.g., provinces and states) code.

It uses the Jollyday library for public holiday calendar that supports more than 70 countries.

Allow events between two times of day.

Compares the rendered date/time (defaults to the trigger timestamp) against optional after and before offsets (ISO-8601 times).

Supports ranges that cross midnight (e.g., 22: 00→02: 00). At least one bound is required; missing both raises an evaluation error.

Example

Trigger condition to execute the flow between two specific times.

yaml
id: schedule_condition_timebetween
namespace: company.team

tasks:
  - id: log_message
    type: io.kestra.plugin.core.log.Log
    message: "This flow will execute every 5 minutes between 4pm and 8pm."

triggers:
  - id: schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "*/5 * * * *"
    conditions:
      - type: io.kestra.plugin.core.condition.TimeBetween
        after: "16:00:00+02:00"
        before: "20:00:00+02:00"

type*object

Possible Values

io.kestra.plugin.core.condition.TimeBetweenio.kestra.plugin.core.condition.TimeBetweenConditionio.kestra.core.models.conditions.types.TimeBetweenCondition

afterstring

The time to test must be after this one.

Must be a valid ISO 8601 time with offset.

beforestring

The time to test must be before this one.

Must be a valid ISO 8601 time with offset.

datestring

Default{{ trigger.date }}

The time to test.

Can be any variable or any valid ISO 8601 time. By default, it will use the trigger date.

Allow events on weekends.

Renders a date (defaults to the trigger timestamp) and passes only if it falls on Saturday or Sunday.

Accepts ISO-8601 date/time strings; uses the rendered timezone to determine the day.

Example

Trigger the flow only on weekend, i.e. on Saturdays and Sundays.

yaml
id: schedule_condition_weekend
namespace: company.team

tasks:
  - id: log_message
    type: io.kestra.plugin.core.log.Log
    message: "This flow will execute only on weekends at 11:00 am."

triggers:
  - id: schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 11 * * *"
    conditions:
      - type: io.kestra.plugin.core.condition.Weekend

type*object

Possible Values

io.kestra.plugin.core.condition.Weekendio.kestra.plugin.core.condition.WeekendConditionio.kestra.core.models.conditions.types.WeekendCondition

datestring

Default{{ trigger.date }}

The date to test.

Can be any variable or any valid ISO 8601 datetime. By default, it will use the trigger date.

groupType string

DefaultCONSUMER

Possible Values

CONSUMERSHARE

Group protocol to consume with

CONSUMER (default) polls with classic consumer-group behavior. SHARE polls with Kafka share-group queue semantics and explicit acknowledgements. In SHARE mode, use topic with groupId; topicPattern, partitions, and since are not supported.

headerFilters object

Filter messages by Kafka headers

Consume records only when all header key/value pairs match exactly (last header wins, UTF-8 comparison)

interval string

DefaultPT1M

Formatduration

Interval between polling.

The interval between 2 different polls of schedule, this can avoid to overload the remote system with too many calls. For most of the triggers that depend on external systems, a minimal interval must be at least PT30S. See ISO_8601 Durations for more information of available interval values.

keyDeserializer string

DefaultSTRING

Possible Values

STRINGINTEGERFLOATDOUBLELONGSHORTBYTE_ARRAYBYTE_BUFFERBYTESUUIDVOIDAVROJSONPROTOBUF

Deserializer used for the key

Default STRING. Options: STRING, INTEGER, FLOAT, DOUBLE, LONG, SHORT, BYTE_ARRAY, BYTE_BUFFER, BYTES, UUID, VOID, AVRO, JSON.

maxDuration string

Maximum duration to wait before stopping

Soft limit checked on each poll.

maxRecords integerstring

Maximum records to consume before stopping

Soft limit checked on each poll.

onSerdeError

Behavior on serde error

Applies when valueDeserializer is JSON: SKIPPED (default), STORE to internal storage, or DLQ to the configured topic.

Definitions

io.kestra.plugin.kafka.KafkaConsumerInterface-OnSerdeError

topicstring

Topic used when type is DLQ

typestring

DefaultSKIPPED

Possible Values

SKIPPEDDLQSTORE

Action to take on serde error

partitions array

SubTypeinteger

Specific partitions to consume

Manually assign partitions; bypasses consumer group rebalancing.

pluginDefaultsRef string

Reference (ref) of the pluginDefaults to apply to this trigger.

pollDuration string

DefaultPT5S

How often to poll for records

Maximum wait per poll when no records are available.

serdeProperties object

Default{}

Serializer or deserializer properties

Passed to serdes; avro.use.logical.type.converters is forced to true by default.

since string

Timestamp to start consuming from

ISO-8601 instant used when no consumer group offsets exist; ignored when a consumer group controls offsets.

stopAfter array

SubTypestring

Possible Values

CREATEDSUBMITTEDRUNNINGPAUSEDRESTARTEDKILLINGSUCCESSWARNINGFAILEDKILLEDCANCELLEDQUEUEDRETRYINGRETRIEDSKIPPEDBREAKPOINTRESUBMITTED

List of execution states after which a trigger should be stopped (a.k.a. disabled).

topic object

Kafka topic(s) to consume from

String or list of strings; mutually exclusive with topicPattern.

topicPattern string

Regex pattern of topics to consume from

Subscribes to topics matching the pattern and receives dynamic partition assignments; mutually exclusive with topic.

valueDeserializer string

DefaultSTRING

Possible Values

STRINGINTEGERFLOATDOUBLELONGSHORTBYTE_ARRAYBYTE_BUFFERBYTESUUIDVOIDAVROJSONPROTOBUF

Deserializer used for the value

Default STRING. Options: STRING, INTEGER, FLOAT, DOUBLE, LONG, SHORT, BYTE_ARRAY, BYTE_BUFFER, BYTES, UUID, VOID, AVRO, JSON.