From ffc45a538d535c2cc9f374918c7db68c09809299 Mon Sep 17 00:00:00 2001 From: chrislu Date: Mon, 13 Oct 2025 21:19:38 -0700 Subject: [PATCH] Added bounds checking after calculating startIdx. Problem: Race condition in cache lookup logic: Thread A reads cache metadata (17+ records, endOffset = 32) Thread B modifies/truncates the cache to 17 records Thread A calculates startIdx = 19 based on old metadata Slice operation consumedRecords[19:17] panics --- .../integration/broker_client_subscribe.go | 19 +++++++++++++------ 1 file changed, 13 insertions(+), 6 deletions(-) diff --git a/weed/mq/kafka/integration/broker_client_subscribe.go b/weed/mq/kafka/integration/broker_client_subscribe.go index e3ee6954b..09015ec60 100644 --- a/weed/mq/kafka/integration/broker_client_subscribe.go +++ b/weed/mq/kafka/integration/broker_client_subscribe.go @@ -271,13 +271,20 @@ func (bc *BrokerClient) ReadRecordsFromOffset(ctx context.Context, session *Brok if requestedOffset >= cacheStartOffset && requestedOffset <= cacheEndOffset { // Found in cache startIdx := int(requestedOffset - cacheStartOffset) - endIdx := startIdx + maxRecords - if endIdx > len(session.consumedRecords) { - endIdx = len(session.consumedRecords) + // CRITICAL: Bounds check to prevent race condition where cache is modified between checks + if startIdx < 0 || startIdx >= len(session.consumedRecords) { + glog.V(2).Infof("[FETCH] Cache index out of bounds (race condition): startIdx=%d, cache size=%d, falling through to normal read", + startIdx, len(session.consumedRecords)) + // Cache was modified, fall through to normal read path + } else { + endIdx := startIdx + maxRecords + if endIdx > len(session.consumedRecords) { + endIdx = len(session.consumedRecords) + } + glog.V(2).Infof("[FETCH] Returning %d cached records for offset %d", endIdx-startIdx, requestedOffset) + session.mu.Unlock() + return session.consumedRecords[startIdx:endIdx], nil } - glog.V(2).Infof("[FETCH] Returning %d cached records for offset %d", endIdx-startIdx, requestedOffset) - session.mu.Unlock() - return session.consumedRecords[startIdx:endIdx], nil } }