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

removing the unnecessary restart logic and relying on the seek mechanism we already implemented

Browse Source
This commit is contained in:
chrislu
2025-10-15 10:50:54 -07:00
parent 6c1298b5f7
commit 3b75e50b04
1 changed files with 0 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
weed/mq/kafka/integration/seaweedmq_handler.go
View File

@@ -94,28 +94,6 @@ func (h *SeaweedMQHandler) GetStoredRecords(ctx context.Context, topic string, p
glog.Errorf("[FETCH] ReadRecords failed: %v", err)
return nil, fmt.Errorf("failed to read records: %v", err)
}
// CRITICAL FIX: If ReadRecords returns 0 but HWM indicates data exists on disk, force a disk read
// This handles the case where subscriber advanced past data that was already on disk
// Only do this ONCE per fetch request to avoid subscriber churn
if len(seaweedRecords) == 0 {
hwm, hwmErr := brokerClient.GetHighWaterMark(topic, partition)
if hwmErr == nil && fromOffset < hwm {
// Restart the existing subscriber at the requested offset for disk read
// This is more efficient than closing and recreating
consumerGroup := "kafka-gateway"
consumerID := fmt.Sprintf("kafka-gateway-%s-%d", topic, partition)
if err := brokerClient.RestartSubscriber(brokerSubscriber, fromOffset, consumerGroup, consumerID); err != nil {
return nil, fmt.Errorf("failed to restart subscriber: %v", err)
}
// Try reading again from restarted subscriber (will do disk read)
seaweedRecords, err = brokerClient.ReadRecordsFromOffset(ctx, brokerSubscriber, fromOffset, maxRecords)
if err != nil {
return nil, fmt.Errorf("failed to read after restart: %v", err)
}
}
}
glog.V(2).Infof("[FETCH] ReadRecords returned %d records", len(seaweedRecords))
//