feat: add reset checkpoint command for offline MQ switch - #523
Merged
Conversation
|
@tinswzy Thanks for your contribution. Please submit with DCO, see the contributing guide https://github.com/milvus-io/milvus/blob/master/CONTRIBUTING.md#developer-certificate-of-origin-dco. |
tinswzy
force-pushed
the
feat/reset-checkpoint
branch
from
August 27, 2026 04:14
b1036dc to
50b4325
Compare
|
Tick the box to add this pull request to the merge queue (same as
|
Switching a stopped instance onto a different MQ requires rewriting every
persisted position at once. `repair checkpoint` cannot do it: it only touches
one vchannel's channel-cp and knows nothing about the streamingnode metadata
introduced in 2.6, so streamingnode still reopens the old WAL.
Add `reset checkpoint --target-wal <woodpecker|pulsar|kafka|rocksmq>`, which
rewrites all five kinds of persisted position in one run:
streamingnode-meta/wal/{pchannel}/consume-checkpoint MessageId
datacoord-meta/channel-cp/{vchannel} MsgID + WALName
root-coord/.../collection-info StartPositions
datacoord-meta/s/... SegmentInfo.StartPosition
datacoord-meta/s/... SegmentInfo.DmlPosition
The last three matter because datacoord's seek falls back three levels
(channel-cp -> segment position -> collection start). Rewriting only the first
leaves the old encoding reachable whenever a vchannel's channel-cp is missing.
WALName is set on every position. Milvus picks the decoder from it, and for
woodpecker it is the only signal separating an empty earliest id (which
serializes to zero bytes) from a position that was never set.
Timestamps are preserved: they are global TSOs, unrelated to which MQ stores
the message. Milvus filters the replay with DeliverFilterTimeTickGTE against
them, so starting from earliest cannot replay anything older than the
checkpoint.
Two preflight gates refuse to run when a reset would lose data: growing
segments (their rows only exist in the old MQ) and an in-flight broadcast task.
Replication topology is not gated, and WALCheckpoint.ReplicateCheckpoint is not
rewritten. That field is a subscription position against a remote cluster, owned
and maintained by Milvus; switching the local WAL says nothing about it, and the
two clusters are independent. Milvus puts no such gate on its own WAL switch
either.
Note there is no safe intermediate state, in any write order: the consume
checkpoint decides which WAL to open and the channel checkpoints decide which
positions to decode, so a partial run crashes milvus on startup. The command is
therefore idempotent and detects a split state, and the guarantee it offers is
"safe to re-run", not "safe to start after a partial run".
Verified end to end against milvus 3.0.0 with real etcd/minio/pulsar/kafka:
pulsar<->woodpecker and pulsar/woodpecker->kafka all switch with data intact.
rocksmq as a target is blocked upstream (v3.0.0's rmq builder uses server.Rmq
without lazy init, so a switched-to rocksmq WAL cannot open); milvus main has
since added the nil check.
Signed-off-by: tinswzy <zhenyuan.wei@zilliz.com>
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
tinswzy
force-pushed
the
feat/reset-checkpoint
branch
from
August 27, 2026 07:22
50b4325 to
4011513
Compare
congqixia
reviewed
Aug 27, 2026
Addresses review feedback on the two cleanup/preflight paths.
A --pchannel-scoped run used to skip the channel-removal markers and the
querycoord target cache wholesale, on the rationale that they "are not
partitioned by pchannel". That rationale was wrong for both:
datacoord-meta/channel-removal/{channelName} keyed by channel
queryCoord-Collection-Target/{collectionID} keyed by collection
Skipping the target cache is the one that hurts. querycoord recovers it into
its in-memory current target (TargetManager.Recover) and hands the seek
positions inside straight to querynodes (the task executor's req.Checkpoint).
Those positions carry the old WAL's encoding and decode cleanly, so nothing
flags them as stale -- a scoped reset would leave querynodes seeking the new MQ
with old-WAL positions. Nor does the split-state guard catch it: detectSplitState
is itself scoped by --pchannel, so it only compares within the reset shard.
Both prefixes are now filtered per key instead: removal markers by channel name,
targets by "does this collection own any vchannel in scope". A collection
sharded across several pchannels keeps positions for all of them in one target,
so touching any one vchannel dirties the whole entry and it has to go. Dropping
the cache costs only a rebuild from datacoord, which an ordinary restart does
anyway. Whole-instance behaviour is unchanged -- everything is in scope.
Also drop the ErrBroadcastTaskNotFound branch in checkNoPendingBroadcast.
ListWalBroadcast never returns that sentinel; only ListWalBroadcastByID does.
A missing prefix already yields an empty slice, which passes the pending check
below, so the branch was unreachable and its comment described behaviour the
code does not have.
Tests: whole-instance run drops everything; scoped run drops the reset shard's
marker and target while leaving another pchannel's alone; dry-run deletes
nothing. The scoped test fails against the previous skip-wholesale behaviour.
Signed-off-by: tinswzy <zhenyuan.wei@zilliz.com>
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Scoping the target cleanup by collection narrowed the whole-instance path by accident: a collection whose meta is already gone leaves its cached target behind, and ListCollections cannot see it, so enumerating collections misses those orphans where the previous RemoveWithPrefix caught them. Restore the prefix-wide delete when --pchannel is absent, and keep the per-collection filtering only for scoped runs. A scoped run cannot attribute an orphan to a pchannel, so it leaves it alone rather than guessing. Signed-off-by: tinswzy <zhenyuan.wei@zilliz.com> Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Scoped runs delete individual keys while whole-instance runs delete whole subtrees, but the summary hardcoded "prefix(es) to DELETE" for both, overstating the blast radius of a --pchannel run. Report the unit the plan actually uses. Signed-off-by: tinswzy <zhenyuan.wei@zilliz.com> Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The cleanup treated datacoord-meta/channel-removal as a set of stale tombstones left over from the old WAL. It is not one. Despite the name it is a two-valued flag, and MarkChannelDeleted has no callers in Milvus today -- so every key under the prefix is the NonRemoveFlagTomestone that MarkChannelAdded writes. That write happens once, in the create-collection DDL callback (rootcoord -> datacoord WatchChannels); nothing rebuilds it on restart. It carries no MQ position, so a rewind does not invalidate it. Worse, datacoord reads it back through ChannelExists to decide whether to hold back GC of dropped segments whose DML position is ahead of the channel checkpoint (garbage_collector.go). Deleting it flips that guard off for the channel, permanently and silently. Drop the deletion and say so in the "not touched" notice, alongside the pchannel assignment and channelwatch info. The two remaining cleanups stand: segment-assign records cannot self-heal after a rewind because the new WAL never delivers the insert/flush messages they wait for, and the querycoord target cache embeds old-WAL seek positions that get handed to querynodes. Signed-off-by: tinswzy <zhenyuan.wei@zilliz.com> Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Collaborator
|
[APPROVALNOTIFIER] This PR is APPROVED This pull-request has been approved by: congqixia, tinswzy The full list of commands accepted by this bot can be found here. The pull request process is described here DetailsNeeds approval from an approver in each of these files:
Approvers can indicate their approval by writing |
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
3 participants
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.
Closes #522
Adds
reset checkpoint, which rewrites every persisted MQ position so a stopped instance can be brought back up on a different MQ.What it writes
streamingnode-meta/wal/{p}/consume-checkpointMessageIddatacoord-meta/channel-cp/{v}MsgID+WALNameroot-coord/.../collection-infoStartPositionsdatacoord-meta/s/...SegmentInfo.StartPositiondatacoord-meta/s/...SegmentInfo.DmlPositionThe last three matter because datacoord's seek falls back three levels. Rewriting only channel-cp leaves the old encoding reachable through segment and collection positions whenever a vchannel's channel-cp is missing.
It also drops adjacent state that a switch invalidates:
segment-assign(stale allocations pinned to the old WAL — the growing-segment gate guarantees these are garbage),channel-removal, and the querycoord target cache. It deliberately leavesstreamingcoord-meta/pchanneland the vchannel schema timeline alone:PChannelMetacarries no position, and the schema timeline must survive.Design notes
New
resetnamespace rather than anotherrepairsubcommand.repairfixes broken metadata and its commands run against a live cluster; this one deliberately discards valid state and requires milvus to be stopped. Burying a "running this online destroys the cluster" mode inside a family people run online seemed like the wrong trade.Timestamps are preserved. They are global TSOs, unrelated to which MQ stores the message. Milvus filters the replay with
DeliverFilterTimeTickGTEagainst them, so starting from earliest cannot replay anything older than the checkpoint. This is also why there is nolatestoption — a persisted checkpoint is always consumed throughDeliverPolicyStartFrom, never as aDeliverPolicy_Latestpolicy, so a latest sentinel written here becomes a position that does not exist and the reader tails it forever. I tried it; the collection never loads.WALNameis set on every position. Milvus picks the decoder from it, and for woodpecker it is the only thing separating an empty earliest id from an unset one.kafka uses offset 0, not
OffsetBeginning.kafkaID.Marshalencodes withEncodeInt64whileunmarshalMessageIDdecodes withDecodeUint64, which rejects the minus sign — a negative offset written into a checkpoint cannot be read back. On the fresh topic this command targets, 0 is the earliest offset.Two preflight gates, each refusing rather than warning: growing segments (their rows only exist in the old MQ) and an in-flight broadcast task (a half-applied DDL).
Replication topology is deliberately not gated, and
WALCheckpoint.ReplicateCheckpointis deliberately not rewritten. That field is a subscription position against a remote cluster, owned and maintained by Milvus; switching the local WAL says nothing about it, and the two clusters are independent. Milvus puts no such gate on its own WAL switch either.No safe intermediate state. The consume checkpoint decides which WAL opens; the channel checkpoints decide which positions get decoded. When they disagree milvus panics in
getRecoveryInfos. Reordering only flips which side is stale, so the command is idempotent, detects a split state and says so, and the guarantee is "safe to re-run" rather than "safe to start after a partial run". This is spelled out in the command description and in the error returned on a failed write.Verification
12 test functions using an embedded etcd (
embed.StartEtcd) — real protobuf seeded and read back, no mocks.tests/e2e/commands/test_reset.shfollows the existing convention and covers the dry-run and rejection paths.Separately, end to end against milvus 3.0.0 with real etcd/minio/pulsar/kafka, checking row content per row (vectors derived from the id) plus an ANN search, not just counts:
Also covered: 4-shard collections, the growing-segment gate (and that
--allow-growingreally does lose the unflushed rows while flushed data survives), idempotent re-runs, and recovery from a deliberately interrupted run. That harness needs pulsar and kafka containers, so it does not fit the single-MQtests/e2ecompose stack; happy to contribute it separately if that would be useful.rocksmq as a target does not work on v3.0.0, and the cause is upstream, not here:
pkg/streaming/walimpls/impls/rmq/builder.gousesserver.Rmqwithout initialising it, and that singleton is only set up when rocksmq is the selected WAL from process start. A freshly created rocksmq instance works fine; only a switched-to one fails, withRmq server is niland a stalled channel tsafe. milvus main has since added theif server.Rmq == nil { InitRocksMQ(...) }guard.Notes for review
pkg/v3/go-api/v3after enhance: upgrade milvus proto go-api to v3 and migrate milvus deps #518.make birdwatcher_wkafka(CGO + librdkafka), as with the existing kafka paths.states/etcd/common/wal_recovery_write.goalongside the ones repair: restore vchannels the streaming node's recovery metadata has lost #531 added;wal_recovery_storage.gois untouched.🤖 Generated with Claude Code