mirror of
https://github.com/chrislusf/seaweedfs
synced 2025-09-19 01:30:23 +02:00
326 lines
11 KiB
Go
326 lines
11 KiB
Go
package protocol
|
|
|
|
import (
|
|
"context"
|
|
"encoding/binary"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
func TestHandler_handleFetch(t *testing.T) {
|
|
h := NewTestHandler()
|
|
correlationID := uint32(666)
|
|
|
|
// Create a topic and add some records
|
|
topicName := "test-topic"
|
|
// Mock SeaweedMQ handler for testing - in real tests, this would use a proper mock
|
|
// For now, just comment out the topic creation as it's handled by SeaweedMQ handler
|
|
|
|
// Add some records to the ledger
|
|
ledger := h.GetOrCreateLedger(topicName, 0)
|
|
baseOffset := ledger.AssignOffsets(3)
|
|
currentTime := time.Now().UnixNano()
|
|
ledger.AppendRecord(baseOffset+0, currentTime+0, 100)
|
|
ledger.AppendRecord(baseOffset+1, currentTime+1000, 200)
|
|
ledger.AppendRecord(baseOffset+2, currentTime+2000, 150)
|
|
|
|
// Build a Fetch request
|
|
requestBody := make([]byte, 0, 256)
|
|
|
|
// NOTE: client_id is handled by HandleConn and stripped before reaching handler
|
|
// Start directly with Fetch-specific fields
|
|
|
|
// Replica ID (-1 for consumer)
|
|
requestBody = append(requestBody, 0xFF, 0xFF, 0xFF, 0xFF)
|
|
|
|
// Max wait time (5000ms)
|
|
requestBody = append(requestBody, 0, 0, 0x13, 0x88)
|
|
|
|
// Min bytes (1)
|
|
requestBody = append(requestBody, 0, 0, 0, 1)
|
|
|
|
// Max bytes (1MB)
|
|
requestBody = append(requestBody, 0, 0x10, 0, 0)
|
|
|
|
// Isolation level (0 = read uncommitted)
|
|
requestBody = append(requestBody, 0)
|
|
|
|
// Session ID (0)
|
|
requestBody = append(requestBody, 0, 0, 0, 0)
|
|
|
|
// Epoch (0)
|
|
requestBody = append(requestBody, 0, 0, 0, 0)
|
|
|
|
// Topics count (1)
|
|
requestBody = append(requestBody, 0, 0, 0, 1)
|
|
|
|
// Topic name
|
|
requestBody = append(requestBody, 0, byte(len(topicName)))
|
|
requestBody = append(requestBody, []byte(topicName)...)
|
|
|
|
// Partitions count (1)
|
|
requestBody = append(requestBody, 0, 0, 0, 1)
|
|
|
|
// Partition 0
|
|
requestBody = append(requestBody, 0, 0, 0, 0) // partition ID
|
|
// NOTE: current leader epoch only in v9+, not v7
|
|
requestBody = append(requestBody, 0, 0, 0, 0, 0, 0, 0, byte(baseOffset)) // fetch offset
|
|
requestBody = append(requestBody, 0, 0, 0, 0, 0, 0, 0, 0) // log start offset
|
|
requestBody = append(requestBody, 0, 0, 0x10, 0) // partition max bytes (1MB)
|
|
|
|
response, err := h.handleFetch(context.Background(), correlationID, 7, requestBody)
|
|
if err != nil {
|
|
t.Fatalf("handleFetch: %v", err)
|
|
}
|
|
|
|
if len(response) < 60 { // minimum expected size
|
|
t.Fatalf("response too short: %d bytes", len(response))
|
|
}
|
|
|
|
// Check response structure
|
|
respCorrelationID := binary.BigEndian.Uint32(response[0:4])
|
|
if respCorrelationID != correlationID {
|
|
t.Errorf("correlation ID: got %d, want %d", respCorrelationID, correlationID)
|
|
}
|
|
|
|
// Check throttle time
|
|
throttleTime := binary.BigEndian.Uint32(response[4:8])
|
|
if throttleTime != 0 {
|
|
t.Errorf("throttle time: got %d, want 0", throttleTime)
|
|
}
|
|
|
|
// Check error code
|
|
errorCode := binary.BigEndian.Uint16(response[8:10])
|
|
if errorCode != 0 {
|
|
t.Errorf("error code: got %d, want 0", errorCode)
|
|
}
|
|
|
|
// Parse response structure (simplified validation)
|
|
offset := 14 // skip correlation_id + throttle_time + error_code + session_id
|
|
topicsCount := binary.BigEndian.Uint32(response[offset : offset+4])
|
|
if topicsCount != 1 {
|
|
t.Errorf("topics count: got %d, want 1", topicsCount)
|
|
}
|
|
|
|
offset += 4
|
|
respTopicNameSize := binary.BigEndian.Uint16(response[offset : offset+2])
|
|
offset += 2
|
|
if respTopicNameSize != uint16(len(topicName)) {
|
|
t.Errorf("response topic name size: got %d, want %d", respTopicNameSize, len(topicName))
|
|
}
|
|
|
|
respTopicName := string(response[offset : offset+int(respTopicNameSize)])
|
|
offset += int(respTopicNameSize)
|
|
if respTopicName != topicName {
|
|
t.Errorf("response topic name: got %s, want %s", respTopicName, topicName)
|
|
}
|
|
|
|
// Partitions count
|
|
respPartitionsCount := binary.BigEndian.Uint32(response[offset : offset+4])
|
|
offset += 4
|
|
if respPartitionsCount != 1 {
|
|
t.Errorf("response partitions count: got %d, want 1", respPartitionsCount)
|
|
}
|
|
|
|
// Partition ID
|
|
partitionID := binary.BigEndian.Uint32(response[offset : offset+4])
|
|
offset += 4
|
|
if partitionID != 0 {
|
|
t.Errorf("partition ID: got %d, want 0", partitionID)
|
|
}
|
|
|
|
// Partition error code
|
|
partitionErrorCode := binary.BigEndian.Uint16(response[offset : offset+2])
|
|
offset += 2
|
|
if partitionErrorCode != 0 {
|
|
t.Errorf("partition error code: got %d, want 0", partitionErrorCode)
|
|
}
|
|
|
|
// High water mark
|
|
highWaterMark := int64(binary.BigEndian.Uint64(response[offset : offset+8]))
|
|
offset += 8
|
|
if highWaterMark != 3 { // baseOffset + 3 records
|
|
t.Errorf("high water mark: got %d, want %d", highWaterMark, baseOffset+3)
|
|
}
|
|
|
|
// Skip last_stable_offset, log_start_offset, aborted_transactions_count
|
|
offset += 8 + 8 + 4
|
|
|
|
// Records size
|
|
recordsSize := binary.BigEndian.Uint32(response[offset : offset+4])
|
|
offset += 4
|
|
if recordsSize == 0 {
|
|
t.Errorf("expected some records, got size 0")
|
|
}
|
|
|
|
// Verify we have records data
|
|
if len(response) < offset+int(recordsSize) {
|
|
t.Errorf("response shorter than expected records size")
|
|
}
|
|
}
|
|
|
|
func TestHandler_handleFetch_UnknownTopic(t *testing.T) {
|
|
h := NewTestHandler()
|
|
correlationID := uint32(777)
|
|
|
|
// Build Fetch request for non-existent topic (don't create it)
|
|
topicName := "non-existent-topic"
|
|
|
|
requestBody := make([]byte, 0, 128)
|
|
|
|
// NOTE: client_id is handled by HandleConn and stripped before reaching handler
|
|
// Start directly with Fetch-specific fields
|
|
|
|
// Standard Fetch parameters
|
|
requestBody = append(requestBody, 0xFF, 0xFF, 0xFF, 0xFF) // replica ID
|
|
requestBody = append(requestBody, 0, 0, 0x13, 0x88) // max wait time
|
|
requestBody = append(requestBody, 0, 0, 0, 1) // min bytes
|
|
requestBody = append(requestBody, 0, 0x10, 0, 0) // max bytes
|
|
requestBody = append(requestBody, 0) // isolation level
|
|
requestBody = append(requestBody, 0, 0, 0, 0) // session ID
|
|
requestBody = append(requestBody, 0, 0, 0, 0) // epoch
|
|
|
|
// Topics count (1)
|
|
requestBody = append(requestBody, 0, 0, 0, 1)
|
|
|
|
// Topic name
|
|
requestBody = append(requestBody, 0, byte(len(topicName)))
|
|
requestBody = append(requestBody, []byte(topicName)...)
|
|
|
|
// Partitions count (1)
|
|
requestBody = append(requestBody, 0, 0, 0, 1)
|
|
|
|
// Partition 0
|
|
requestBody = append(requestBody, 0, 0, 0, 0) // partition ID
|
|
requestBody = append(requestBody, 0, 0, 0, 0) // current leader epoch
|
|
requestBody = append(requestBody, 0, 0, 0, 0, 0, 0, 0, 0) // fetch offset
|
|
requestBody = append(requestBody, 0, 0, 0, 0, 0, 0, 0, 0) // log start offset
|
|
requestBody = append(requestBody, 0, 0, 0x10, 0) // partition max bytes
|
|
|
|
response, err := h.handleFetch(context.Background(), correlationID, 7, requestBody)
|
|
if err != nil {
|
|
t.Fatalf("handleFetch: %v", err)
|
|
}
|
|
|
|
// Parse response to check for UNKNOWN_TOPIC_OR_PARTITION error
|
|
offset := 14 + 4 + 2 + len(topicName) + 4 + 4 // skip to partition error code
|
|
partitionErrorCode := binary.BigEndian.Uint16(response[offset : offset+2])
|
|
if partitionErrorCode != 3 { // UNKNOWN_TOPIC_OR_PARTITION
|
|
t.Errorf("expected UNKNOWN_TOPIC_OR_PARTITION error (3), got: %d", partitionErrorCode)
|
|
}
|
|
}
|
|
|
|
func TestHandler_handleFetch_EmptyPartition(t *testing.T) {
|
|
h := NewTestHandler()
|
|
correlationID := uint32(888)
|
|
|
|
// Create a topic but don't add any records
|
|
topicName := "empty-topic"
|
|
// Mock SeaweedMQ handler for testing - in real tests, this would use a proper mock
|
|
// For now, just comment out the topic creation as it's handled by SeaweedMQ handler
|
|
|
|
// Get ledger but don't add records
|
|
ledger := h.GetOrCreateLedger(topicName, 0)
|
|
_ = ledger // ledger exists but is empty
|
|
|
|
// Build Fetch request
|
|
requestBody := make([]byte, 0, 128)
|
|
|
|
// NOTE: client_id is handled by HandleConn and stripped before reaching handler
|
|
// Start directly with Fetch-specific fields
|
|
|
|
// Standard parameters
|
|
requestBody = append(requestBody, 0xFF, 0xFF, 0xFF, 0xFF) // replica ID
|
|
requestBody = append(requestBody, 0, 0, 0x13, 0x88) // max wait time
|
|
requestBody = append(requestBody, 0, 0, 0, 1) // min bytes
|
|
requestBody = append(requestBody, 0, 0x10, 0, 0) // max bytes
|
|
requestBody = append(requestBody, 0) // isolation level
|
|
requestBody = append(requestBody, 0, 0, 0, 0) // session ID
|
|
requestBody = append(requestBody, 0, 0, 0, 0) // epoch
|
|
|
|
// Topics count (1)
|
|
requestBody = append(requestBody, 0, 0, 0, 1)
|
|
|
|
// Topic name
|
|
requestBody = append(requestBody, 0, byte(len(topicName)))
|
|
requestBody = append(requestBody, []byte(topicName)...)
|
|
|
|
// Partitions count (1)
|
|
requestBody = append(requestBody, 0, 0, 0, 1)
|
|
|
|
// Partition 0 - fetch from offset 0
|
|
requestBody = append(requestBody, 0, 0, 0, 0) // partition ID
|
|
// NOTE: current leader epoch only in v9+, not v7
|
|
requestBody = append(requestBody, 0, 0, 0, 0, 0, 0, 0, 0) // fetch offset
|
|
requestBody = append(requestBody, 0, 0, 0, 0, 0, 0, 0, 0) // log start offset
|
|
requestBody = append(requestBody, 0, 0, 0x10, 0) // partition max bytes
|
|
|
|
response, err := h.handleFetch(context.Background(), correlationID, 7, requestBody)
|
|
if err != nil {
|
|
t.Fatalf("handleFetch: %v", err)
|
|
}
|
|
|
|
// Parse response - should have no error but empty records
|
|
offset := 14 + 4 + 2 + len(topicName) + 4 + 4 // skip to partition error code
|
|
partitionErrorCode := binary.BigEndian.Uint16(response[offset : offset+2])
|
|
if partitionErrorCode != 0 {
|
|
t.Errorf("partition error code: got %d, want 0", partitionErrorCode)
|
|
}
|
|
|
|
// High water mark should be 0
|
|
offset += 2
|
|
highWaterMark := int64(binary.BigEndian.Uint64(response[offset : offset+8]))
|
|
if highWaterMark != 0 {
|
|
t.Errorf("high water mark: got %d, want 0", highWaterMark)
|
|
}
|
|
|
|
// Skip to records size
|
|
offset += 8 + 8 + 4 // skip last_stable_offset, log_start_offset, aborted_transactions_count
|
|
recordsSize := binary.BigEndian.Uint32(response[offset : offset+4])
|
|
if recordsSize != 0 {
|
|
t.Errorf("records size: got %d, want 0", recordsSize)
|
|
}
|
|
}
|
|
|
|
func TestHandler_constructRecordBatch(t *testing.T) {
|
|
h := NewTestHandler()
|
|
|
|
// Test with simple parameters
|
|
records := h.constructRecordBatch(nil, 0, 3)
|
|
if len(records) == 0 {
|
|
t.Errorf("expected some records, got empty")
|
|
}
|
|
|
|
// Should have proper record batch structure
|
|
if len(records) < 61 { // minimum record batch header size
|
|
t.Errorf("record batch too small: %d bytes", len(records))
|
|
}
|
|
|
|
// Check base offset
|
|
baseOffset := int64(binary.BigEndian.Uint64(records[0:8]))
|
|
if baseOffset != 0 {
|
|
t.Errorf("base offset: got %d, want 0", baseOffset)
|
|
}
|
|
|
|
// Check magic byte
|
|
if records[16] != 2 {
|
|
t.Errorf("magic byte: got %d, want 2", records[16])
|
|
}
|
|
|
|
// Test with no records to fetch
|
|
emptyRecords := h.constructRecordBatch(nil, 5, 5)
|
|
if len(emptyRecords) != 0 {
|
|
t.Errorf("expected empty records, got %d bytes", len(emptyRecords))
|
|
}
|
|
|
|
// Test with large range (should be limited)
|
|
largeRecords := h.constructRecordBatch(nil, 0, 100)
|
|
if len(largeRecords) == 0 {
|
|
t.Errorf("expected some records for large range")
|
|
}
|
|
|
|
// Should be limited to reasonable size
|
|
if len(largeRecords) > 10000 { // arbitrary reasonable limit
|
|
t.Errorf("record batch too large: %d bytes", len(largeRecords))
|
|
}
|
|
}
|