Apache Kafka ConnectorAlterOffsets
CertifiedAlter a Kafka Connect connector's offsets
Apache Kafka ConnectorAlterOffsets
Alter a Kafka Connect connector's offsets
Overwrites the connector's offsets with the given offsets value — typically obtained from ConnectorGetOffsets and edited, or built from scratch to skip/replay records.
The connector must be in the STOPPED state for this call to succeed; altering offsets via PATCH /connectors/{name}/offsets is a [KIP-875](https://cwiki.apache.org/confluence/display/KAFKA/KIP-875: +First-class+offsets+support+in+Kafka+Connect) concept only available on Kafka Connect clusters running Kafka 3.6+. This task does not pre-validate the connector's state client-side; it always sends the request and surfaces Connect's error body verbatim (HTTP 400) if the connector isn't STOPPED. Use ConnectorStop first and poll ConnectorGetStatus until it reports STOPPED. On older clusters that don't support this, delete and recreate the connector instead.
type: io.kestra.plugin.kafka.ConnectorAlterOffsetsExamples
Rewind a source connector's offsets
id: kafka_connector_alter_offsets
namespace: company.team
tasks:
- id: alter_offsets
type: io.kestra.plugin.kafka.ConnectorAlterOffsets
connectUrl: http://connect:8083
connectorName: orders_jdbc_source
offsets:
- partition:
table: orders
offset:
incrementing: 0
Installation
Our standard Docker images already include this plugin. If you use a -slim image or another setup, install it with the following command:
kestractl plugins get io.kestra.plugin:plugin-kafka --compatible-for $KESTRA_VERSION For kestractl, use your Kestra version with --compatible-for, or specify an exact plugin version.
For Kestra CLI, replace LATEST with a specific version or version range to pin the plugin version.
Properties
connectUrl *string
Kafka Connect REST API base URL
For example http://connect: 8083. Kafka Connect has no dedicated Java admin client — every operation goes through this REST API.
connectorName *string
Connector name
offsets *array
Offsets to apply
Same shape as ConnectorGetOffsets's offsets output: a list of {partition, offset} entries. The connector must be STOPPED — see the task description.
assets
Assets this task consumes as inputs or produces as outputs, for lineage tracking and the asset graph (Enterprise Edition). A flow declaring this property on a task is rejected in the open-source edition.
io.kestra.core.models.assets.AssetsDeclaration
IGNOREFAILWARNAsset failure behavior
Behavior applied to the task state when a declared asset fails to render, emit, or be persisted (e.g. a lock conflict): FAIL escalates it to FAILED, WARN (default) warns it if it would otherwise succeed, IGNORE leaves the state untouched.
Whether to auto-register assets referenced dynamically at runtime that are not statically declared in inputs or outputs.
The assets consumed as inputs.
io.kestra.core.models.assets.AssetIdentifier
1The assets produced as outputs.
io.kestra.plugin.ee.assets.Dataset
1150{}1150io.kestra.plugin.ee.assets.File
1150{}1150io.kestra.plugin.ee.assets.Table
1150{}1150io.kestra.plugin.ee.assets.VM
1150{}1150io.kestra.core.models.assets.External
1150{}1150io.kestra.core.models.assets.Custom
11501Custom asset type
{}1150headers object
Additional HTTP headers
Sent on every request to the Connect REST API. Useful when the worker sits behind a reverse proxy or expects a bearer token, e.g. Authorization: Bearer ....
password string
Basic auth password
username string
Basic auth username
Required together with password when the Connect REST API is protected with HTTP basic auth. Leave both unset to call an unauthenticated worker — no Authorization header is sent in that case.
Outputs
connectorName string
Connector name
message string
Confirmation message returned by the Connect API
