[CORE-14911] Fix _schemas replay for a truncated log - #31602
WillemKauf wants to merge 2 commits into
Conversation
The RPC consume path built a reader without validating the requested
offset, unlike the Kafka fetch path. Reading below the partition's start
offset therefore threw `translation_offset_out_of_range` out of the
handler, escaping its own error mapping and reaching the caller as a
generic failure. Validate the offset first and return a new
`cluster::errc::offset_out_of_range` error.
`get_offsets()` now also reports the partition's start offset, so a caller
can find the earliest readable offset instead of guessing at `model::offset{0}`.
0f57129 to
1185599
Compare
There was a problem hiding this comment.
Pull request overview
Improves schema registry resiliency when the internal _schemas topic has been prefix-truncated (log start offset > 0), by teaching the schema registry transports to discover the log start offset and by making both startup replay and incremental reads resume from the earliest surviving offset instead of failing with offset_out_of_range.
Changes:
- Extend the schema registry transport abstraction with
get_log_start()and use it to replay_schemasfrom the earliest available offset on startup. - Make incremental reads tolerate truncation mid-flight by detecting
offset_out_of_range, refreshing log start, and resuming from the surviving range. - Extend Kafka data RPC offset reporting to include
start_offset, and add targeted tests covering truncated replay and offset-out-of-range behavior.
Reviewed changes
Copilot reviewed 20 out of 20 changed files in this pull request and generated no comments.
Show a summary per file
| File | Description |
|---|---|
| src/v/pandaproxy/schema_registry/transport.h | Adds get_log_start() to the transport interface. |
| src/v/pandaproxy/schema_registry/service.cc | Replays _schemas starting at log start and logs truncation with guidance. |
| src/v/pandaproxy/schema_registry/seq_writer.h | Introduces do_wait_for() to encapsulate reader-shard logic safely. |
| src/v/pandaproxy/schema_registry/seq_writer.cc | Recovers from offset_out_of_range during incremental reads by resyncing to log start. |
| src/v/pandaproxy/schema_registry/rpc_transport.h | Declares get_log_start() for the RPC-backed transport. |
| src/v/pandaproxy/schema_registry/rpc_transport.cc | Implements log-start retrieval via get_partition_offsets; maps offset_out_of_range. |
| src/v/pandaproxy/schema_registry/kafka_client_transport.h | Declares get_log_start() and a shared list_offset() helper. |
| src/v/pandaproxy/schema_registry/kafka_client_transport.cc | Implements log-start/high-watermark via ListOffsets earliest/latest timestamps. |
| src/v/pandaproxy/schema_registry/BUILD | Adds protocol deps needed for ListOffsets usage. |
| src/v/pandaproxy/schema_registry/test/utils.h | Updates test transports to satisfy the new get_log_start() interface. |
| src/v/pandaproxy/schema_registry/test/rpc_transport_test.cc | Adds coverage for get_log_start() and consuming below log start via RPC transport. |
| src/v/pandaproxy/schema_registry/test/truncated_replay.cc | New fixture test covering end-to-end replay from truncated _schemas (RPC + kafka_client modes). |
| src/v/pandaproxy/schema_registry/test/BUILD | Registers the new truncated replay gtest target. |
| src/v/kafka/data/rpc/service.cc | Exposes partition start_offset and returns offset_out_of_range on invalid/translation offsets. |
| src/v/kafka/data/rpc/serde.h | Extends partition_offsets serde payload with optional start_offset (v1, compat v0). |
| src/v/kafka/data/rpc/serde.cc | Updates formatting to include start_offset. |
| src/v/kafka/data/rpc/test/deps.h | Enhances in-memory proxy to model start offsets and validate fetch offsets for tests. |
| src/v/kafka/server/errors.h | Maps cluster::errc::offset_out_of_range to Kafka offset_out_of_range. |
| src/v/cluster/errc.h | Adds cluster::errc::offset_out_of_range and its error message. |
| src/v/cluster/errors.cc | Adds string formatting support for the new cluster::errc value. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
Recovery replayed `_schemas` from a hardcoded offset 0, and the transport had no way to ask where the log actually starts. Once the topic has been prefix truncated that read is out of range, so the registry fails to load and every request retries a fetch request with an offset below the log's start. The Kafka client transport gets `offset_out_of_range`, and the RPC transport gets the offset translator's out-of-range error. Add `transport::get_log_start()` (`ListOffsets` at the earliest timestamp, or the start offset the `get_offsets` RPC now reports) and replay from there, logging an error that says what was deleted, since availability is preferable to a registry that cannot start. Also, performing re-syncing of the start offset if we end up in a case where the log is prefix truncated after replay but during normal operations in `seq_writer::wait_for()`.
1185599 to
b60534d
Compare
|
code looks sensible to me. Q: Are we sure operating with a truncated schemas topic won't break other invariants downstream? First thing that springs to mind is transitive compatibility rules. e.g. you've lost some old subject version so we don't reject some new version that would break it, producers write freely in the new schema but some downstream pipeline expected the whole log to be self compatible. Your change seems like an operational improvement prima facie, but I want to be sure we've considered other angles. |
Great question and I have a feeling the answer isn't exactly cut and dry. maybe it is true that some people would rather the schema registry refuse to start if it had data loss associated with it for these safety reasons. we're already in between a rock and a hard place if we get into this situation, so maybe it is better to fail catastrophically 🤷 |
Retry command for Build#88826please wait until all jobs are finished before running the slash command |
CI test resultstest results on build#88826
|
Yeah. Similar question for references, i.e. could prefix truncation break some reference link that would have been validated at creation time? Idk exactly how that stuff works but maybe @pgellert can provide some color. Broadly, I suspect many parts of SR's design hinge on the underlying topic being compact-only, but I don't know enough to make that claim offhand. |
pgellert
left a comment
pgellert
left a comment
There was a problem hiding this comment.
I think it makes sense for us to support replaying the schema registry from a non-0 log start offset under the assumption that the contents of the schemas topic are valid.
But I don't think we should try to support any guarantees in the scenario that is motivating this PR of accidental prefix truncation.
There are a bunch of assumptions that could be broken if we start up under these circumstance (e.g. the transitive compat check issue Oren mentioned; in the case of missing references the referring schemas would be unparsable and would block any further schema registration if any compat checks need to be done against them; the modes/compatibility levels of some subjects could change; and just generally that whole prefix is gone so clients will likely see missing schema errors due to that). And it seems counterproductive to try to support any guarantees here long-term.
In most cases, in this scenario I think the resolution would be to re-create their schemas topic, so it seems preferable to not even start up their schema registry.
But again, it seems fine to me to support this new behaviour, but for different motivations than what the PR sets out.
| vlog( | ||
| srlog.error, | ||
| "The {} topic has been truncated to offset {}: records [0, {}) have " | ||
| "been deleted and the schemas, subject configs and modes they held " | ||
| "are lost. Recovering from offset {} with whatever survives. The " | ||
| "registry may be missing schemas that produced data still refers to, " | ||
| "and may reissue a schema id that was assigned in the deleted range. " | ||
| "Restore {} from a backup, and do not enable deletion (a cleanup " | ||
| "policy including 'delete', or any retention limit) on it.", | ||
| model::schema_registry_internal_tp.topic, | ||
| start_offset, | ||
| start_offset, | ||
| start_offset, | ||
| model::schema_registry_internal_tp.topic); |
There was a problem hiding this comment.
If we add support for this scenario of a prefix-truncated log, I think we should treat this scenario as a non-error.
We've been considering prefix truncation as an escape hatch as part of "manually compacting" a large schemas topic. If we did go ahead with that, we should avoid printing large and scary error log messages on every restart.
I think, since a prefix-truncated schemas topic is not necessarily a cause for concern, but rather just highly unusual, I'd keep this one a brief WARN log line here.
If we are unfortunate enough to be in a bad situation where
_schemashas been prefix truncated and the start offset is no longer 0, either_schemasreplay upon restart OR log read requests through the RPC client are all doomed to fail.Since a working schema registry with truncated data is preferable to one that refuses to start or fails to serve requests, the changes here enable users to inspect the start offset of a log through the Kafka RPC client, and the
schema_registrynow uses this tool to replay from the start offset of the log and to allow re-syncing of the log start offset during normal read operations which would otherwise be stuck.We will be loud about these cases with
ERRORlog lines, since we really shouldn't have a truncated_schemastopic in the first place.Backports Required
Release Notes
Improvements
_schemastopic instead of failing to start the schema registry or serving requests from it.