lsm/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2025-11-09 03:40:35 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

avoid deadlock

Browse Source
This commit is contained in:
chrislu
2025-10-14 07:51:06 -07:00
parent fe9e0161d5
commit 73ebc69a82
1 changed files with 5 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
weed/mq/kafka/integration/broker_client_subscribe.go
View File

@@ -287,6 +287,11 @@ func (bc *BrokerClient) ReadRecordsFromOffset(ctx context.Context, session *Brok
session.Key(), requestedOffset, cacheStartOffset, cacheEndOffset)
}
}
// CRITICAL: Get the current offset atomically before making recreation decision
// We need to unlock first (lock acquired at line 257) then re-acquire for atomic read
currentStartOffset := session.StartOffset
session.mu.Unlock()
// CRITICAL FIX for Schema Registry: Keep subscriber alive across multiple fetch requests
// Schema Registry expects to make multiple poll() calls on the same consumer connection
@@ -299,12 +304,6 @@ func (bc *BrokerClient) ReadRecordsFromOffset(ctx context.Context, session *Brok
// The session will naturally advance as records are consumed, so we should NOT
// recreate it just because requestedOffset != session.StartOffset
// CRITICAL: Re-check the offset under session lock to prevent race conditions
// Another thread might be reading from this session right now and advancing the offset
session.mu.Lock()
currentStartOffset := session.StartOffset
session.mu.Unlock()
if requestedOffset < currentStartOffset {
// Need to seek backward - close old session and create a fresh subscriber
// Restarting an existing stream doesn't work reliably because the broker may still