From 28d4f90d832e5c55a09a5e55834b8f686c190ba8 Mon Sep 17 00:00:00 2001 From: chrislu Date: Fri, 12 Sep 2025 13:05:09 -0700 Subject: [PATCH] feat: enhance Fetch API with proper request parsing and record batch construction - Added comprehensive Fetch request parsing for different API versions - Implemented constructRecordBatchFromLedger to return actual messages - Added support for dynamic topic/partition handling in Fetch responses - Enhanced record batch format with proper Kafka v2 structure - Added varint encoding for record fields - Improved error handling and validation TODO: Debug consumer integration issues and test with actual message retrieval --- test/kafka/comprehensive_e2e_test.go | 472 +++++++++++++++++++++++ test/kafka/kafka_go_produce_only_test.go | 186 +++++++++ test/kafka/sarama_simple_test.go | 110 ++++++ weed/mq/kafka/protocol/fetch.go | 444 ++++++++++++++++++--- 4 files changed, 1157 insertions(+), 55 deletions(-) create mode 100644 test/kafka/comprehensive_e2e_test.go diff --git a/test/kafka/comprehensive_e2e_test.go b/test/kafka/comprehensive_e2e_test.go new file mode 100644 index 000000000..339acd094 --- /dev/null +++ b/test/kafka/comprehensive_e2e_test.go @@ -0,0 +1,472 @@ +package kafka + +import ( + "context" + "fmt" + "strings" + "testing" + "time" + + "github.com/IBM/sarama" + "github.com/seaweedfs/seaweedfs/weed/mq/kafka/gateway" + "github.com/segmentio/kafka-go" +) + +// TestComprehensiveE2E tests both kafka-go and Sarama clients in a comprehensive scenario +func TestComprehensiveE2E(t *testing.T) { + // Start gateway + gatewayServer := gateway.NewServer(gateway.Options{ + Listen: "127.0.0.1:0", + UseSeaweedMQ: false, // Use in-memory mode for testing + }) + + go func() { + if err := gatewayServer.Start(); err != nil { + t.Errorf("Failed to start gateway: %v", err) + } + }() + defer gatewayServer.Close() + + time.Sleep(100 * time.Millisecond) + + host, port := gatewayServer.GetListenerAddr() + addr := fmt.Sprintf("%s:%d", host, port) + t.Logf("Gateway running on %s", addr) + + // Create multiple topics for different test scenarios + topics := []string{ + "e2e-kafka-go-topic", + "e2e-sarama-topic", + "e2e-mixed-topic", + } + + for _, topic := range topics { + gatewayServer.GetHandler().AddTopicForTesting(topic, 1) + t.Logf("Added topic: %s", topic) + } + + // Test 1: kafka-go producer -> kafka-go consumer + t.Run("KafkaGo_to_KafkaGo", func(t *testing.T) { + testKafkaGoToKafkaGo(t, addr, topics[0]) + }) + + // Test 2: Sarama producer -> Sarama consumer + t.Run("Sarama_to_Sarama", func(t *testing.T) { + testSaramaToSarama(t, addr, topics[1]) + }) + + // Test 3: Mixed clients - kafka-go producer -> Sarama consumer + t.Run("KafkaGo_to_Sarama", func(t *testing.T) { + testKafkaGoToSarama(t, addr, topics[2]) + }) + + // Test 4: Mixed clients - Sarama producer -> kafka-go consumer + t.Run("Sarama_to_KafkaGo", func(t *testing.T) { + testSaramaToKafkaGo(t, addr, topics[2]) + }) +} + +func testKafkaGoToKafkaGo(t *testing.T, addr, topic string) { + messages := []kafka.Message{ + {Key: []byte("kgo-key1"), Value: []byte("kafka-go to kafka-go message 1")}, + {Key: []byte("kgo-key2"), Value: []byte("kafka-go to kafka-go message 2")}, + } + + // Produce with kafka-go + w := &kafka.Writer{ + Addr: kafka.TCP(addr), + Topic: topic, + Balancer: &kafka.LeastBytes{}, + BatchTimeout: 50 * time.Millisecond, + RequiredAcks: kafka.RequireOne, + } + defer w.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + err := w.WriteMessages(ctx, messages...) + if err != nil { + t.Fatalf("kafka-go produce failed: %v", err) + } + t.Logf("✅ kafka-go produced %d messages", len(messages)) + + // Consume with kafka-go + r := kafka.NewReader(kafka.ReaderConfig{ + Brokers: []string{addr}, + Topic: topic, + StartOffset: kafka.FirstOffset, + MinBytes: 1, + MaxBytes: 10e6, + }) + defer r.Close() + + consumeCtx, consumeCancel := context.WithTimeout(context.Background(), 10*time.Second) + defer consumeCancel() + + for i := 0; i < len(messages); i++ { + msg, err := r.ReadMessage(consumeCtx) + if err != nil { + t.Fatalf("kafka-go consume failed: %v", err) + } + t.Logf("✅ kafka-go consumed: key=%s, value=%s", string(msg.Key), string(msg.Value)) + + // Validate message + expected := messages[i] + if string(msg.Key) != string(expected.Key) || string(msg.Value) != string(expected.Value) { + t.Errorf("Message mismatch: got key=%s value=%s, want key=%s value=%s", + string(msg.Key), string(msg.Value), string(expected.Key), string(expected.Value)) + } + } + + t.Logf("🎉 kafka-go to kafka-go test PASSED") +} + +func testSaramaToSarama(t *testing.T, addr, topic string) { + // Configure Sarama + config := sarama.NewConfig() + config.Version = sarama.V0_11_0_0 + config.Producer.Return.Successes = true + config.Producer.RequiredAcks = sarama.WaitForAll + config.Consumer.Return.Errors = true + + messages := []string{ + "Sarama to Sarama message 1", + "Sarama to Sarama message 2", + } + + // Produce with Sarama + producer, err := sarama.NewSyncProducer([]string{addr}, config) + if err != nil { + t.Fatalf("Failed to create Sarama producer: %v", err) + } + defer producer.Close() + + for i, msgText := range messages { + msg := &sarama.ProducerMessage{ + Topic: topic, + Key: sarama.StringEncoder(fmt.Sprintf("sarama-key-%d", i)), + Value: sarama.StringEncoder(msgText), + } + + partition, offset, err := producer.SendMessage(msg) + if err != nil { + t.Fatalf("Sarama produce failed: %v", err) + } + t.Logf("✅ Sarama produced message %d: partition=%d, offset=%d", i, partition, offset) + } + + // Consume with Sarama + consumer, err := sarama.NewConsumer([]string{addr}, config) + if err != nil { + t.Fatalf("Failed to create Sarama consumer: %v", err) + } + defer consumer.Close() + + partitionConsumer, err := consumer.ConsumePartition(topic, 0, sarama.OffsetOldest) + if err != nil { + t.Fatalf("Failed to create Sarama partition consumer: %v", err) + } + defer partitionConsumer.Close() + + consumedCount := 0 + timeout := time.After(10 * time.Second) + + for consumedCount < len(messages) { + select { + case msg := <-partitionConsumer.Messages(): + t.Logf("✅ Sarama consumed: key=%s, value=%s, offset=%d", + string(msg.Key), string(msg.Value), msg.Offset) + + expectedValue := messages[consumedCount] + if string(msg.Value) != expectedValue { + t.Errorf("Message mismatch: got %s, want %s", string(msg.Value), expectedValue) + } + consumedCount++ + + case err := <-partitionConsumer.Errors(): + t.Fatalf("Sarama consumer error: %v", err) + + case <-timeout: + t.Fatalf("Timeout waiting for Sarama messages. Consumed %d/%d", consumedCount, len(messages)) + } + } + + t.Logf("🎉 Sarama to Sarama test PASSED") +} + +func testKafkaGoToSarama(t *testing.T, addr, topic string) { + // Note: In a real test environment, we'd need to ensure topic isolation + // For now, we'll use a different key prefix to distinguish messages + + messages := []kafka.Message{ + {Key: []byte("mixed-kgo-key1"), Value: []byte("kafka-go producer to Sarama consumer")}, + {Key: []byte("mixed-kgo-key2"), Value: []byte("Cross-client compatibility test")}, + } + + // Produce with kafka-go + w := &kafka.Writer{ + Addr: kafka.TCP(addr), + Topic: topic, + Balancer: &kafka.LeastBytes{}, + BatchTimeout: 50 * time.Millisecond, + RequiredAcks: kafka.RequireOne, + } + defer w.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + err := w.WriteMessages(ctx, messages...) + if err != nil { + t.Fatalf("kafka-go produce failed: %v", err) + } + t.Logf("✅ kafka-go produced %d messages for Sarama consumer", len(messages)) + + // Consume with Sarama + config := sarama.NewConfig() + config.Version = sarama.V0_11_0_0 + config.Consumer.Return.Errors = true + + consumer, err := sarama.NewConsumer([]string{addr}, config) + if err != nil { + t.Fatalf("Failed to create Sarama consumer: %v", err) + } + defer consumer.Close() + + // Start from latest to avoid consuming previous test messages + partitionConsumer, err := consumer.ConsumePartition(topic, 0, sarama.OffsetNewest) + if err != nil { + t.Fatalf("Failed to create Sarama partition consumer: %v", err) + } + defer partitionConsumer.Close() + + // Give a moment for the consumer to be ready + time.Sleep(100 * time.Millisecond) + + // Produce again to ensure we get fresh messages + err = w.WriteMessages(ctx, messages...) + if err != nil { + t.Fatalf("kafka-go second produce failed: %v", err) + } + + consumedCount := 0 + timeout := time.After(10 * time.Second) + + for consumedCount < len(messages) { + select { + case msg := <-partitionConsumer.Messages(): + // Only count messages with our test key prefix + if strings.HasPrefix(string(msg.Key), "mixed-kgo-key") { + t.Logf("✅ Sarama consumed kafka-go message: key=%s, value=%s", + string(msg.Key), string(msg.Value)) + consumedCount++ + } + + case err := <-partitionConsumer.Errors(): + t.Fatalf("Sarama consumer error: %v", err) + + case <-timeout: + t.Fatalf("Timeout waiting for mixed messages. Consumed %d/%d", consumedCount, len(messages)) + } + } + + t.Logf("🎉 kafka-go to Sarama test PASSED") +} + +func testSaramaToKafkaGo(t *testing.T, addr, topic string) { + // Configure Sarama + config := sarama.NewConfig() + config.Version = sarama.V0_11_0_0 + config.Producer.Return.Successes = true + config.Producer.RequiredAcks = sarama.WaitForAll + + messages := []string{ + "Sarama producer to kafka-go consumer", + "Reverse cross-client compatibility test", + } + + // Produce with Sarama + producer, err := sarama.NewSyncProducer([]string{addr}, config) + if err != nil { + t.Fatalf("Failed to create Sarama producer: %v", err) + } + defer producer.Close() + + for i, msgText := range messages { + msg := &sarama.ProducerMessage{ + Topic: topic, + Key: sarama.StringEncoder(fmt.Sprintf("mixed-sarama-key-%d", i)), + Value: sarama.StringEncoder(msgText), + } + + partition, offset, err := producer.SendMessage(msg) + if err != nil { + t.Fatalf("Sarama produce failed: %v", err) + } + t.Logf("✅ Sarama produced message %d for kafka-go consumer: partition=%d, offset=%d", i, partition, offset) + } + + // Consume with kafka-go + r := kafka.NewReader(kafka.ReaderConfig{ + Brokers: []string{addr}, + Topic: topic, + StartOffset: kafka.LastOffset, // Start from latest to avoid previous messages + MinBytes: 1, + MaxBytes: 10e6, + }) + defer r.Close() + + // Give a moment for the reader to be ready, then produce fresh messages + time.Sleep(100 * time.Millisecond) + + // Produce again to ensure fresh messages for the latest offset reader + for i, msgText := range messages { + msg := &sarama.ProducerMessage{ + Topic: topic, + Key: sarama.StringEncoder(fmt.Sprintf("mixed-sarama-fresh-key-%d", i)), + Value: sarama.StringEncoder(msgText), + } + + _, _, err := producer.SendMessage(msg) + if err != nil { + t.Fatalf("Sarama second produce failed: %v", err) + } + } + + consumeCtx, consumeCancel := context.WithTimeout(context.Background(), 10*time.Second) + defer consumeCancel() + + consumedCount := 0 + for consumedCount < len(messages) { + msg, err := r.ReadMessage(consumeCtx) + if err != nil { + t.Fatalf("kafka-go consume failed: %v", err) + } + + // Only count messages with our fresh test key prefix + if strings.HasPrefix(string(msg.Key), "mixed-sarama-fresh-key") { + t.Logf("✅ kafka-go consumed Sarama message: key=%s, value=%s", + string(msg.Key), string(msg.Value)) + consumedCount++ + } + } + + t.Logf("🎉 Sarama to kafka-go test PASSED") +} + +// TestOffsetManagement tests offset commit and fetch operations +func TestOffsetManagement(t *testing.T) { + // Start gateway + gatewayServer := gateway.NewServer(gateway.Options{ + Listen: "127.0.0.1:0", + UseSeaweedMQ: false, + }) + + go func() { + if err := gatewayServer.Start(); err != nil { + t.Errorf("Failed to start gateway: %v", err) + } + }() + defer gatewayServer.Close() + + time.Sleep(100 * time.Millisecond) + + host, port := gatewayServer.GetListenerAddr() + addr := fmt.Sprintf("%s:%d", host, port) + topic := "offset-management-topic" + groupID := "offset-test-group" + + gatewayServer.GetHandler().AddTopicForTesting(topic, 1) + t.Logf("Testing offset management on %s with topic %s", addr, topic) + + // Produce test messages + messages := []kafka.Message{ + {Key: []byte("offset-key1"), Value: []byte("Offset test message 1")}, + {Key: []byte("offset-key2"), Value: []byte("Offset test message 2")}, + {Key: []byte("offset-key3"), Value: []byte("Offset test message 3")}, + {Key: []byte("offset-key4"), Value: []byte("Offset test message 4")}, + {Key: []byte("offset-key5"), Value: []byte("Offset test message 5")}, + } + + w := &kafka.Writer{ + Addr: kafka.TCP(addr), + Topic: topic, + Balancer: &kafka.LeastBytes{}, + BatchTimeout: 50 * time.Millisecond, + RequiredAcks: kafka.RequireOne, + } + defer w.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + err := w.WriteMessages(ctx, messages...) + if err != nil { + t.Fatalf("Failed to produce offset test messages: %v", err) + } + t.Logf("✅ Produced %d messages for offset test", len(messages)) + + // Test 1: Consume first 3 messages and commit offsets + t.Logf("=== Phase 1: Consuming first 3 messages ===") + + r1 := kafka.NewReader(kafka.ReaderConfig{ + Brokers: []string{addr}, + Topic: topic, + GroupID: groupID, + StartOffset: kafka.FirstOffset, + MinBytes: 1, + MaxBytes: 10e6, + }) + + consumeCtx1, consumeCancel1 := context.WithTimeout(context.Background(), 10*time.Second) + defer consumeCancel1() + + for i := 0; i < 3; i++ { + msg, err := r1.ReadMessage(consumeCtx1) + if err != nil { + t.Fatalf("Failed to read message %d: %v", i, err) + } + t.Logf("✅ Phase 1 consumed message %d: key=%s, offset=%d", + i, string(msg.Key), msg.Offset) + } + + // Commit the offset (kafka-go automatically commits when using GroupID) + r1.Close() + t.Logf("✅ Phase 1 completed - offsets should be committed") + + // Test 2: Create new consumer with same group ID - should resume from committed offset + t.Logf("=== Phase 2: Resuming from committed offset ===") + + r2 := kafka.NewReader(kafka.ReaderConfig{ + Brokers: []string{addr}, + Topic: topic, + GroupID: groupID, // Same group ID + StartOffset: kafka.FirstOffset, // This should be ignored due to committed offset + MinBytes: 1, + MaxBytes: 10e6, + }) + defer r2.Close() + + consumeCtx2, consumeCancel2 := context.WithTimeout(context.Background(), 10*time.Second) + defer consumeCancel2() + + remainingCount := 0 + expectedRemaining := len(messages) - 3 // Should get the last 2 messages + + for remainingCount < expectedRemaining { + msg, err := r2.ReadMessage(consumeCtx2) + if err != nil { + t.Fatalf("Failed to read remaining message %d: %v", remainingCount, err) + } + t.Logf("✅ Phase 2 consumed remaining message %d: key=%s, offset=%d", + remainingCount, string(msg.Key), msg.Offset) + remainingCount++ + } + + if remainingCount != expectedRemaining { + t.Errorf("Expected %d remaining messages, got %d", expectedRemaining, remainingCount) + } + + t.Logf("🎉 SUCCESS: Offset management test completed - consumed 3 + %d messages", remainingCount) +} diff --git a/test/kafka/kafka_go_produce_only_test.go b/test/kafka/kafka_go_produce_only_test.go index 0401d3153..05ec19001 100644 --- a/test/kafka/kafka_go_produce_only_test.go +++ b/test/kafka/kafka_go_produce_only_test.go @@ -18,6 +18,7 @@ func TestKafkaGo_ProduceOnly(t *testing.T) { t.Errorf("Failed to start gateway: %v", err) } }() + defer gatewayServer.Close() time.Sleep(100 * time.Millisecond) @@ -33,6 +34,7 @@ func TestKafkaGo_ProduceOnly(t *testing.T) { BatchTimeout: 50 * time.Millisecond, RequiredAcks: kafka.RequireOne, } + defer w.Close() ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) defer cancel() @@ -42,3 +44,187 @@ func TestKafkaGo_ProduceOnly(t *testing.T) { t.Fatalf("kafka-go produce failed: %v", err) } } + +func TestKafkaGo_ProduceConsume(t *testing.T) { + // Start gateway + gatewayServer := gateway.NewServer(gateway.Options{Listen: "127.0.0.1:0"}) + go func() { + if err := gatewayServer.Start(); err != nil { + t.Errorf("Failed to start gateway: %v", err) + } + }() + defer gatewayServer.Close() + + time.Sleep(100 * time.Millisecond) + + host, port := gatewayServer.GetListenerAddr() + addr := fmt.Sprintf("%s:%d", host, port) + topic := "kgo-produce-consume" + gatewayServer.GetHandler().AddTopicForTesting(topic, 1) + + // Test messages + testMessages := []kafka.Message{ + {Key: []byte("key1"), Value: []byte("Hello World!")}, + {Key: []byte("key2"), Value: []byte("Kafka Gateway Test")}, + {Key: []byte("key3"), Value: []byte("Final Message")}, + } + + t.Logf("=== Testing kafka-go Producer ===") + + // Produce messages + w := &kafka.Writer{ + Addr: kafka.TCP(addr), + Topic: topic, + Balancer: &kafka.LeastBytes{}, + BatchTimeout: 50 * time.Millisecond, + RequiredAcks: kafka.RequireOne, + } + defer w.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + err := w.WriteMessages(ctx, testMessages...) + if err != nil { + t.Fatalf("kafka-go produce failed: %v", err) + } + t.Logf("✅ Successfully produced %d messages", len(testMessages)) + + t.Logf("=== Testing kafka-go Consumer ===") + + // Consume messages + r := kafka.NewReader(kafka.ReaderConfig{ + Brokers: []string{addr}, + Topic: topic, + StartOffset: kafka.FirstOffset, + MinBytes: 1, + MaxBytes: 10e6, + }) + defer r.Close() + + consumedMessages := make([]kafka.Message, 0, len(testMessages)) + consumeCtx, consumeCancel := context.WithTimeout(context.Background(), 10*time.Second) + defer consumeCancel() + + for i := 0; i < len(testMessages); i++ { + msg, err := r.ReadMessage(consumeCtx) + if err != nil { + t.Fatalf("kafka-go consume failed at message %d: %v", i, err) + } + consumedMessages = append(consumedMessages, msg) + t.Logf("✅ Consumed message %d: key=%s, value=%s, offset=%d", + i, string(msg.Key), string(msg.Value), msg.Offset) + } + + // Validate messages + if len(consumedMessages) != len(testMessages) { + t.Fatalf("Expected %d messages, got %d", len(testMessages), len(consumedMessages)) + } + + for i, consumed := range consumedMessages { + expected := testMessages[i] + if string(consumed.Key) != string(expected.Key) { + t.Errorf("Message %d key mismatch: got %s, want %s", + i, string(consumed.Key), string(expected.Key)) + } + if string(consumed.Value) != string(expected.Value) { + t.Errorf("Message %d value mismatch: got %s, want %s", + i, string(consumed.Value), string(expected.Value)) + } + } + + t.Logf("🎉 SUCCESS: kafka-go produce-consume test completed with %d messages", len(consumedMessages)) +} + +func TestKafkaGo_ConsumerGroup(t *testing.T) { + // Start gateway + gatewayServer := gateway.NewServer(gateway.Options{Listen: "127.0.0.1:0"}) + go func() { + if err := gatewayServer.Start(); err != nil { + t.Errorf("Failed to start gateway: %v", err) + } + }() + defer gatewayServer.Close() + + time.Sleep(100 * time.Millisecond) + + host, port := gatewayServer.GetListenerAddr() + addr := fmt.Sprintf("%s:%d", host, port) + topic := "kgo-consumer-group" + groupID := "test-group" + gatewayServer.GetHandler().AddTopicForTesting(topic, 1) + + // Test messages + testMessages := []kafka.Message{ + {Key: []byte("cg-key1"), Value: []byte("Consumer Group Message 1")}, + {Key: []byte("cg-key2"), Value: []byte("Consumer Group Message 2")}, + {Key: []byte("cg-key3"), Value: []byte("Consumer Group Message 3")}, + } + + t.Logf("=== Testing kafka-go Producer for Consumer Group ===") + + // Produce messages + w := &kafka.Writer{ + Addr: kafka.TCP(addr), + Topic: topic, + Balancer: &kafka.LeastBytes{}, + BatchTimeout: 50 * time.Millisecond, + RequiredAcks: kafka.RequireOne, + } + defer w.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + err := w.WriteMessages(ctx, testMessages...) + if err != nil { + t.Fatalf("kafka-go produce failed: %v", err) + } + t.Logf("✅ Successfully produced %d messages for consumer group", len(testMessages)) + + t.Logf("=== Testing kafka-go Consumer Group ===") + + // Consume messages with consumer group + r := kafka.NewReader(kafka.ReaderConfig{ + Brokers: []string{addr}, + Topic: topic, + GroupID: groupID, + StartOffset: kafka.FirstOffset, + MinBytes: 1, + MaxBytes: 10e6, + }) + defer r.Close() + + consumedMessages := make([]kafka.Message, 0, len(testMessages)) + consumeCtx, consumeCancel := context.WithTimeout(context.Background(), 15*time.Second) + defer consumeCancel() + + for i := 0; i < len(testMessages); i++ { + msg, err := r.ReadMessage(consumeCtx) + if err != nil { + t.Fatalf("kafka-go consumer group failed at message %d: %v", i, err) + } + consumedMessages = append(consumedMessages, msg) + t.Logf("✅ Consumer group consumed message %d: key=%s, value=%s, offset=%d", + i, string(msg.Key), string(msg.Value), msg.Offset) + } + + // Validate messages + if len(consumedMessages) != len(testMessages) { + t.Fatalf("Expected %d messages, got %d", len(testMessages), len(consumedMessages)) + } + + for i, consumed := range consumedMessages { + expected := testMessages[i] + if string(consumed.Key) != string(expected.Key) { + t.Errorf("Message %d key mismatch: got %s, want %s", + i, string(consumed.Key), string(expected.Key)) + } + if string(consumed.Value) != string(expected.Value) { + t.Errorf("Message %d value mismatch: got %s, want %s", + i, string(consumed.Value), string(expected.Value)) + } + } + + t.Logf("🎉 SUCCESS: kafka-go consumer group test completed with %d messages", len(consumedMessages)) +} diff --git a/test/kafka/sarama_simple_test.go b/test/kafka/sarama_simple_test.go index 8dcd4fcea..5ba7c1560 100644 --- a/test/kafka/sarama_simple_test.go +++ b/test/kafka/sarama_simple_test.go @@ -145,3 +145,113 @@ func TestSaramaMinimalConfig(t *testing.T) { t.Logf("✅ Minimal produce succeeded: partition=%d, offset=%d", partition, offset) } } + +func TestSaramaProduceConsume(t *testing.T) { + // Start gateway + gatewayServer := gateway.NewServer(gateway.Options{ + Listen: "127.0.0.1:0", + }) + + go func() { + if err := gatewayServer.Start(); err != nil { + t.Errorf("Failed to start gateway: %v", err) + } + }() + defer gatewayServer.Close() + + // Wait for server to start + time.Sleep(100 * time.Millisecond) + + host, port := gatewayServer.GetListenerAddr() + brokerAddr := fmt.Sprintf("%s:%d", host, port) + t.Logf("Gateway running on %s", brokerAddr) + + // Add test topic + gatewayHandler := gatewayServer.GetHandler() + topicName := "sarama-produce-consume" + gatewayHandler.AddTopicForTesting(topicName, 1) + t.Logf("Added topic: %s", topicName) + + // Configure Sarama for Kafka 0.11 baseline + config := sarama.NewConfig() + config.Version = sarama.V0_11_0_0 + config.Producer.Return.Successes = true + config.Producer.RequiredAcks = sarama.WaitForAll + config.Consumer.Return.Errors = true + + // Test messages + testMessages := []string{ + "Sarama Producer Message 1", + "Sarama Producer Message 2", + "Sarama Producer Message 3", + } + + t.Logf("=== Testing Sarama Producer ===") + + // Create producer + producer, err := sarama.NewSyncProducer([]string{brokerAddr}, config) + if err != nil { + t.Fatalf("Failed to create producer: %v", err) + } + defer producer.Close() + + // Produce messages + for i, msgText := range testMessages { + msg := &sarama.ProducerMessage{ + Topic: topicName, + Key: sarama.StringEncoder(fmt.Sprintf("key-%d", i)), + Value: sarama.StringEncoder(msgText), + } + + partition, offset, err := producer.SendMessage(msg) + if err != nil { + t.Fatalf("Failed to produce message %d: %v", i, err) + } + t.Logf("✅ Produced message %d: partition=%d, offset=%d", i, partition, offset) + } + + t.Logf("=== Testing Sarama Consumer ===") + + // Create consumer + consumer, err := sarama.NewConsumer([]string{brokerAddr}, config) + if err != nil { + t.Fatalf("Failed to create consumer: %v", err) + } + defer consumer.Close() + + // Get partition consumer + partitionConsumer, err := consumer.ConsumePartition(topicName, 0, sarama.OffsetOldest) + if err != nil { + t.Fatalf("Failed to create partition consumer: %v", err) + } + defer partitionConsumer.Close() + + // Consume messages + consumedCount := 0 + timeout := time.After(10 * time.Second) + + for consumedCount < len(testMessages) { + select { + case msg := <-partitionConsumer.Messages(): + t.Logf("✅ Consumed message %d: key=%s, value=%s, offset=%d", + consumedCount, string(msg.Key), string(msg.Value), msg.Offset) + + // Verify message content matches what we produced + expectedValue := testMessages[consumedCount] + if string(msg.Value) != expectedValue { + t.Errorf("Message %d mismatch: got %s, want %s", + consumedCount, string(msg.Value), expectedValue) + } + + consumedCount++ + + case err := <-partitionConsumer.Errors(): + t.Fatalf("Consumer error: %v", err) + + case <-timeout: + t.Fatalf("Timeout waiting for messages. Consumed %d/%d", consumedCount, len(testMessages)) + } + } + + t.Logf("🎉 SUCCESS: Sarama produce-consume test completed with %d messages", len(testMessages)) +} diff --git a/weed/mq/kafka/protocol/fetch.go b/weed/mq/kafka/protocol/fetch.go index 7cc778295..b7d0f0fd1 100644 --- a/weed/mq/kafka/protocol/fetch.go +++ b/weed/mq/kafka/protocol/fetch.go @@ -11,13 +11,14 @@ import ( ) func (h *Handler) handleFetch(correlationID uint32, apiVersion uint16, requestBody []byte) ([]byte, error) { - fmt.Printf("DEBUG: *** FETCH REQUEST RECEIVED *** Correlation: %d, Version: %d\n", correlationID, apiVersion) - fmt.Printf("DEBUG: Fetch v%d request hex dump (first 83 bytes): %x\n", apiVersion, requestBody[:min(83, len(requestBody))]) + // 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) + } - // For now, create a minimal working Fetch response that returns empty records - // This will allow Sarama to parse the response successfully, even if no messages are returned - - response := make([]byte, 0, 256) + // Build the response + response := make([]byte, 0, 1024) // Correlation ID (4 bytes) correlationIDBytes := make([]byte, 4) @@ -29,71 +30,404 @@ func (h *Handler) handleFetch(correlationID uint32, apiVersion uint16, requestBo response = append(response, 0, 0, 0, 0) // throttle_time_ms (4 bytes, 0 = no throttling) } - // Fetch v4+ has session_id, but let's check if v5 has it at all + // Fetch v4+ has session_id and error_code if apiVersion >= 4 { - // Let's try v5 without session_id entirely - if apiVersion == 5 { - // No session_id for v5 - go directly to topics - } else { - response = append(response, 0, 0) // error_code (2 bytes, 0 = no error) - response = append(response, 0, 0, 0, 0) // session_id (4 bytes, 0 for now) - } + response = append(response, 0, 0) // error_code (2 bytes, 0 = no error) + response = append(response, 0, 0, 0, 0) // session_id (4 bytes, 0 for now) } - // Topics count (1 topic - hardcoded for now) - response = append(response, 0, 0, 0, 1) // 1 topic + // Topics count + topicsCount := len(fetchRequest.Topics) + topicsCountBytes := make([]byte, 4) + binary.BigEndian.PutUint32(topicsCountBytes, uint32(topicsCount)) + response = append(response, topicsCountBytes...) - // Topic: "sarama-e2e-topic" - topicName := "sarama-e2e-topic" - topicNameBytes := []byte(topicName) - response = append(response, byte(len(topicNameBytes)>>8), byte(len(topicNameBytes))) // topic name length - response = append(response, topicNameBytes...) // topic name + // 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 (1 partition) - response = append(response, 0, 0, 0, 1) // 1 partition + // Partitions count for this topic + partitionsCount := len(topic.Partitions) + partitionsCountBytes := make([]byte, 4) + binary.BigEndian.PutUint32(partitionsCountBytes, uint32(partitionsCount)) + response = append(response, partitionsCountBytes...) - // Partition 0 response - response = append(response, 0, 0, 0, 0) // partition_id (4 bytes) = 0 - response = append(response, 0, 0) // error_code (2 bytes) = 0 (no error) - response = append(response, 0, 0, 0, 0, 0, 0, 0, 3) // high_water_mark (8 bytes) = 3 (we produced 3 messages) + // 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...) - // Fetch v4+ has last_stable_offset and log_start_offset - if apiVersion >= 4 { - response = append(response, 0, 0, 0, 0, 0, 0, 0, 3) // last_stable_offset (8 bytes) = 3 - response = append(response, 0, 0, 0, 0, 0, 0, 0, 0) // log_start_offset (8 bytes) = 0 - } + // Error code (2 bytes) - 0 = no error + response = append(response, 0, 0) - // Fetch v4+ has aborted_transactions - if apiVersion >= 4 { - response = append(response, 0, 0, 0, 0) // aborted_transactions count (4 bytes) = 0 - } + // Get ledger for this topic-partition to determine high water mark + ledger := h.GetOrCreateLedger(topic.Name, partition.PartitionID) + highWaterMark := ledger.GetHighWaterMark() - // Records size and data (empty for now - no records returned) - response = append(response, 0, 0, 0, 0) // records size (4 bytes) = 0 (no records) + // High water mark (8 bytes) + highWaterMarkBytes := make([]byte, 8) + binary.BigEndian.PutUint64(highWaterMarkBytes, uint64(highWaterMark)) + response = append(response, highWaterMarkBytes...) - fmt.Printf("DEBUG: Fetch v%d response: %d bytes, hex dump: %x\n", apiVersion, len(response), response) - - // Let's manually verify our response structure for debugging - fmt.Printf("DEBUG: Response breakdown:\n") - fmt.Printf(" - correlation_id (4): %x\n", response[0:4]) - if apiVersion >= 1 { - fmt.Printf(" - throttle_time_ms (4): %x\n", response[4:8]) - if apiVersion >= 4 { - if apiVersion == 5 { - // v5 doesn't have session_id at all - fmt.Printf(" - topics_count (4): %x\n", response[8:12]) - } else { - fmt.Printf(" - error_code (2): %x\n", response[8:10]) - fmt.Printf(" - session_id (4): %x\n", response[10:14]) - fmt.Printf(" - topics_count (4): %x\n", response[14:18]) + // 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) } - } else { - fmt.Printf(" - topics_count (4): %x\n", response[8:12]) + + // Fetch v4+ has aborted_transactions + if apiVersion >= 4 { + response = append(response, 0, 0, 0, 0) // aborted_transactions count (4 bytes) = 0 + } + + // Records - construct record batch with actual messages + recordBatch := h.constructRecordBatchFromLedger(ledger, partition.FetchOffset, highWaterMark) + + // 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 +} + // 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 {