lsm/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2025-11-10 07:21:59 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

add some delays

Browse Source
This commit is contained in:
chrislu
2025-10-14 16:15:28 -07:00
parent 090f73dc66
commit e9101d9733
1 changed files with 4 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
weed/mq/kafka/protocol/fetch.go
View File

@@ -244,8 +244,10 @@ func (h *Handler) handleFetch(ctx context.Context, correlationID uint32, apiVers
// Phase 2: Wait for all results with adequate timeout for CI environments // Phase 2: Wait for all results with adequate timeout for CI environments
// CRITICAL: We MUST return a result for every requested partition or Sarama will error // CRITICAL: We MUST return a result for every requested partition or Sarama will error
results := make([]*partitionFetchResult, len(pending)) results := make([]*partitionFetchResult, len(pending))
// Use the client's requested MaxWaitTime // Use the client's requested MaxWaitTime plus a small buffer
deadline := time.After(time.Duration(maxWaitMs) * time.Millisecond) // The 100ms buffer provides backpressure for downstream consumers (like Schema Registry's
// internal consumer) that may need time to catch up after rapid writes
deadline := time.After(time.Duration(maxWaitMs)*time.Millisecond + 100*time.Millisecond)
// Collect results one by one with shared deadline // Collect results one by one with shared deadline
for i, pf := range pending { for i, pf := range pending {