From bb0e6132756f6db4f9f51df580baf9ac1ee5eb55 Mon Sep 17 00:00:00 2001 From: chrislu Date: Tue, 14 Oct 2025 16:58:50 -0700 Subject: [PATCH] more time --- weed/mq/kafka/protocol/fetch.go | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/weed/mq/kafka/protocol/fetch.go b/weed/mq/kafka/protocol/fetch.go index 5d1b66ff5..802351528 100644 --- a/weed/mq/kafka/protocol/fetch.go +++ b/weed/mq/kafka/protocol/fetch.go @@ -244,13 +244,15 @@ func (h *Handler) handleFetch(ctx context.Context, correlationID uint32, apiVers // 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 results := make([]*partitionFetchResult, len(pending)) - // Use 90% of client's MaxWaitTime to ensure we return BEFORE client timeout - // This gives us time to package and transmit the response before the client gives up - // For 500ms client timeout, we use 450ms internally, leaving 50ms buffer for: + // Use 95% of client's MaxWaitTime to ensure we return BEFORE client timeout + // This maximizes data collection time while leaving a safety buffer for: // - Response serialization, network transmission, client processing - effectiveDeadlineMs := time.Duration(maxWaitMs) * 9 / 10 + // For 500ms client timeout: 475ms internal fetch, 25ms buffer + // For 100ms client timeout: 95ms internal fetch, 5ms buffer + effectiveDeadlineMs := time.Duration(maxWaitMs) * 95 / 100 deadline := time.After(effectiveDeadlineMs * time.Millisecond) - if maxWaitMs < 10 { + if maxWaitMs < 20 { + // For very short timeouts (< 20ms), use full timeout to maximize data collection deadline = time.After(time.Duration(maxWaitMs) * time.Millisecond) }