mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-02 10:17:10 +01:00
chrislu
014db6f999
- Fixed throttle_time_ms field: only include in v2+, not v1 - Reduced kafka-go 'unread bytes' error from 60 to 56 bytes - Added comprehensive API request debugging to identify format mismatches - kafka-go now progresses further but still has 56 bytes format issue in some API response Progress: kafka-go client can now parse ListOffsets v1 responses correctly but still fails before making Fetch requests due to remaining API format issues.
943 lines
32 KiB
Go
943 lines
32 KiB
Go
package protocol
|
|
|
|
import (
|
|
"encoding/binary"
|
|
"fmt"
|
|
"hash/crc32"
|
|
"time"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/mq/kafka/compression"
|
|
"github.com/seaweedfs/seaweedfs/weed/mq/kafka/schema"
|
|
"github.com/seaweedfs/seaweedfs/weed/pb/schema_pb"
|
|
)
|
|
|
|
func (h *Handler) handleFetch(correlationID uint32, apiVersion uint16, requestBody []byte) ([]byte, error) {
|
|
fmt.Printf("DEBUG: *** FETCH HANDLER CALLED *** Correlation: %d, Version: %d\n", correlationID, apiVersion)
|
|
// Parse the Fetch request to get the requested topics and partitions
|
|
fetchRequest, err := h.parseFetchRequest(apiVersion, requestBody)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("parse fetch request: %w", err)
|
|
}
|
|
|
|
// Build the response
|
|
response := make([]byte, 0, 1024)
|
|
|
|
// Correlation ID (4 bytes)
|
|
correlationIDBytes := make([]byte, 4)
|
|
binary.BigEndian.PutUint32(correlationIDBytes, correlationID)
|
|
response = append(response, correlationIDBytes...)
|
|
|
|
// Fetch v1+ has throttle_time_ms at the beginning
|
|
if apiVersion >= 1 {
|
|
response = append(response, 0, 0, 0, 0) // throttle_time_ms (4 bytes, 0 = no throttling)
|
|
}
|
|
|
|
// Topics count
|
|
topicsCount := len(fetchRequest.Topics)
|
|
topicsCountBytes := make([]byte, 4)
|
|
binary.BigEndian.PutUint32(topicsCountBytes, uint32(topicsCount))
|
|
response = append(response, topicsCountBytes...)
|
|
|
|
// Process each requested topic
|
|
for _, topic := range fetchRequest.Topics {
|
|
topicNameBytes := []byte(topic.Name)
|
|
|
|
// Topic name length and name
|
|
response = append(response, byte(len(topicNameBytes)>>8), byte(len(topicNameBytes)))
|
|
response = append(response, topicNameBytes...)
|
|
|
|
// Partitions count for this topic
|
|
partitionsCount := len(topic.Partitions)
|
|
partitionsCountBytes := make([]byte, 4)
|
|
binary.BigEndian.PutUint32(partitionsCountBytes, uint32(partitionsCount))
|
|
response = append(response, partitionsCountBytes...)
|
|
|
|
// Process each requested partition
|
|
for _, partition := range topic.Partitions {
|
|
// Partition ID
|
|
partitionIDBytes := make([]byte, 4)
|
|
binary.BigEndian.PutUint32(partitionIDBytes, uint32(partition.PartitionID))
|
|
response = append(response, partitionIDBytes...)
|
|
|
|
// Error code (2 bytes) - 0 = no error
|
|
response = append(response, 0, 0)
|
|
|
|
// Get ledger for this topic-partition to determine high water mark
|
|
ledger := h.GetOrCreateLedger(topic.Name, partition.PartitionID)
|
|
highWaterMark := ledger.GetHighWaterMark()
|
|
|
|
fmt.Printf("DEBUG: Fetch - topic: %s, partition: %d, fetchOffset: %d, highWaterMark: %d\n",
|
|
topic.Name, partition.PartitionID, partition.FetchOffset, highWaterMark)
|
|
|
|
// High water mark (8 bytes)
|
|
highWaterMarkBytes := make([]byte, 8)
|
|
binary.BigEndian.PutUint64(highWaterMarkBytes, uint64(highWaterMark))
|
|
response = append(response, highWaterMarkBytes...)
|
|
|
|
// Fetch v4+ has last_stable_offset and log_start_offset
|
|
if apiVersion >= 4 {
|
|
// Last stable offset (8 bytes) - same as high water mark for non-transactional
|
|
response = append(response, highWaterMarkBytes...)
|
|
// Log start offset (8 bytes) - 0 for simplicity
|
|
response = append(response, 0, 0, 0, 0, 0, 0, 0, 0)
|
|
|
|
// Aborted transactions count (4 bytes) = 0
|
|
response = append(response, 0, 0, 0, 0)
|
|
}
|
|
|
|
// Records - get actual stored record batches
|
|
var recordBatch []byte
|
|
if highWaterMark > partition.FetchOffset {
|
|
// Try to get the actual stored record batch first
|
|
if storedBatch, exists := h.GetRecordBatch(topic.Name, partition.PartitionID, partition.FetchOffset); exists {
|
|
recordBatch = storedBatch
|
|
fmt.Printf("DEBUG: Using stored record batch for offset %d, size: %d bytes\n", partition.FetchOffset, len(storedBatch))
|
|
} else {
|
|
fmt.Printf("DEBUG: No stored record batch found for offset %d, using synthetic batch\n", partition.FetchOffset)
|
|
// Fallback to synthetic batch if no stored batch found
|
|
recordBatch = h.constructSimpleRecordBatch(partition.FetchOffset, highWaterMark)
|
|
fmt.Printf("DEBUG: Using synthetic record batch for offset %d, size: %d bytes\n", partition.FetchOffset, len(recordBatch))
|
|
}
|
|
} else {
|
|
fmt.Printf("DEBUG: No messages available - fetchOffset %d >= highWaterMark %d\n", partition.FetchOffset, highWaterMark)
|
|
recordBatch = []byte{} // No messages available
|
|
}
|
|
|
|
// Records size (4 bytes)
|
|
recordsSizeBytes := make([]byte, 4)
|
|
binary.BigEndian.PutUint32(recordsSizeBytes, uint32(len(recordBatch)))
|
|
response = append(response, recordsSizeBytes...)
|
|
|
|
// Records data
|
|
response = append(response, recordBatch...)
|
|
|
|
fmt.Printf("DEBUG: Would fetch schematized records - topic: %s, partition: %d, offset: %d, maxBytes: %d\n",
|
|
topic.Name, partition.PartitionID, partition.FetchOffset, partition.MaxBytes)
|
|
}
|
|
}
|
|
|
|
return response, nil
|
|
}
|
|
|
|
// FetchRequest represents a parsed Kafka Fetch request
|
|
type FetchRequest struct {
|
|
ReplicaID int32
|
|
MaxWaitTime int32
|
|
MinBytes int32
|
|
MaxBytes int32
|
|
IsolationLevel int8
|
|
Topics []FetchTopic
|
|
}
|
|
|
|
type FetchTopic struct {
|
|
Name string
|
|
Partitions []FetchPartition
|
|
}
|
|
|
|
type FetchPartition struct {
|
|
PartitionID int32
|
|
FetchOffset int64
|
|
LogStartOffset int64
|
|
MaxBytes int32
|
|
}
|
|
|
|
// parseFetchRequest parses a Kafka Fetch request
|
|
func (h *Handler) parseFetchRequest(apiVersion uint16, requestBody []byte) (*FetchRequest, error) {
|
|
if len(requestBody) < 16 {
|
|
return nil, fmt.Errorf("fetch request too short: %d bytes", len(requestBody))
|
|
}
|
|
|
|
offset := 0
|
|
request := &FetchRequest{}
|
|
|
|
// Skip client_id (string) - read length first
|
|
if offset+2 > len(requestBody) {
|
|
return nil, fmt.Errorf("insufficient data for client_id length")
|
|
}
|
|
clientIDLength := int(binary.BigEndian.Uint16(requestBody[offset : offset+2]))
|
|
offset += 2
|
|
|
|
if clientIDLength >= 0 {
|
|
if offset+clientIDLength > len(requestBody) {
|
|
return nil, fmt.Errorf("insufficient data for client_id")
|
|
}
|
|
offset += clientIDLength
|
|
}
|
|
|
|
// Replica ID (4 bytes)
|
|
if offset+4 > len(requestBody) {
|
|
return nil, fmt.Errorf("insufficient data for replica_id")
|
|
}
|
|
request.ReplicaID = int32(binary.BigEndian.Uint32(requestBody[offset : offset+4]))
|
|
offset += 4
|
|
|
|
// Max wait time (4 bytes)
|
|
if offset+4 > len(requestBody) {
|
|
return nil, fmt.Errorf("insufficient data for max_wait_time")
|
|
}
|
|
request.MaxWaitTime = int32(binary.BigEndian.Uint32(requestBody[offset : offset+4]))
|
|
offset += 4
|
|
|
|
// Min bytes (4 bytes)
|
|
if offset+4 > len(requestBody) {
|
|
return nil, fmt.Errorf("insufficient data for min_bytes")
|
|
}
|
|
request.MinBytes = int32(binary.BigEndian.Uint32(requestBody[offset : offset+4]))
|
|
offset += 4
|
|
|
|
// Max bytes (4 bytes) - only in v3+
|
|
if apiVersion >= 3 {
|
|
if offset+4 > len(requestBody) {
|
|
return nil, fmt.Errorf("insufficient data for max_bytes")
|
|
}
|
|
request.MaxBytes = int32(binary.BigEndian.Uint32(requestBody[offset : offset+4]))
|
|
offset += 4
|
|
}
|
|
|
|
// Isolation level (1 byte) - only in v4+
|
|
if apiVersion >= 4 {
|
|
if offset+1 > len(requestBody) {
|
|
return nil, fmt.Errorf("insufficient data for isolation_level")
|
|
}
|
|
request.IsolationLevel = int8(requestBody[offset])
|
|
offset += 1
|
|
}
|
|
|
|
// Session ID (4 bytes) and Session Epoch (4 bytes) - only in v7+
|
|
if apiVersion >= 7 {
|
|
if offset+8 > len(requestBody) {
|
|
return nil, fmt.Errorf("insufficient data for session_id and epoch")
|
|
}
|
|
offset += 8 // Skip session_id and session_epoch
|
|
}
|
|
|
|
// Topics count (4 bytes)
|
|
if offset+4 > len(requestBody) {
|
|
return nil, fmt.Errorf("insufficient data for topics count")
|
|
}
|
|
topicsCount := int(binary.BigEndian.Uint32(requestBody[offset : offset+4]))
|
|
offset += 4
|
|
|
|
// Parse topics
|
|
request.Topics = make([]FetchTopic, topicsCount)
|
|
for i := 0; i < topicsCount; i++ {
|
|
// Topic name length (2 bytes)
|
|
if offset+2 > len(requestBody) {
|
|
return nil, fmt.Errorf("insufficient data for topic name length")
|
|
}
|
|
topicNameLength := int(binary.BigEndian.Uint16(requestBody[offset : offset+2]))
|
|
offset += 2
|
|
|
|
// Topic name
|
|
if offset+topicNameLength > len(requestBody) {
|
|
return nil, fmt.Errorf("insufficient data for topic name")
|
|
}
|
|
request.Topics[i].Name = string(requestBody[offset : offset+topicNameLength])
|
|
offset += topicNameLength
|
|
|
|
// Partitions count (4 bytes)
|
|
if offset+4 > len(requestBody) {
|
|
return nil, fmt.Errorf("insufficient data for partitions count")
|
|
}
|
|
partitionsCount := int(binary.BigEndian.Uint32(requestBody[offset : offset+4]))
|
|
offset += 4
|
|
|
|
// Parse partitions
|
|
request.Topics[i].Partitions = make([]FetchPartition, partitionsCount)
|
|
for j := 0; j < partitionsCount; j++ {
|
|
// Partition ID (4 bytes)
|
|
if offset+4 > len(requestBody) {
|
|
return nil, fmt.Errorf("insufficient data for partition ID")
|
|
}
|
|
request.Topics[i].Partitions[j].PartitionID = int32(binary.BigEndian.Uint32(requestBody[offset : offset+4]))
|
|
offset += 4
|
|
|
|
// Current leader epoch (4 bytes) - only in v9+
|
|
if apiVersion >= 9 {
|
|
if offset+4 > len(requestBody) {
|
|
return nil, fmt.Errorf("insufficient data for current leader epoch")
|
|
}
|
|
offset += 4 // Skip current leader epoch
|
|
}
|
|
|
|
// Fetch offset (8 bytes)
|
|
if offset+8 > len(requestBody) {
|
|
return nil, fmt.Errorf("insufficient data for fetch offset")
|
|
}
|
|
request.Topics[i].Partitions[j].FetchOffset = int64(binary.BigEndian.Uint64(requestBody[offset : offset+8]))
|
|
offset += 8
|
|
|
|
// Log start offset (8 bytes) - only in v5+
|
|
if apiVersion >= 5 {
|
|
if offset+8 > len(requestBody) {
|
|
return nil, fmt.Errorf("insufficient data for log start offset")
|
|
}
|
|
request.Topics[i].Partitions[j].LogStartOffset = int64(binary.BigEndian.Uint64(requestBody[offset : offset+8]))
|
|
offset += 8
|
|
}
|
|
|
|
// Partition max bytes (4 bytes)
|
|
if offset+4 > len(requestBody) {
|
|
return nil, fmt.Errorf("insufficient data for partition max bytes")
|
|
}
|
|
request.Topics[i].Partitions[j].MaxBytes = int32(binary.BigEndian.Uint32(requestBody[offset : offset+4]))
|
|
offset += 4
|
|
}
|
|
}
|
|
|
|
return request, nil
|
|
}
|
|
|
|
// constructRecordBatchFromLedger creates a record batch from messages stored in the ledger
|
|
func (h *Handler) constructRecordBatchFromLedger(ledger interface{}, fetchOffset, highWaterMark int64) []byte {
|
|
// Get the actual ledger interface
|
|
offsetLedger, ok := ledger.(interface {
|
|
GetMessages(startOffset, endOffset int64) []interface{}
|
|
})
|
|
|
|
if !ok {
|
|
// If ledger doesn't support GetMessages, return empty batch
|
|
return []byte{}
|
|
}
|
|
|
|
// Calculate how many records to fetch
|
|
recordsToFetch := highWaterMark - fetchOffset
|
|
if recordsToFetch <= 0 {
|
|
return []byte{} // no records to fetch
|
|
}
|
|
|
|
// Limit the number of records for performance
|
|
if recordsToFetch > 100 {
|
|
recordsToFetch = 100
|
|
}
|
|
|
|
// Get messages from ledger
|
|
messages := offsetLedger.GetMessages(fetchOffset, fetchOffset+recordsToFetch)
|
|
if len(messages) == 0 {
|
|
return []byte{} // no messages available
|
|
}
|
|
|
|
// Create a realistic record batch
|
|
batch := make([]byte, 0, 1024)
|
|
|
|
// Record batch header
|
|
baseOffsetBytes := make([]byte, 8)
|
|
binary.BigEndian.PutUint64(baseOffsetBytes, uint64(fetchOffset))
|
|
batch = append(batch, baseOffsetBytes...) // base offset (8 bytes)
|
|
|
|
// Calculate batch length (will be filled after we know the size)
|
|
batchLengthPos := len(batch)
|
|
batch = append(batch, 0, 0, 0, 0) // batch length placeholder (4 bytes)
|
|
|
|
batch = append(batch, 0, 0, 0, 0) // partition leader epoch (4 bytes)
|
|
batch = append(batch, 2) // magic byte (version 2) (1 byte)
|
|
|
|
// CRC placeholder (4 bytes) - for testing, use 0
|
|
batch = append(batch, 0, 0, 0, 0) // CRC32
|
|
|
|
// Batch attributes (2 bytes) - no compression, no transactional
|
|
batch = append(batch, 0, 0) // attributes
|
|
|
|
// Last offset delta (4 bytes)
|
|
lastOffsetDelta := uint32(len(messages) - 1)
|
|
lastOffsetDeltaBytes := make([]byte, 4)
|
|
binary.BigEndian.PutUint32(lastOffsetDeltaBytes, lastOffsetDelta)
|
|
batch = append(batch, lastOffsetDeltaBytes...)
|
|
|
|
// First timestamp (8 bytes)
|
|
firstTimestamp := time.Now().UnixMilli()
|
|
firstTimestampBytes := make([]byte, 8)
|
|
binary.BigEndian.PutUint64(firstTimestampBytes, uint64(firstTimestamp))
|
|
batch = append(batch, firstTimestampBytes...)
|
|
|
|
// Max timestamp (8 bytes)
|
|
maxTimestamp := firstTimestamp + int64(len(messages)) - 1
|
|
maxTimestampBytes := make([]byte, 8)
|
|
binary.BigEndian.PutUint64(maxTimestampBytes, uint64(maxTimestamp))
|
|
batch = append(batch, maxTimestampBytes...)
|
|
|
|
// Producer ID (8 bytes) - -1 for non-transactional
|
|
batch = append(batch, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF)
|
|
|
|
// Producer Epoch (2 bytes) - -1 for non-transactional
|
|
batch = append(batch, 0xFF, 0xFF)
|
|
|
|
// Base Sequence (4 bytes) - -1 for non-transactional
|
|
batch = append(batch, 0xFF, 0xFF, 0xFF, 0xFF)
|
|
|
|
// Record count (4 bytes)
|
|
recordCountBytes := make([]byte, 4)
|
|
binary.BigEndian.PutUint32(recordCountBytes, uint32(len(messages)))
|
|
batch = append(batch, recordCountBytes...)
|
|
|
|
// Add individual records
|
|
for i, msg := range messages {
|
|
// Try to extract key and value from the message
|
|
var key, value []byte
|
|
|
|
// Handle different message types
|
|
if msgMap, ok := msg.(map[string]interface{}); ok {
|
|
if keyVal, exists := msgMap["key"]; exists {
|
|
if keyBytes, ok := keyVal.([]byte); ok {
|
|
key = keyBytes
|
|
} else if keyStr, ok := keyVal.(string); ok {
|
|
key = []byte(keyStr)
|
|
}
|
|
}
|
|
if valueVal, exists := msgMap["value"]; exists {
|
|
if valueBytes, ok := valueVal.([]byte); ok {
|
|
value = valueBytes
|
|
} else if valueStr, ok := valueVal.(string); ok {
|
|
value = []byte(valueStr)
|
|
}
|
|
}
|
|
}
|
|
|
|
// If we couldn't extract key/value, create default ones
|
|
if value == nil {
|
|
value = []byte(fmt.Sprintf("Message %d", fetchOffset+int64(i)))
|
|
}
|
|
|
|
// Build individual record
|
|
record := make([]byte, 0, 128)
|
|
|
|
// Record attributes (1 byte)
|
|
record = append(record, 0)
|
|
|
|
// Timestamp delta (varint)
|
|
timestampDelta := int64(i)
|
|
record = append(record, encodeVarint(timestampDelta)...)
|
|
|
|
// Offset delta (varint)
|
|
offsetDelta := int64(i)
|
|
record = append(record, encodeVarint(offsetDelta)...)
|
|
|
|
// Key length and key (varint + data)
|
|
if key == nil {
|
|
record = append(record, encodeVarint(-1)...) // null key
|
|
} else {
|
|
record = append(record, encodeVarint(int64(len(key)))...)
|
|
record = append(record, key...)
|
|
}
|
|
|
|
// Value length and value (varint + data)
|
|
record = append(record, encodeVarint(int64(len(value)))...)
|
|
record = append(record, value...)
|
|
|
|
// Headers count (varint) - 0 headers
|
|
record = append(record, encodeVarint(0)...)
|
|
|
|
// Prepend record length (varint)
|
|
recordLength := int64(len(record))
|
|
batch = append(batch, encodeVarint(recordLength)...)
|
|
batch = append(batch, record...)
|
|
}
|
|
|
|
// Fill in the batch length
|
|
batchLength := uint32(len(batch) - batchLengthPos - 4)
|
|
binary.BigEndian.PutUint32(batch[batchLengthPos:batchLengthPos+4], batchLength)
|
|
|
|
return batch
|
|
}
|
|
|
|
// constructSimpleRecordBatch creates a simple record batch for testing
|
|
func (h *Handler) constructSimpleRecordBatch(fetchOffset, highWaterMark int64) []byte {
|
|
recordsToFetch := highWaterMark - fetchOffset
|
|
if recordsToFetch <= 0 {
|
|
return []byte{} // no records to fetch
|
|
}
|
|
|
|
// Limit the number of records for testing
|
|
if recordsToFetch > 10 {
|
|
recordsToFetch = 10
|
|
}
|
|
|
|
// Create a simple record batch
|
|
batch := make([]byte, 0, 512)
|
|
|
|
// Record batch header
|
|
baseOffsetBytes := make([]byte, 8)
|
|
binary.BigEndian.PutUint64(baseOffsetBytes, uint64(fetchOffset))
|
|
batch = append(batch, baseOffsetBytes...) // base offset (8 bytes)
|
|
|
|
// Calculate batch length (will be filled after we know the size)
|
|
batchLengthPos := len(batch)
|
|
batch = append(batch, 0, 0, 0, 0) // batch length placeholder (4 bytes)
|
|
|
|
batch = append(batch, 0, 0, 0, 0) // partition leader epoch (4 bytes)
|
|
batch = append(batch, 2) // magic byte (version 2) (1 byte)
|
|
|
|
// CRC placeholder (4 bytes) - will be calculated at the end
|
|
crcPos := len(batch)
|
|
batch = append(batch, 0, 0, 0, 0) // CRC32 placeholder
|
|
|
|
// Batch attributes (2 bytes) - no compression, no transactional
|
|
batch = append(batch, 0, 0) // attributes
|
|
|
|
// Last offset delta (4 bytes)
|
|
lastOffsetDelta := uint32(recordsToFetch - 1)
|
|
lastOffsetDeltaBytes := make([]byte, 4)
|
|
binary.BigEndian.PutUint32(lastOffsetDeltaBytes, lastOffsetDelta)
|
|
batch = append(batch, lastOffsetDeltaBytes...)
|
|
|
|
// First timestamp (8 bytes)
|
|
firstTimestamp := time.Now().UnixMilli()
|
|
firstTimestampBytes := make([]byte, 8)
|
|
binary.BigEndian.PutUint64(firstTimestampBytes, uint64(firstTimestamp))
|
|
batch = append(batch, firstTimestampBytes...)
|
|
|
|
// Max timestamp (8 bytes)
|
|
maxTimestamp := firstTimestamp + recordsToFetch - 1
|
|
maxTimestampBytes := make([]byte, 8)
|
|
binary.BigEndian.PutUint64(maxTimestampBytes, uint64(maxTimestamp))
|
|
batch = append(batch, maxTimestampBytes...)
|
|
|
|
// Producer ID (8 bytes) - -1 for non-transactional
|
|
batch = append(batch, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF)
|
|
|
|
// Producer Epoch (2 bytes) - -1 for non-transactional
|
|
batch = append(batch, 0xFF, 0xFF)
|
|
|
|
// Base Sequence (4 bytes) - -1 for non-transactional
|
|
batch = append(batch, 0xFF, 0xFF, 0xFF, 0xFF)
|
|
|
|
// Record count (4 bytes)
|
|
recordCountBytes := make([]byte, 4)
|
|
binary.BigEndian.PutUint32(recordCountBytes, uint32(recordsToFetch))
|
|
batch = append(batch, recordCountBytes...)
|
|
|
|
// Add individual records
|
|
for i := int64(0); i < recordsToFetch; i++ {
|
|
// Create test message
|
|
key := []byte(fmt.Sprintf("key-%d", fetchOffset+i))
|
|
value := []byte(fmt.Sprintf("Test message %d", fetchOffset+i))
|
|
|
|
// Build individual record
|
|
record := make([]byte, 0, 128)
|
|
|
|
// Record attributes (1 byte)
|
|
record = append(record, 0)
|
|
|
|
// Timestamp delta (varint)
|
|
timestampDelta := i
|
|
record = append(record, encodeVarint(timestampDelta)...)
|
|
|
|
// Offset delta (varint)
|
|
offsetDelta := i
|
|
record = append(record, encodeVarint(offsetDelta)...)
|
|
|
|
// Key length and key (varint + data)
|
|
record = append(record, encodeVarint(int64(len(key)))...)
|
|
record = append(record, key...)
|
|
|
|
// Value length and value (varint + data)
|
|
record = append(record, encodeVarint(int64(len(value)))...)
|
|
record = append(record, value...)
|
|
|
|
// Headers count (varint) - 0 headers
|
|
record = append(record, encodeVarint(0)...)
|
|
|
|
// Prepend record length (varint)
|
|
recordLength := int64(len(record))
|
|
batch = append(batch, encodeVarint(recordLength)...)
|
|
batch = append(batch, record...)
|
|
}
|
|
|
|
// Fill in the batch length
|
|
batchLength := uint32(len(batch) - batchLengthPos - 4)
|
|
binary.BigEndian.PutUint32(batch[batchLengthPos:batchLengthPos+4], batchLength)
|
|
|
|
// Calculate CRC32 for the batch
|
|
// CRC is calculated over: attributes + last_offset_delta + first_timestamp + max_timestamp + producer_id + producer_epoch + base_sequence + records_count + records
|
|
// This starts after the CRC field (which comes after magic byte)
|
|
crcStartPos := crcPos + 4 // start after the CRC field
|
|
crcData := batch[crcStartPos:]
|
|
crc := crc32.ChecksumIEEE(crcData)
|
|
binary.BigEndian.PutUint32(batch[crcPos:crcPos+4], crc)
|
|
|
|
return batch
|
|
}
|
|
|
|
// constructRecordBatch creates a realistic Kafka record batch that matches produced messages
|
|
// This creates record batches that mirror what was actually stored during Produce operations
|
|
func (h *Handler) constructRecordBatch(ledger interface{}, fetchOffset, highWaterMark int64) []byte {
|
|
recordsToFetch := highWaterMark - fetchOffset
|
|
if recordsToFetch <= 0 {
|
|
return []byte{} // no records to fetch
|
|
}
|
|
|
|
// Limit the number of records for testing
|
|
if recordsToFetch > 10 {
|
|
recordsToFetch = 10
|
|
}
|
|
|
|
// Create a realistic record batch that matches what clients expect
|
|
// This simulates the same format that would be stored during Produce operations
|
|
batch := make([]byte, 0, 512)
|
|
|
|
// Record batch header (61 bytes total)
|
|
baseOffsetBytes := make([]byte, 8)
|
|
binary.BigEndian.PutUint64(baseOffsetBytes, uint64(fetchOffset))
|
|
batch = append(batch, baseOffsetBytes...) // base offset (8 bytes)
|
|
|
|
// Calculate batch length (will be filled after we know the size)
|
|
batchLengthPos := len(batch)
|
|
batch = append(batch, 0, 0, 0, 0) // batch length placeholder (4 bytes)
|
|
|
|
batch = append(batch, 0, 0, 0, 0) // partition leader epoch (4 bytes)
|
|
batch = append(batch, 2) // magic byte (version 2) (1 byte)
|
|
|
|
// CRC placeholder (4 bytes) - for testing, use 0
|
|
batch = append(batch, 0, 0, 0, 0) // CRC32
|
|
|
|
// Batch attributes (2 bytes) - no compression, no transactional
|
|
batch = append(batch, 0, 0) // attributes
|
|
|
|
// Last offset delta (4 bytes)
|
|
lastOffsetDelta := uint32(recordsToFetch - 1)
|
|
lastOffsetDeltaBytes := make([]byte, 4)
|
|
binary.BigEndian.PutUint32(lastOffsetDeltaBytes, lastOffsetDelta)
|
|
batch = append(batch, lastOffsetDeltaBytes...)
|
|
|
|
// First timestamp (8 bytes)
|
|
firstTimestamp := time.Now().UnixMilli() // Use milliseconds like Kafka
|
|
firstTimestampBytes := make([]byte, 8)
|
|
binary.BigEndian.PutUint64(firstTimestampBytes, uint64(firstTimestamp))
|
|
batch = append(batch, firstTimestampBytes...)
|
|
|
|
// Max timestamp (8 bytes)
|
|
maxTimestamp := firstTimestamp + recordsToFetch - 1 // 1ms per record
|
|
maxTimestampBytes := make([]byte, 8)
|
|
binary.BigEndian.PutUint64(maxTimestampBytes, uint64(maxTimestamp))
|
|
batch = append(batch, maxTimestampBytes...)
|
|
|
|
// Producer ID (8 bytes) - -1 for non-transactional
|
|
batch = append(batch, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF)
|
|
|
|
// Producer Epoch (2 bytes) - -1 for non-transactional
|
|
batch = append(batch, 0xFF, 0xFF)
|
|
|
|
// Base Sequence (4 bytes) - -1 for non-transactional
|
|
batch = append(batch, 0xFF, 0xFF, 0xFF, 0xFF)
|
|
|
|
// Record count (4 bytes)
|
|
recordCountBytes := make([]byte, 4)
|
|
binary.BigEndian.PutUint32(recordCountBytes, uint32(recordsToFetch))
|
|
batch = append(batch, recordCountBytes...)
|
|
|
|
// Add records that match typical client expectations
|
|
for i := int64(0); i < recordsToFetch; i++ {
|
|
// Build individual record
|
|
record := make([]byte, 0, 64)
|
|
|
|
// Record attributes (1 byte)
|
|
record = append(record, 0)
|
|
|
|
// Timestamp delta (varint) - use proper varint encoding
|
|
timestampDelta := i // milliseconds from first timestamp
|
|
record = append(record, encodeVarint(timestampDelta)...)
|
|
|
|
// Offset delta (varint)
|
|
offsetDelta := i
|
|
record = append(record, encodeVarint(offsetDelta)...)
|
|
|
|
// Key length (varint) - -1 for null key
|
|
record = append(record, encodeVarint(-1)...)
|
|
|
|
// Value length and value
|
|
value := fmt.Sprintf("Test message %d", fetchOffset+i)
|
|
record = append(record, encodeVarint(int64(len(value)))...)
|
|
record = append(record, []byte(value)...)
|
|
|
|
// Headers count (varint) - 0 headers
|
|
record = append(record, encodeVarint(0)...)
|
|
|
|
// Prepend record length (varint)
|
|
recordLength := int64(len(record))
|
|
batch = append(batch, encodeVarint(recordLength)...)
|
|
batch = append(batch, record...)
|
|
}
|
|
|
|
// Fill in the batch length
|
|
batchLength := uint32(len(batch) - batchLengthPos - 4)
|
|
binary.BigEndian.PutUint32(batch[batchLengthPos:batchLengthPos+4], batchLength)
|
|
|
|
return batch
|
|
}
|
|
|
|
// encodeVarint encodes a signed integer using Kafka's varint encoding
|
|
func encodeVarint(value int64) []byte {
|
|
// Kafka uses zigzag encoding for signed integers
|
|
zigzag := uint64((value << 1) ^ (value >> 63))
|
|
|
|
var buf []byte
|
|
for zigzag >= 0x80 {
|
|
buf = append(buf, byte(zigzag)|0x80)
|
|
zigzag >>= 7
|
|
}
|
|
buf = append(buf, byte(zigzag))
|
|
return buf
|
|
}
|
|
|
|
// getMultipleRecordBatches retrieves and combines multiple record batches starting from the given offset
|
|
func (h *Handler) getMultipleRecordBatches(topicName string, partitionID int32, startOffset, highWaterMark int64) []byte {
|
|
var combinedBatch []byte
|
|
|
|
// Try to get all available record batches from startOffset to highWaterMark-1
|
|
for offset := startOffset; offset < highWaterMark; offset++ {
|
|
if batch, exists := h.GetRecordBatch(topicName, partitionID, offset); exists {
|
|
// For the first batch, include the full record batch
|
|
if len(combinedBatch) == 0 {
|
|
combinedBatch = append(combinedBatch, batch...)
|
|
} else {
|
|
// For subsequent batches, we need to append them properly
|
|
// For now, just return the first batch to avoid format issues
|
|
// TODO: Implement proper record batch concatenation
|
|
break
|
|
}
|
|
}
|
|
}
|
|
|
|
return combinedBatch
|
|
}
|
|
|
|
// reconstructSchematizedMessage reconstructs a schematized message from SMQ RecordValue
|
|
func (h *Handler) reconstructSchematizedMessage(recordValue *schema_pb.RecordValue, metadata map[string]string) ([]byte, error) {
|
|
// Only reconstruct if schema management is enabled
|
|
if !h.IsSchemaEnabled() {
|
|
return nil, fmt.Errorf("schema management not enabled")
|
|
}
|
|
|
|
// Extract schema information from metadata
|
|
schemaIDStr, exists := metadata["schema_id"]
|
|
if !exists {
|
|
return nil, fmt.Errorf("no schema ID in metadata")
|
|
}
|
|
|
|
var schemaID uint32
|
|
if _, err := fmt.Sscanf(schemaIDStr, "%d", &schemaID); err != nil {
|
|
return nil, fmt.Errorf("invalid schema ID: %w", err)
|
|
}
|
|
|
|
formatStr, exists := metadata["schema_format"]
|
|
if !exists {
|
|
return nil, fmt.Errorf("no schema format in metadata")
|
|
}
|
|
|
|
var format schema.Format
|
|
switch formatStr {
|
|
case "AVRO":
|
|
format = schema.FormatAvro
|
|
case "PROTOBUF":
|
|
format = schema.FormatProtobuf
|
|
case "JSON_SCHEMA":
|
|
format = schema.FormatJSONSchema
|
|
default:
|
|
return nil, fmt.Errorf("unsupported schema format: %s", formatStr)
|
|
}
|
|
|
|
// Use schema manager to encode back to original format
|
|
return h.schemaManager.EncodeMessage(recordValue, schemaID, format)
|
|
}
|
|
|
|
// fetchSchematizedRecords fetches and reconstructs schematized records from SeaweedMQ
|
|
func (h *Handler) fetchSchematizedRecords(topicName string, partitionID int32, offset int64, maxBytes int32) ([][]byte, error) {
|
|
// This is a placeholder for Phase 7
|
|
// In Phase 8, this will integrate with SeaweedMQ to:
|
|
// 1. Fetch stored RecordValues and metadata from SeaweedMQ
|
|
// 2. Reconstruct original Kafka message format using schema information
|
|
// 3. Handle schema evolution and compatibility
|
|
// 4. Return properly formatted Kafka record batches
|
|
|
|
fmt.Printf("DEBUG: Would fetch schematized records - topic: %s, partition: %d, offset: %d, maxBytes: %d\n",
|
|
topicName, partitionID, offset, maxBytes)
|
|
|
|
// For Phase 7, return empty records
|
|
// In Phase 8, this will return actual reconstructed messages
|
|
return [][]byte{}, nil
|
|
}
|
|
|
|
// createSchematizedRecordBatch creates a Kafka record batch from reconstructed schematized messages
|
|
func (h *Handler) createSchematizedRecordBatch(messages [][]byte, baseOffset int64) []byte {
|
|
if len(messages) == 0 {
|
|
// Return empty record batch
|
|
return h.createEmptyRecordBatch(baseOffset)
|
|
}
|
|
|
|
// For Phase 7, create a simple record batch
|
|
// In Phase 8, this will properly format multiple messages into a record batch
|
|
// with correct headers, compression, and CRC validation
|
|
|
|
// Combine all messages into a single batch payload
|
|
var batchPayload []byte
|
|
for _, msg := range messages {
|
|
// Add message length prefix (for record batch format)
|
|
msgLen := len(msg)
|
|
lengthBytes := make([]byte, 4)
|
|
binary.BigEndian.PutUint32(lengthBytes, uint32(msgLen))
|
|
batchPayload = append(batchPayload, lengthBytes...)
|
|
batchPayload = append(batchPayload, msg...)
|
|
}
|
|
|
|
return h.createRecordBatchWithPayload(baseOffset, int32(len(messages)), batchPayload)
|
|
}
|
|
|
|
// createEmptyRecordBatch creates an empty Kafka record batch using the new parser
|
|
func (h *Handler) createEmptyRecordBatch(baseOffset int64) []byte {
|
|
// Use the new record batch creation function with no compression
|
|
emptyRecords := []byte{}
|
|
batch, err := CreateRecordBatch(baseOffset, emptyRecords, compression.None)
|
|
if err != nil {
|
|
// Fallback to manual creation if there's an error
|
|
return h.createEmptyRecordBatchManual(baseOffset)
|
|
}
|
|
return batch
|
|
}
|
|
|
|
// createEmptyRecordBatchManual creates an empty Kafka record batch manually (fallback)
|
|
func (h *Handler) createEmptyRecordBatchManual(baseOffset int64) []byte {
|
|
// Create a minimal empty record batch
|
|
batch := make([]byte, 0, 61) // Standard record batch header size
|
|
|
|
// Base offset (8 bytes)
|
|
baseOffsetBytes := make([]byte, 8)
|
|
binary.BigEndian.PutUint64(baseOffsetBytes, uint64(baseOffset))
|
|
batch = append(batch, baseOffsetBytes...)
|
|
|
|
// Batch length (4 bytes) - will be filled at the end
|
|
lengthPlaceholder := len(batch)
|
|
batch = append(batch, 0, 0, 0, 0)
|
|
|
|
// Partition leader epoch (4 bytes) - 0 for simplicity
|
|
batch = append(batch, 0, 0, 0, 0)
|
|
|
|
// Magic byte (1 byte) - version 2
|
|
batch = append(batch, 2)
|
|
|
|
// CRC32 (4 bytes) - placeholder, should be calculated
|
|
batch = append(batch, 0, 0, 0, 0)
|
|
|
|
// Attributes (2 bytes) - no compression, no transactional
|
|
batch = append(batch, 0, 0)
|
|
|
|
// Last offset delta (4 bytes) - 0 for empty batch
|
|
batch = append(batch, 0xFF, 0xFF, 0xFF, 0xFF)
|
|
|
|
// First timestamp (8 bytes) - current time
|
|
timestamp := time.Now().UnixMilli()
|
|
timestampBytes := make([]byte, 8)
|
|
binary.BigEndian.PutUint64(timestampBytes, uint64(timestamp))
|
|
batch = append(batch, timestampBytes...)
|
|
|
|
// Max timestamp (8 bytes) - same as first for empty batch
|
|
batch = append(batch, timestampBytes...)
|
|
|
|
// Producer ID (8 bytes) - -1 for non-transactional
|
|
batch = append(batch, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF)
|
|
|
|
// Producer Epoch (2 bytes) - -1 for non-transactional
|
|
batch = append(batch, 0xFF, 0xFF)
|
|
|
|
// Base Sequence (4 bytes) - -1 for non-transactional
|
|
batch = append(batch, 0xFF, 0xFF, 0xFF, 0xFF)
|
|
|
|
// Record count (4 bytes) - 0 for empty batch
|
|
batch = append(batch, 0, 0, 0, 0)
|
|
|
|
// Fill in the batch length
|
|
batchLength := len(batch) - 12 // Exclude base offset and length field itself
|
|
binary.BigEndian.PutUint32(batch[lengthPlaceholder:lengthPlaceholder+4], uint32(batchLength))
|
|
|
|
return batch
|
|
}
|
|
|
|
// createRecordBatchWithPayload creates a record batch with the given payload
|
|
func (h *Handler) createRecordBatchWithPayload(baseOffset int64, recordCount int32, payload []byte) []byte {
|
|
// For Phase 7, create a simplified record batch
|
|
// In Phase 8, this will implement proper Kafka record batch format v2
|
|
|
|
batch := h.createEmptyRecordBatch(baseOffset)
|
|
|
|
// Update record count
|
|
recordCountOffset := len(batch) - 4
|
|
binary.BigEndian.PutUint32(batch[recordCountOffset:recordCountOffset+4], uint32(recordCount))
|
|
|
|
// Append payload (simplified - real implementation would format individual records)
|
|
batch = append(batch, payload...)
|
|
|
|
// Update batch length
|
|
batchLength := len(batch) - 12
|
|
binary.BigEndian.PutUint32(batch[8:12], uint32(batchLength))
|
|
|
|
return batch
|
|
}
|
|
|
|
// handleSchematizedFetch handles fetch requests for topics with schematized messages
|
|
func (h *Handler) handleSchematizedFetch(topicName string, partitionID int32, offset int64, maxBytes int32) ([]byte, error) {
|
|
// Check if this topic uses schema management
|
|
if !h.IsSchemaEnabled() {
|
|
// Fall back to regular fetch handling
|
|
return nil, fmt.Errorf("schema management not enabled")
|
|
}
|
|
|
|
// Fetch schematized records from SeaweedMQ
|
|
messages, err := h.fetchSchematizedRecords(topicName, partitionID, offset, maxBytes)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to fetch schematized records: %w", err)
|
|
}
|
|
|
|
// Create record batch from reconstructed messages
|
|
recordBatch := h.createSchematizedRecordBatch(messages, offset)
|
|
|
|
fmt.Printf("DEBUG: Created schematized record batch: %d bytes for %d messages\n",
|
|
len(recordBatch), len(messages))
|
|
|
|
return recordBatch, nil
|
|
}
|
|
|
|
// isSchematizedTopic checks if a topic uses schema management
|
|
func (h *Handler) isSchematizedTopic(topicName string) bool {
|
|
// For Phase 7, we'll implement a simple check
|
|
// In Phase 8, this will check SeaweedMQ metadata or configuration
|
|
// to determine if a topic has schematized messages
|
|
|
|
// For now, assume topics ending with "-value" or "-key" are schematized
|
|
// This is a common Confluent Schema Registry convention
|
|
if len(topicName) > 6 {
|
|
suffix := topicName[len(topicName)-6:]
|
|
if suffix == "-value" {
|
|
return true
|
|
}
|
|
}
|
|
if len(topicName) > 4 {
|
|
suffix := topicName[len(topicName)-4:]
|
|
if suffix == "-key" {
|
|
return true
|
|
}
|
|
}
|
|
|
|
return false
|
|
}
|
|
|
|
// getSchemaMetadataForTopic retrieves schema metadata for a topic
|
|
func (h *Handler) getSchemaMetadataForTopic(topicName string) (map[string]string, error) {
|
|
// This is a placeholder for Phase 7
|
|
// In Phase 8, this will retrieve actual schema metadata from SeaweedMQ
|
|
// including schema ID, format, subject, version, etc.
|
|
|
|
if !h.IsSchemaEnabled() {
|
|
return nil, fmt.Errorf("schema management not enabled")
|
|
}
|
|
|
|
// For Phase 7, return mock metadata
|
|
metadata := map[string]string{
|
|
"schema_id": "1",
|
|
"schema_format": "AVRO",
|
|
"schema_subject": topicName,
|
|
"schema_version": "1",
|
|
}
|
|
|
|
fmt.Printf("DEBUG: Retrieved schema metadata for topic %s: %v\n", topicName, metadata)
|
|
return metadata, nil
|
|
}
|