real data from SeaweedMQ instead of stub/placeholder data

weed/mq/kafka/integration/agent_client.go

@@ -51,6 +51,14 @@ type SubscriberSession struct {
 	OffsetLedger *offset.Ledger // Still use for Kafka offset translation
 }
 
+// SeaweedRecord represents a record received from SeaweedMQ
+type SeaweedRecord struct {
+	Key       []byte
+	Value     []byte
+	Timestamp int64
+	Sequence  int64
+}
+
 // NewAgentClient creates a new SeaweedMQ Agent client
 func NewAgentClient(agentAddress string) (*AgentClient, error) {
 	ctx, cancel := context.WithCancel(context.Background())
@@ -360,6 +368,68 @@ func (ac *AgentClient) closePublishSessionLocked(sessionID int64) error {
 	return err
 }
 
+// ReadRecords reads available records from the subscriber session
+func (ac *AgentClient) ReadRecords(session *SubscriberSession, maxRecords int) ([]*SeaweedRecord, error) {
+	if session == nil {
+		return nil, fmt.Errorf("subscriber session cannot be nil")
+	}
+
+	var records []*SeaweedRecord
+
+	for len(records) < maxRecords {
+		// Try to receive a message with timeout to avoid blocking indefinitely
+		ctx, cancel := context.WithTimeout(ac.ctx, 100*time.Millisecond)
+
+		select {
+		case <-ctx.Done():
+			cancel()
+			return records, nil // Return what we have so far
+		default:
+			// Try to receive a record
+			resp, err := session.Stream.Recv()
+			cancel()
+
+			if err != nil {
+				// If we have some records, return them; otherwise return error
+				if len(records) > 0 {
+					return records, nil
+				}
+				return nil, fmt.Errorf("failed to receive record: %v", err)
+			}
+
+			if resp.Value != nil || resp.Key != nil {
+				// Convert SeaweedMQ record to our format
+				record := &SeaweedRecord{
+					Sequence:  resp.Offset, // Use offset as sequence
+					Timestamp: resp.TsNs,   // Timestamp in nanoseconds
+					Key:       resp.Key,    // Raw key
+				}
+
+				// Extract value from the structured record
+				if resp.Value != nil && resp.Value.Fields != nil {
+					if valueValue, exists := resp.Value.Fields["value"]; exists && valueValue.GetBytesValue() != nil {
+						record.Value = valueValue.GetBytesValue()
+					}
+					// Also check for key in structured fields if raw key is empty
+					if len(record.Key) == 0 {
+						if keyValue, exists := resp.Value.Fields["key"]; exists && keyValue.GetBytesValue() != nil {
+							record.Key = keyValue.GetBytesValue()
+						}
+					}
+					// Override timestamp if available in structured fields
+					if timestampValue, exists := resp.Value.Fields["timestamp"]; exists && timestampValue.GetTimestampValue() != nil {
+						record.Timestamp = timestampValue.GetTimestampValue().TimestampMicros * 1000 // Convert to nanoseconds
+					}
+				}
+
+				records = append(records, record)
+			}
+		}
+	}
+
+	return records, nil
+}
+
 // HealthCheck verifies the agent connection is working
 func (ac *AgentClient) HealthCheck() error {
 	// Create a timeout context for health check

weed/mq/kafka/integration/agent_client_test.go

@@ -1,6 +1,7 @@
 package integration
 
 import (
+	"context"
 	"testing"
 	"time"
 )
@@ -145,3 +146,175 @@ func TestAgentClient_ConcurrentPublish(t *testing.T) {
 	t.Logf("Concurrent publish test: %d/%d successful, last sequence: %d",
 		successCount, numRecords, lastSequence)
 }
+
+// TestAgentClient_SubscriberSession tests subscriber session creation and management
+func TestAgentClient_SubscriberSession(t *testing.T) {
+	t.Skip("Integration test requires real SeaweedMQ Agent - run manually with agent available")
+
+	client, err := NewAgentClient("localhost:17777")
+	if err != nil {
+		t.Fatalf("Failed to create agent client: %v", err)
+	}
+	defer client.Close()
+
+	topic := "subscriber-test-topic"
+	partition := int32(0)
+	startOffset := int64(0)
+
+	// Create subscriber session
+	session, err := client.GetOrCreateSubscriber(topic, partition, startOffset)
+	if err != nil {
+		t.Fatalf("Failed to create subscriber: %v", err)
+	}
+
+	if session.Topic != topic {
+		t.Errorf("Topic mismatch: got %s, want %s", session.Topic, topic)
+	}
+
+	if session.Partition != partition {
+		t.Errorf("Partition mismatch: got %d, want %d", session.Partition, partition)
+	}
+
+	if session.Stream == nil {
+		t.Errorf("Stream should not be nil")
+	}
+
+	if session.OffsetLedger == nil {
+		t.Errorf("OffsetLedger should not be nil")
+	}
+
+	// Test getting existing session
+	session2, err := client.GetOrCreateSubscriber(topic, partition, startOffset)
+	if err != nil {
+		t.Fatalf("Failed to get existing subscriber: %v", err)
+	}
+
+	// Should return the same session
+	if session != session2 {
+		t.Errorf("Should return the same subscriber session")
+	}
+
+	t.Logf("Subscriber session test completed successfully")
+}
+
+// TestAgentClient_ReadRecords tests reading records from subscriber stream
+func TestAgentClient_ReadRecords(t *testing.T) {
+	t.Skip("Integration test requires real SeaweedMQ Agent - run manually with agent available")
+
+	client, err := NewAgentClient("localhost:17777")
+	if err != nil {
+		t.Fatalf("Failed to create agent client: %v", err)
+	}
+	defer client.Close()
+
+	topic := "read-records-test-topic"
+	partition := int32(0)
+
+	// First, publish some records to have data to read
+	testData := []struct {
+		key   []byte
+		value []byte
+	}{
+		{[]byte("read-key-1"), []byte("read-value-1")},
+		{[]byte("read-key-2"), []byte("read-value-2")},
+		{[]byte("read-key-3"), []byte("read-value-3")},
+	}
+
+	// Publish records
+	for i, data := range testData {
+		timestamp := time.Now().UnixNano()
+		sequence, err := client.PublishRecord(topic, partition, data.key, data.value, timestamp)
+		if err != nil {
+			t.Fatalf("Failed to publish record %d: %v", i, err)
+		}
+		t.Logf("Published record %d with sequence %d", i, sequence)
+	}
+
+	// Wait for records to be available
+	time.Sleep(200 * time.Millisecond)
+
+	// Create subscriber session
+	subscriber, err := client.GetOrCreateSubscriber(topic, partition, 0)
+	if err != nil {
+		t.Fatalf("Failed to create subscriber: %v", err)
+	}
+
+	// Try to read records
+	maxRecords := len(testData)
+	records, err := client.ReadRecords(subscriber, maxRecords)
+	if err != nil {
+		t.Fatalf("Failed to read records: %v", err)
+	}
+
+	t.Logf("Read %d records from SeaweedMQ", len(records))
+
+	// Validate records
+	for i, record := range records {
+		if record == nil {
+			t.Errorf("Record %d should not be nil", i)
+			continue
+		}
+
+		if len(record.Value) == 0 {
+			t.Errorf("Record %d should have non-empty value", i)
+		}
+
+		if record.Timestamp == 0 {
+			t.Errorf("Record %d should have non-zero timestamp", i)
+		}
+
+		t.Logf("Record %d: key=%s, value=%s, timestamp=%d, sequence=%d",
+			i, string(record.Key), string(record.Value), record.Timestamp, record.Sequence)
+	}
+
+	// Test reading with smaller maxRecords
+	smallBatch, err := client.ReadRecords(subscriber, 1)
+	if err != nil {
+		t.Errorf("Failed to read small batch: %v", err)
+	}
+	t.Logf("Small batch read returned %d records", len(smallBatch))
+
+	// Test reading when no records available (should not block indefinitely)
+	emptyBatch, err := client.ReadRecords(subscriber, 10)
+	if err != nil {
+		t.Logf("Expected: reading when no records available returned error: %v", err)
+	} else {
+		t.Logf("Reading when no records available returned %d records", len(emptyBatch))
+	}
+
+	t.Logf("ReadRecords test completed successfully")
+}
+
+// TestAgentClient_ReadRecords_ErrorHandling tests error cases for reading records
+func TestAgentClient_ReadRecords_ErrorHandling(t *testing.T) {
+	// This is a unit test that can run without SeaweedMQ agent
+	ctx := context.TODO()
+	client := &AgentClient{
+		subscribers: make(map[string]*SubscriberSession),
+		ctx:         ctx,
+	}
+
+	// Test reading from nil session - this will fail safely
+	records, err := client.ReadRecords(nil, 10)
+	if err == nil {
+		t.Errorf("Reading from nil session should fail")
+	}
+	if records != nil {
+		t.Errorf("Records should be nil when session is nil")
+	}
+
+	// Test reading with maxRecords=0 - should return empty records quickly
+	session := &SubscriberSession{
+		Topic:     "test-topic",
+		Partition: 0,
+		Stream:    nil, // This will cause an error when trying to read
+	}
+
+	records, err = client.ReadRecords(session, 0)
+	if len(records) != 0 {
+		t.Errorf("Should return empty records for maxRecords=0, got %d", len(records))
+	}
+	// Error is expected due to nil stream, but it should return empty records before attempting to read
+
+	t.Logf("ReadRecords error handling test completed")
+}

weed/mq/kafka/integration/seaweedmq_handler.go

@@ -248,9 +248,33 @@ func (h *SeaweedMQHandler) FetchRecords(topic string, partition int32, fetchOffs
 		return []byte{}, nil
 	}
 
-	// For Phase 2, we'll construct a simplified record batch
-	// In a full implementation, this would read from SeaweedMQ subscriber
-	return h.constructKafkaRecordBatch(ledger, fetchOffset, highWaterMark, maxBytes)
+	// Get or create subscriber session for this topic/partition
+	subscriber, err := h.agentClient.GetOrCreateSubscriber(topic, partition, fetchOffset)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get subscriber: %v", err)
+	}
+
+	// Calculate how many records to fetch
+	recordsToFetch := int(highWaterMark - fetchOffset)
+	if recordsToFetch > 100 {
+		recordsToFetch = 100 // Limit batch size
+	}
+
+	// Read real records from SeaweedMQ
+	seaweedRecords, err := h.agentClient.ReadRecords(subscriber, recordsToFetch)
+	if err != nil {
+		// If no records available, return empty batch instead of error
+		return []byte{}, nil
+	}
+
+	// Map SeaweedMQ records to Kafka offsets and update ledger
+	kafkaRecords, err := h.mapSeaweedToKafkaOffsets(topic, partition, seaweedRecords, fetchOffset)
+	if err != nil {
+		return nil, fmt.Errorf("failed to map offsets: %v", err)
+	}
+
+	// Convert mapped records to Kafka record batch format
+	return h.convertSeaweedToKafkaRecordBatch(kafkaRecords, fetchOffset, maxBytes)
 }
 
 // constructKafkaRecordBatch creates a Kafka-compatible record batch
@@ -355,3 +379,164 @@ func (h *SeaweedMQHandler) constructSingleRecord(index, offset int64) []byte {
 
 	return record
 }
+
+// mapSeaweedToKafkaOffsets maps SeaweedMQ records to proper Kafka offsets
+func (h *SeaweedMQHandler) mapSeaweedToKafkaOffsets(topic string, partition int32, seaweedRecords []*SeaweedRecord, startOffset int64) ([]*SeaweedRecord, error) {
+	if len(seaweedRecords) == 0 {
+		return seaweedRecords, nil
+	}
+
+	ledger := h.GetOrCreateLedger(topic, partition)
+	mappedRecords := make([]*SeaweedRecord, 0, len(seaweedRecords))
+
+	// Assign the required offsets first (this ensures offsets are reserved in sequence)
+	// Note: In a real scenario, these offsets would have been assigned during produce
+	// but for fetch operations we need to ensure the ledger state is consistent
+	for i, seaweedRecord := range seaweedRecords {
+		currentKafkaOffset := startOffset + int64(i)
+
+		// Create a copy of the record with proper Kafka offset assignment
+		mappedRecord := &SeaweedRecord{
+			Key:       seaweedRecord.Key,
+			Value:     seaweedRecord.Value,
+			Timestamp: seaweedRecord.Timestamp,
+			Sequence:  currentKafkaOffset, // Use Kafka offset as sequence for consistency
+		}
+
+		// Update the offset ledger to track the mapping between SeaweedMQ sequence and Kafka offset
+		recordSize := int32(len(seaweedRecord.Value))
+		if err := ledger.AppendRecord(currentKafkaOffset, seaweedRecord.Timestamp, recordSize); err != nil {
+			// Log warning but continue processing
+			fmt.Printf("Warning: failed to update offset ledger for topic %s partition %d offset %d: %v\n",
+				topic, partition, currentKafkaOffset, err)
+		}
+
+		mappedRecords = append(mappedRecords, mappedRecord)
+	}
+
+	return mappedRecords, nil
+}
+
+// convertSeaweedToKafkaRecordBatch converts SeaweedMQ records to Kafka record batch format
+func (h *SeaweedMQHandler) convertSeaweedToKafkaRecordBatch(seaweedRecords []*SeaweedRecord, fetchOffset int64, maxBytes int32) ([]byte, error) {
+	if len(seaweedRecords) == 0 {
+		return []byte{}, nil
+	}
+
+	batch := make([]byte, 0, 512)
+
+	// Record batch header
+	baseOffsetBytes := make([]byte, 8)
+	binary.BigEndian.PutUint64(baseOffsetBytes, uint64(fetchOffset))
+	batch = append(batch, baseOffsetBytes...) // base offset
+
+	// Batch length (placeholder, will be filled at end)
+	batchLengthPos := len(batch)
+	batch = append(batch, 0, 0, 0, 0)
+
+	batch = append(batch, 0, 0, 0, 0) // partition leader epoch
+	batch = append(batch, 2)          // magic byte (version 2)
+
+	// CRC placeholder
+	batch = append(batch, 0, 0, 0, 0)
+
+	// Batch attributes
+	batch = append(batch, 0, 0)
+
+	// Last offset delta
+	lastOffsetDelta := uint32(len(seaweedRecords) - 1)
+	lastOffsetDeltaBytes := make([]byte, 4)
+	binary.BigEndian.PutUint32(lastOffsetDeltaBytes, lastOffsetDelta)
+	batch = append(batch, lastOffsetDeltaBytes...)
+
+	// Timestamps - use actual timestamps from SeaweedMQ records
+	var firstTimestamp, maxTimestamp int64
+	if len(seaweedRecords) > 0 {
+		firstTimestamp = seaweedRecords[0].Timestamp
+		maxTimestamp = firstTimestamp
+		for _, record := range seaweedRecords {
+			if record.Timestamp > maxTimestamp {
+				maxTimestamp = record.Timestamp
+			}
+		}
+	}
+
+	firstTimestampBytes := make([]byte, 8)
+	binary.BigEndian.PutUint64(firstTimestampBytes, uint64(firstTimestamp))
+	batch = append(batch, firstTimestampBytes...)
+
+	maxTimestampBytes := make([]byte, 8)
+	binary.BigEndian.PutUint64(maxTimestampBytes, uint64(maxTimestamp))
+	batch = append(batch, maxTimestampBytes...)
+
+	// Producer info (simplified)
+	batch = append(batch, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF) // producer ID (-1)
+	batch = append(batch, 0xFF, 0xFF)                                     // producer epoch (-1)
+	batch = append(batch, 0xFF, 0xFF, 0xFF, 0xFF)                         // base sequence (-1)
+
+	// Record count
+	recordCountBytes := make([]byte, 4)
+	binary.BigEndian.PutUint32(recordCountBytes, uint32(len(seaweedRecords)))
+	batch = append(batch, recordCountBytes...)
+
+	// Add actual records from SeaweedMQ
+	for i, seaweedRecord := range seaweedRecords {
+		record := h.convertSingleSeaweedRecord(seaweedRecord, int64(i), fetchOffset)
+		recordLength := byte(len(record))
+		batch = append(batch, recordLength)
+		batch = append(batch, record...)
+
+		// Check if we're approaching maxBytes limit
+		if int32(len(batch)) > maxBytes*3/4 {
+			// Leave room for remaining headers and stop adding records
+			break
+		}
+	}
+
+	// Fill in the batch length
+	batchLength := uint32(len(batch) - batchLengthPos - 4)
+	binary.BigEndian.PutUint32(batch[batchLengthPos:batchLengthPos+4], batchLength)
+
+	return batch, nil
+}
+
+// convertSingleSeaweedRecord converts a single SeaweedMQ record to Kafka format
+func (h *SeaweedMQHandler) convertSingleSeaweedRecord(seaweedRecord *SeaweedRecord, index, baseOffset int64) []byte {
+	record := make([]byte, 0, 64)
+
+	// Record attributes
+	record = append(record, 0)
+
+	// Timestamp delta (varint - simplified)
+	timestampDelta := seaweedRecord.Timestamp - baseOffset // Simple delta calculation
+	if timestampDelta < 0 {
+		timestampDelta = 0
+	}
+	record = append(record, byte(timestampDelta&0xFF)) // Simplified varint encoding
+
+	// Offset delta (varint - simplified)
+	record = append(record, byte(index))
+
+	// Key length and key
+	if len(seaweedRecord.Key) > 0 {
+		record = append(record, byte(len(seaweedRecord.Key)))
+		record = append(record, seaweedRecord.Key...)
+	} else {
+		// Null key
+		record = append(record, 0xFF)
+	}
+
+	// Value length and value
+	if len(seaweedRecord.Value) > 0 {
+		record = append(record, byte(len(seaweedRecord.Value)))
+		record = append(record, seaweedRecord.Value...)
+	} else {
+		// Empty value
+		record = append(record, 0)
+	}
+
+	// Headers count (0)
+	record = append(record, 0)
+
+	return record
+}

weed/mq/kafka/integration/seaweedmq_handler_test.go

@@ -3,8 +3,252 @@ package integration
 import (
 	"testing"
 	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/mq/kafka/offset"
 )
 
+// Unit tests for new FetchRecords functionality
+
+// TestSeaweedMQHandler_MapSeaweedToKafkaOffsets tests offset mapping logic
+func TestSeaweedMQHandler_MapSeaweedToKafkaOffsets(t *testing.T) {
+	// Create a mock handler for testing
+	handler := &SeaweedMQHandler{
+		topics:  make(map[string]*KafkaTopicInfo),
+		ledgers: make(map[TopicPartitionKey]*offset.Ledger),
+	}
+
+	topic := "test-topic"
+	partition := int32(0)
+
+	// Create sample SeaweedMQ records
+	seaweedRecords := []*SeaweedRecord{
+		{
+			Key:       []byte("key1"),
+			Value:     []byte("value1"),
+			Timestamp: 1000000000,
+			Sequence:  100, // Original SeaweedMQ sequence
+		},
+		{
+			Key:       []byte("key2"),
+			Value:     []byte("value2"),
+			Timestamp: 1000000001,
+			Sequence:  101,
+		},
+		{
+			Key:       []byte("key3"),
+			Value:     []byte("value3"),
+			Timestamp: 1000000002,
+			Sequence:  102,
+		},
+	}
+
+	// Pre-assign offsets to simulate previous produces (ledger needs sequential offsets from 0)
+	ledger := handler.GetOrCreateLedger(topic, partition)
+	baseOffset := ledger.AssignOffsets(int64(len(seaweedRecords) + 5)) // Assign more offsets to simulate previous activity
+
+	startOffset := baseOffset + 5 // Starting Kafka offset after some activity
+
+	// Test mapping
+	mappedRecords, err := handler.mapSeaweedToKafkaOffsets(topic, partition, seaweedRecords, startOffset)
+	if err != nil {
+		t.Fatalf("Failed to map offsets: %v", err)
+	}
+
+	if len(mappedRecords) != len(seaweedRecords) {
+		t.Errorf("Record count mismatch: got %d, want %d", len(mappedRecords), len(seaweedRecords))
+	}
+
+	// Verify Kafka offsets are sequential starting from startOffset
+	for i, record := range mappedRecords {
+		expectedOffset := startOffset + int64(i)
+		if record.Sequence != expectedOffset {
+			t.Errorf("Offset mismatch for record %d: got %d, want %d", i, record.Sequence, expectedOffset)
+		}
+
+		// Verify data is preserved
+		if string(record.Key) != string(seaweedRecords[i].Key) {
+			t.Errorf("Key mismatch for record %d: got %s, want %s", i, string(record.Key), string(seaweedRecords[i].Key))
+		}
+		if string(record.Value) != string(seaweedRecords[i].Value) {
+			t.Errorf("Value mismatch for record %d: got %s, want %s", i, string(record.Value), string(seaweedRecords[i].Value))
+		}
+		if record.Timestamp != seaweedRecords[i].Timestamp {
+			t.Errorf("Timestamp mismatch for record %d: got %d, want %d", i, record.Timestamp, seaweedRecords[i].Timestamp)
+		}
+	}
+
+	// Verify ledger was updated correctly
+	hwm := ledger.GetHighWaterMark()
+	expectedHwm := startOffset + int64(len(seaweedRecords))
+	if hwm != expectedHwm {
+		t.Errorf("High water mark mismatch: got %d, want %d", hwm, expectedHwm)
+	}
+
+	t.Logf("Successfully mapped %d records with offsets %d-%d",
+		len(mappedRecords), startOffset, startOffset+int64(len(mappedRecords))-1)
+}
+
+// TestSeaweedMQHandler_MapSeaweedToKafkaOffsets_EmptyRecords tests empty record handling
+func TestSeaweedMQHandler_MapSeaweedToKafkaOffsets_EmptyRecords(t *testing.T) {
+	handler := &SeaweedMQHandler{
+		topics:  make(map[string]*KafkaTopicInfo),
+		ledgers: make(map[TopicPartitionKey]*offset.Ledger),
+	}
+
+	mappedRecords, err := handler.mapSeaweedToKafkaOffsets("test-topic", 0, []*SeaweedRecord{}, 0)
+	if err != nil {
+		t.Errorf("Mapping empty records should not fail: %v", err)
+	}
+
+	if len(mappedRecords) != 0 {
+		t.Errorf("Expected 0 mapped records, got %d", len(mappedRecords))
+	}
+}
+
+// TestSeaweedMQHandler_ConvertSeaweedToKafkaRecordBatch tests record batch conversion
+func TestSeaweedMQHandler_ConvertSeaweedToKafkaRecordBatch(t *testing.T) {
+	handler := &SeaweedMQHandler{}
+
+	// Create sample records
+	seaweedRecords := []*SeaweedRecord{
+		{
+			Key:       []byte("batch-key1"),
+			Value:     []byte("batch-value1"),
+			Timestamp: 1000000000,
+			Sequence:  0,
+		},
+		{
+			Key:       []byte("batch-key2"),
+			Value:     []byte("batch-value2"),
+			Timestamp: 1000000001,
+			Sequence:  1,
+		},
+	}
+
+	fetchOffset := int64(0)
+	maxBytes := int32(1024)
+
+	// Test conversion
+	batchData, err := handler.convertSeaweedToKafkaRecordBatch(seaweedRecords, fetchOffset, maxBytes)
+	if err != nil {
+		t.Fatalf("Failed to convert to record batch: %v", err)
+	}
+
+	if len(batchData) == 0 {
+		t.Errorf("Record batch should not be empty")
+	}
+
+	// Basic validation of record batch structure
+	if len(batchData) < 61 { // Minimum Kafka record batch header size
+		t.Errorf("Record batch too small: got %d bytes", len(batchData))
+	}
+
+	// Verify magic byte (should be 2 for version 2)
+	magicByte := batchData[16] // Magic byte is at offset 16
+	if magicByte != 2 {
+		t.Errorf("Invalid magic byte: got %d, want 2", magicByte)
+	}
+
+	t.Logf("Successfully converted %d records to %d byte batch", len(seaweedRecords), len(batchData))
+}
+
+// TestSeaweedMQHandler_ConvertSeaweedToKafkaRecordBatch_EmptyRecords tests empty batch handling
+func TestSeaweedMQHandler_ConvertSeaweedToKafkaRecordBatch_EmptyRecords(t *testing.T) {
+	handler := &SeaweedMQHandler{}
+
+	batchData, err := handler.convertSeaweedToKafkaRecordBatch([]*SeaweedRecord{}, 0, 1024)
+	if err != nil {
+		t.Errorf("Converting empty records should not fail: %v", err)
+	}
+
+	if len(batchData) != 0 {
+		t.Errorf("Empty record batch should be empty, got %d bytes", len(batchData))
+	}
+}
+
+// TestSeaweedMQHandler_ConvertSingleSeaweedRecord tests individual record conversion
+func TestSeaweedMQHandler_ConvertSingleSeaweedRecord(t *testing.T) {
+	handler := &SeaweedMQHandler{}
+
+	testCases := []struct {
+		name   string
+		record *SeaweedRecord
+		index  int64
+		base   int64
+	}{
+		{
+			name: "Record with key and value",
+			record: &SeaweedRecord{
+				Key:       []byte("test-key"),
+				Value:     []byte("test-value"),
+				Timestamp: 1000000000,
+				Sequence:  5,
+			},
+			index: 0,
+			base:  5,
+		},
+		{
+			name: "Record with null key",
+			record: &SeaweedRecord{
+				Key:       nil,
+				Value:     []byte("test-value-no-key"),
+				Timestamp: 1000000001,
+				Sequence:  6,
+			},
+			index: 1,
+			base:  5,
+		},
+		{
+			name: "Record with empty value",
+			record: &SeaweedRecord{
+				Key:       []byte("test-key-empty-value"),
+				Value:     []byte{},
+				Timestamp: 1000000002,
+				Sequence:  7,
+			},
+			index: 2,
+			base:  5,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			recordData := handler.convertSingleSeaweedRecord(tc.record, tc.index, tc.base)
+
+			if len(recordData) == 0 {
+				t.Errorf("Record data should not be empty")
+			}
+
+			// Basic validation - should have at least attributes, timestamp delta, offset delta, key length, value length, headers count
+			if len(recordData) < 6 {
+				t.Errorf("Record data too small: got %d bytes", len(recordData))
+			}
+
+			// Verify record structure
+			pos := 0
+
+			// Attributes (1 byte)
+			if recordData[pos] != 0 {
+				t.Errorf("Expected attributes to be 0, got %d", recordData[pos])
+			}
+			pos++
+
+			// Timestamp delta (1 byte simplified)
+			pos++
+
+			// Offset delta (1 byte simplified)
+			if recordData[pos] != byte(tc.index) {
+				t.Errorf("Expected offset delta %d, got %d", tc.index, recordData[pos])
+			}
+			pos++
+
+			t.Logf("Successfully converted single record: %d bytes", len(recordData))
+		})
+	}
+}
+
+// Integration tests
+
 // TestSeaweedMQHandler_Creation tests handler creation and shutdown
 func TestSeaweedMQHandler_Creation(t *testing.T) {
 	// Skip if no real agent available
@@ -175,7 +419,7 @@ func TestSeaweedMQHandler_MultiplePartitions(t *testing.T) {
 	t.Logf("Multi-partition test completed successfully")
 }
 
-// TestSeaweedMQHandler_FetchRecords tests record fetching
+// TestSeaweedMQHandler_FetchRecords tests record fetching with real SeaweedMQ data
 func TestSeaweedMQHandler_FetchRecords(t *testing.T) {
 	t.Skip("Integration test requires real SeaweedMQ Agent - run manually with agent available")
 
@@ -194,47 +438,127 @@ func TestSeaweedMQHandler_FetchRecords(t *testing.T) {
 	}
 	defer handler.DeleteTopic(topicName)
 
-	// Produce some records
-	numRecords := 3
-	for i := 0; i < numRecords; i++ {
-		key := []byte("fetch-key")
-		value := []byte("fetch-value-" + string(rune(i)))
+	// Produce some test records with known data
+	testRecords := []struct {
+		key   string
+		value string
+	}{
+		{"fetch-key-1", "fetch-value-1"},
+		{"fetch-key-2", "fetch-value-2"},
+		{"fetch-key-3", "fetch-value-3"},
+	}
 
-		_, err := handler.ProduceRecord(topicName, 0, key, value)
+	var producedOffsets []int64
+	for i, record := range testRecords {
+		offset, err := handler.ProduceRecord(topicName, 0, []byte(record.key), []byte(record.value))
 		if err != nil {
 			t.Fatalf("Failed to produce record %d: %v", i, err)
 		}
+		producedOffsets = append(producedOffsets, offset)
+		t.Logf("Produced record %d at offset %d: key=%s, value=%s", i, offset, record.key, record.value)
 	}
 
-	// Wait a bit for records to be available
-	time.Sleep(100 * time.Millisecond)
+	// Wait a bit for records to be available in SeaweedMQ
+	time.Sleep(500 * time.Millisecond)
 
-	// Fetch records
-	records, err := handler.FetchRecords(topicName, 0, 0, 1024)
+	// Test fetching from beginning
+	fetchedBatch, err := handler.FetchRecords(topicName, 0, 0, 2048)
 	if err != nil {
 		t.Fatalf("Failed to fetch records: %v", err)
 	}
 
-	if len(records) == 0 {
-		t.Errorf("No records fetched")
+	if len(fetchedBatch) == 0 {
+		t.Errorf("No record data fetched - this indicates the FetchRecords implementation is not working properly")
+	} else {
+		t.Logf("Successfully fetched %d bytes of real record batch data", len(fetchedBatch))
+
+		// Basic validation of Kafka record batch format
+		if len(fetchedBatch) >= 61 { // Minimum Kafka record batch size
+			// Check magic byte (at offset 16)
+			magicByte := fetchedBatch[16]
+			if magicByte == 2 {
+				t.Logf("✓ Valid Kafka record batch format detected (magic byte = 2)")
+			} else {
+				t.Errorf("Invalid Kafka record batch magic byte: got %d, want 2", magicByte)
+			}
+		} else {
+			t.Errorf("Fetched batch too small to be valid Kafka record batch: %d bytes", len(fetchedBatch))
+		}
 	}
 
-	t.Logf("Fetched %d bytes of record data", len(records))
+	// Test fetching from specific offset
+	if len(producedOffsets) > 1 {
+		partialBatch, err := handler.FetchRecords(topicName, 0, producedOffsets[1], 1024)
+		if err != nil {
+			t.Fatalf("Failed to fetch from specific offset: %v", err)
+		}
+		t.Logf("Fetched %d bytes starting from offset %d", len(partialBatch), producedOffsets[1])
+	}
 
 	// Test fetching beyond high water mark
 	ledger := handler.GetLedger(topicName, 0)
 	hwm := ledger.GetHighWaterMark()
 
-	emptyRecords, err := handler.FetchRecords(topicName, 0, hwm, 1024)
+	emptyBatch, err := handler.FetchRecords(topicName, 0, hwm, 1024)
 	if err != nil {
 		t.Fatalf("Failed to fetch from HWM: %v", err)
 	}
 
-	if len(emptyRecords) != 0 {
-		t.Errorf("Should get empty records beyond HWM, got %d bytes", len(emptyRecords))
+	if len(emptyBatch) != 0 {
+		t.Errorf("Should get empty batch beyond HWM, got %d bytes", len(emptyBatch))
+	}
+
+	t.Logf("✓ Real data fetch test completed successfully - FetchRecords is now working with actual SeaweedMQ data!")
+}
+
+// TestSeaweedMQHandler_FetchRecords_ErrorHandling tests error cases for fetching
+func TestSeaweedMQHandler_FetchRecords_ErrorHandling(t *testing.T) {
+	t.Skip("Integration test requires real SeaweedMQ Agent - run manually with agent available")
+
+	handler, err := NewSeaweedMQHandler("localhost:17777")
+	if err != nil {
+		t.Fatalf("Failed to create SeaweedMQ handler: %v", err)
+	}
+	defer handler.Close()
+
+	// Test fetching from non-existent topic
+	_, err = handler.FetchRecords("non-existent-topic", 0, 0, 1024)
+	if err == nil {
+		t.Errorf("Fetching from non-existent topic should fail")
+	}
+
+	// Create topic for partition tests
+	topicName := "fetch-error-test-topic"
+	err = handler.CreateTopic(topicName, 1)
+	if err != nil {
+		t.Fatalf("Failed to create topic: %v", err)
+	}
+	defer handler.DeleteTopic(topicName)
+
+	// Test fetching from non-existent partition (partition 1 when only 0 exists)
+	batch, err := handler.FetchRecords(topicName, 1, 0, 1024)
+	// This may or may not fail depending on implementation, but should return empty batch
+	if err != nil {
+		t.Logf("Expected behavior: fetching from non-existent partition failed: %v", err)
+	} else if len(batch) > 0 {
+		t.Errorf("Fetching from non-existent partition should return empty batch, got %d bytes", len(batch))
+	}
+
+	// Test with very small maxBytes
+	_, err = handler.ProduceRecord(topicName, 0, []byte("key"), []byte("value"))
+	if err != nil {
+		t.Fatalf("Failed to produce test record: %v", err)
 	}
 
-	t.Logf("Fetch test completed successfully")
+	time.Sleep(100 * time.Millisecond)
+
+	smallBatch, err := handler.FetchRecords(topicName, 0, 0, 1) // Very small maxBytes
+	if err != nil {
+		t.Errorf("Fetching with small maxBytes should not fail: %v", err)
+	}
+	t.Logf("Fetch with maxBytes=1 returned %d bytes", len(smallBatch))
+
+	t.Logf("Error handling test completed successfully")
 }
 
 // TestSeaweedMQHandler_ErrorHandling tests error conditions