psiby opened a new pull request, #23268: URL: https://github.com/apache/kafka/pull/23268
- https://cwiki.apache.org/confluence/display/KAFKA/KIP-995%3A+Allow+users+to+specify+initial+offsets+while+creating+connectors - https://issues.apache.org/jira/browse/KAFKA-15976 Today, creating a connector that starts from a specific position takes three REST calls — create it stopped, `PATCH` its offsets, then resume it — and any one 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 - Add an optional `initial_offsets` field to the `POST /connectors` request body, in the same format as the `PATCH /connectors/{connector}/offsets` body (KIP-875). When supplied, all existing offsets for the connector name are wiped and replaced with the requested offsets before the connector is created. - Add an optional `offsets_status` field to the creation response (`@JsonInclude(NON_NULL)`), populated only when `initial_offsets` was supplied so every other endpoint returning `ConnectorInfo` is unchanged. - Support the new field in both distributed and standalone modes, and in the standalone CLI's JSON connector configuration files. - Unit and integration tests for both modes, and documentation. ### Sequence Following the sequence agreed on the discussion thread: 1. Validate the connector config and the initial offsets. 2. Write the offsets (wiping any existing offsets for the name first). 3. Write the connector config **last**. 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 - **Offsets before config.** The config write is done last, on the tick thread, and the offsets are written first off it; the connector is never rebalanced in the middle of the request. - **Replace, not merge.** Existing offsets for the name are wiped before the new ones are written, so the connector starts from exactly the requested offsets. - **Null offset values are rejected with a 400.** On create the wipe has already 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. - **Response wording** reuses the existing KIP-875 message template, so the framework-managed variant reads "The Connect framework-managed offsets ...". - Because the wipe and write happen in a single `Worker` call, a connector's `alterOffsets()` hook may receive a map mixing tombstones (reset) and values (alter). This is permitted by the `alterOffsets()` contract, which documents that a single map may contain `null` values for reset alongside non-null values for alter; `POST /connectors` with `initial_offsets` is simply the first path to produce this at scale. ### Testing - `WorkerTest`: replacing offsets for source and sink connectors (with and without 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-config ordering (via `InOrder`), null-value rejection, and — distributed — the cleanup that wipes the offsets when the config write fails. - `ConnectWorkerIntegrationTest`: end-to-end for both offset mechanisms — a source connector created at offset 5 produces only its last 5 messages (and the response carries `offsets_status`), and a sink connector created at offset 5 never sees a record below offset 5. ### Committer Checklist (excluded from commit message) - [ ] Verify design and implementation - [ ] Verify test coverage and CI build status - [ ] Verify documentation (including upgrade notes) --- _Developed with assistance from Claude (Anthropic)._ -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
