KAFKA-15976: Allow specifying initial offsets while creating connectors - #23268
Open
psiby wants to merge 4 commits into
Open
KAFKA-15976: Allow specifying initial offsets while creating connectors#23268psiby wants to merge 4 commits into
psiby wants to merge 4 commits into
Conversation
…creation REST entities
Adds the two wire-format fields KIP-995 introduces, without wiring them
into any behaviour yet:
- CreateConnectorRequest gains an optional `initial_offsets` component
plus initialOffsetsMap(), which reuses ConnectorOffsets.toMap() so the
payload format matches PATCH /connectors/{connector}/offsets exactly.
Absent stays distinguishable from empty: absent yields null, an empty
list yields an empty map.
- ConnectorInfo gains an optional `offsets_status` component annotated
@JsonInclude(NON_NULL), so the responses of every other endpoint that
returns this type are byte-for-byte unchanged.
Both records keep a delegating constructor that fills in null for the new
component, so none of the existing construction sites change. Each also
declares its canonical constructor explicitly with @JsonCreator, following
the ConnectorTaskId precedent, rather than relying on Jackson's implicit
record-creator detection now that a second constructor exists; the
round-trip tests cover this rather than assuming it.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Worker.modifyConnectorOffsets has had exactly two modes, selected by whether the offsets argument is null: reset (wipe everything) and alter (write these, leave the rest alone). Creating a connector at a chosen position needs both at once, since offsets are keyed by connector name and outlive the connector, so a recreated name can otherwise inherit stale positions. Adds a third mode behind a `replaceAllOffsets` flag: tombstone every offset the connector currently has, then overlay the requested ones on top, so the connector ends up with exactly the requested offsets rather than a merge. Both write paths already accept a mixed map where a null value means delete and a non-null value means write, so this is a single call and a single flush rather than a reset followed by an alter -- which would double the slow steps, invoke alterOffsets() twice, and introduce a "wiped but not written" intermediate state. The sink path parses the requested offsets before listing the consumer group, so a malformed request is rejected before any consumer group offsets are listed, altered, or deleted. Reset and replace share one listing block, since they differed only in log wording. completeModifyOffsetsCallback takes the verb as a String rather than a boolean isReset, so the "set" wording KIP-995 specifies is available. The pre-existing four- and six-argument entry points are kept as delegates, so every existing call site and test is untouched. Tests cover replacing offsets for source and sink (with and without pre-existing offsets), the framework-managed response wording, and that a malformed offset is rejected before anything is wiped. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Wires initial_offsets through both herders, following the sequence KIP-995
specifies: validate the config, validate the offsets, wipe all existing
offsets for the name, write the new offsets, and write the connector config
last. Config-last is the correctness guarantee -- writing the config record
is what makes a connector startable, so its offsets are always correct
before it can run. This ordering was requested on the dev@ discussion to
avoid rebalancing inside the REST request.
Herder gains a putConnectorConfig overload taking the offsets; the existing
overload is kept and delegates with null, mirroring how KIP-980 added its
target-state overload. Both herders branch on initialOffsets == null so the
no-offsets path stays exactly as it was.
AbstractHerder hosts the shared pieces so the two herders cannot diverge:
validateInitialOffsets (pure, so an invalid request is rejected before
anything destructive happens), setInitialConnectorOffsets (documents why
the usual exists/STOPPED preconditions are skipped -- the connector is
being created, so it is not expected to exist), and the cleanup that wipes
the offsets if the config write then fails, per the KIP.
Null offset values are rejected with a 400. On create the wipe has already
removed every partition, so a null has nothing to delete, and issuing the
delete anyway targets the consumer group the wipe just removed, which the
alter path does not tolerate. This is deliberately stricter than
PATCH /connectors/{connector}/offsets and is called out as such.
DistributedHerder hands back to the tick thread for the config write rather
than blocking it through the offsets write, re-checking its preconditions
afterwards; this mirrors the existing modifyConnectorOffsets plus zombie
fencing flow. StandaloneHerder hops back onto its request executor for the
same reason.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
End-to-end coverage for both offset mechanisms, which are entirely separate
code paths:
- testCreateSourceConnectorWithInitialOffsets creates a source connector
positioned at offset 5 and asserts it produces only the last 5 of its
10 messages, and that the creation response carries offsets_status,
proving the initial offsets were written to the offsets topic before the
connector's tasks started.
- testCreateSinkConnectorWithInitialOffsets produces 10 records, creates a
sink connector positioned at offset 5, and fails the task if it ever
sees a record below offset 5, proving the initial offsets were committed
to the consumer group before the connector's tasks started.
Both mirror the create-then-modify-offsets tests added for KIP-980,
collapsing the create-stopped / alter-offsets / resume sequence into a
single create request that carries initial_offsets.
The Connect REST API reference gains a description of initial_offsets on
POST /connectors: its format (the same as the PATCH offsets body), that
existing offsets for the name are wiped rather than merged, and that an
invalid offsets payload fails the request without creating the connector.
The OpenAPI reference is generated from the REST entities at build time, so
no checked-in spec needs updating.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
|
A label of 'needs-attention' was automatically added to this PR in order to raise the |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Today, creating a connector that starts from a specific position takes three
REST calls — create it stopped,
PATCHits offsets, then resume it — and anyone of them can fail partway. This also closes a gap in Connect's state model:
offsets are keyed by connector name and outlive the connector, so deleting a
connector and recreating it with the same name silently resumes from the old
position. This KIP makes the safe path a single call.
Changes in this PR
initial_offsetsfield to thePOST /connectorsrequestbody, in the same format as the
PATCH /connectors/{connector}/offsetsbody(KIP-875). When supplied, all existing offsets for the connector name are
wiped and replaced with the requested offsets before the connector is created.
offsets_statusfield to the creation response(
@JsonInclude(NON_NULL)), populated only wheninitial_offsetswas suppliedso every other endpoint returning
ConnectorInfois unchanged.standalone CLI's JSON connector configuration files.
Sequence
Following the sequence agreed on the discussion thread:
Writing the config record is what makes a connector startable, so the offsets
are always correct before it can run. Every failure before the config write
leaves only orphaned offsets — the ordinary state of any connector that has been
deleted — and if the config write itself fails, the just-written offsets are
wiped back out, per the KIP.
Notable decisions
and the offsets are written first off it; the connector is never rebalanced
in the middle of the request.
ones are written, so the connector starts from exactly the requested offsets.
removed every partition, so a null offset has nothing to delete; issuing the
delete anyway targets the consumer group the wipe just removed, which the alter
path does not tolerate. This is deliberately stricter than
PATCH /{connector}/offsets, which allows nulls.framework-managed variant reads "The Connect framework-managed offsets ...".
Workercall, a connector'salterOffsets()hook may receive a map mixing tombstones (reset) and values(alter). This is permitted by the
alterOffsets()contract, which documentsthat a single map may contain
nullvalues for reset alongside non-nullvalues for alter;
POST /connectorswithinitial_offsetsis simply thefirst path to produce this at scale.
Testing
WorkerTest: replacing offsets for source and sink connectors (with andwithout pre-existing offsets), the exact framework-managed response wording,
and that a malformed offset is rejected before anything is wiped.
DistributedHerderTest/StandaloneHerderTest: the offsets-before-configordering (via
InOrder), null-value rejection, and — distributed — thecleanup that wipes the offsets when the config write fails.
ConnectWorkerIntegrationTest: end-to-end for both offset mechanisms — asource connector created at offset 5 produces only its last 5 messages (and
the response carries
offsets_status), and a sink connector created at offset5 never sees a record below offset 5.
Co-authored-by: Claude Opus 4.8 noreply@anthropic.com