admin/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2025-12-06 17:18:35 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: Fix Fix replication txn data loss during chaos (#44963)

Browse Source 
Only confirm CommitMsg for txn messages to prevent data loss.

issue: https://github.com/milvus-io/milvus/issues/44962,
https://github.com/milvus-io/milvus/issues/44123

Signed-off-by: bigsheeper <yihao.dai@zilliz.com>
This commit is contained in:
yihao.dai 2025-10-21 14:16:04 +08:00 committed by GitHub
parent 21076196bf
commit 0c61d969bb
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 6 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
internal/proxy/replicate/replicate_stream_server.go
View File

@ -125,7 +125,7 @@ func (p *ReplicateStreamServer) handleReplicateMessage(req *milvuspb.ReplicateRe
// Append message to wal. // Append message to wal.
_, err := streaming.WAL().Replicate().Append(p.streamServer.Context(), msg) _, err := streaming.WAL().Replicate().Append(p.streamServer.Context(), msg)
if err == nil { if err == nil {
p.sendReplicateResult(sourceTs) p.sendReplicateResult(sourceTs, msg)
return nil return nil
} }
if status.AsStreamingError(err).IsIgnoredOperation() { if status.AsStreamingError(err).IsIgnoredOperation() {
@ -138,7 +138,11 @@ func (p *ReplicateStreamServer) handleReplicateMessage(req *milvuspb.ReplicateRe
} }
// sendReplicateResult sends the replicate result to client. // sendReplicateResult sends the replicate result to client.
func (p *ReplicateStreamServer) sendReplicateResult(sourceTimeTick uint64) { func (p *ReplicateStreamServer) sendReplicateResult(sourceTimeTick uint64, msg message.ReplicateMutableMessage) {
if msg.TxnContext() != nil && msg.MessageType() != message.MessageTypeCommitTxn {
// Only confirm the commit message of a transaction.
return
}
resp := &milvuspb.ReplicateResponse{ resp := &milvuspb.ReplicateResponse{
Response: &milvuspb.ReplicateResponse_ReplicateConfirmedMessageInfo{ Response: &milvuspb.ReplicateResponse_ReplicateConfirmedMessageInfo{
ReplicateConfirmedMessageInfo: &milvuspb.ReplicateConfirmedMessageInfo{ ReplicateConfirmedMessageInfo: &milvuspb.ReplicateConfirmedMessageInfo{