diff --git a/test/kafka/api_sequence_test.go b/test/kafka/api_sequence_test.go index 17eb12d5b..6ad51daae 100644 --- a/test/kafka/api_sequence_test.go +++ b/test/kafka/api_sequence_test.go @@ -48,7 +48,7 @@ func TestKafkaGateway_APISequence(t *testing.T) { defer writer.Close() // Try to write a single message and log the full API sequence - ctx, cancel := context.WithTimeout(context.Background(), 12*time.Second) + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) // Longer timeout to see all connection attempts defer cancel() fmt.Printf("\n=== STARTING kafka-go WRITE ATTEMPT ===\n") diff --git a/test/kafka/consumer_test.go b/test/kafka/consumer_test.go new file mode 100644 index 000000000..80e3d93ef --- /dev/null +++ b/test/kafka/consumer_test.go @@ -0,0 +1,67 @@ +package kafka + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/mq/kafka/gateway" + "github.com/segmentio/kafka-go" +) + +// TestKafkaGoReader tests if kafka-go Reader has different validation than Writer +func TestKafkaGoReader(t *testing.T) { + // Start the gateway server + srv := gateway.NewServer(gateway.Options{ + Listen: ":0", + UseSeaweedMQ: false, + }) + + if err := srv.Start(); err != nil { + t.Fatalf("Failed to start gateway: %v", err) + } + defer srv.Close() + + brokerAddr := srv.Addr() + t.Logf("Gateway running on %s", brokerAddr) + + // Pre-create topic + topicName := "reader-test-topic" + handler := srv.GetHandler() + handler.AddTopicForTesting(topicName, 1) + + // Create a Reader (consumer) with detailed logging + reader := kafka.NewReader(kafka.ReaderConfig{ + Brokers: []string{brokerAddr}, + Topic: topicName, + Logger: kafka.LoggerFunc(func(msg string, args ...interface{}) { + fmt.Printf("KAFKA-GO READER LOG: "+msg+"\n", args...) + }), + ErrorLogger: kafka.LoggerFunc(func(msg string, args ...interface{}) { + fmt.Printf("KAFKA-GO READER ERROR: "+msg+"\n", args...) + }), + }) + defer reader.Close() + + // Try to read messages (this will trigger Metadata requests) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + fmt.Printf("\n=== STARTING kafka-go READER TEST ===\n") + + // This should trigger Metadata and Fetch requests + message, err := reader.ReadMessage(ctx) + + fmt.Printf("\n=== kafka-go READER COMPLETED ===\n") + + if err != nil { + t.Logf("ReadMessage result: %v", err) + // Check if it's a timeout (expected for empty topic) vs validation error + if err == context.DeadlineExceeded { + t.Log("✅ Reader timed out waiting for messages - this suggests Metadata validation passed!") + } + } else { + t.Logf("ReadMessage succeeded: key=%s, value=%s", string(message.Key), string(message.Value)) + } +} diff --git a/weed/mq/kafka/protocol/handler.go b/weed/mq/kafka/protocol/handler.go index 6578743db..b3e8ae03a 100644 --- a/weed/mq/kafka/protocol/handler.go +++ b/weed/mq/kafka/protocol/handler.go @@ -141,7 +141,7 @@ func (h *Handler) HandleConn(conn net.Conn) error { fmt.Printf("DEBUG: [%s] Connection closing\n", connectionID) conn.Close() }() - + fmt.Printf("DEBUG: [%s] New connection established\n", connectionID) r := bufio.NewReader(conn) @@ -399,25 +399,37 @@ func (h *Handler) handleMetadata(correlationID uint32, requestBody []byte) ([]by // Parse topics from request (for metadata discovery) requestedTopics := h.parseMetadataTopics(requestBody) - // fmt.Printf("DEBUG: Metadata request for topics: %v (empty=all topics)\n", requestedTopics) + fmt.Printf("DEBUG: 🔍 METADATA REQUEST - Requested topics: %v (empty=all topics)\n", requestedTopics) // Build topics array response - return existing topics only h.topicsMu.RLock() + + // Debug: Show all available topics + availableTopics := make([]string, 0, len(h.topics)) + for topicName := range h.topics { + availableTopics = append(availableTopics, topicName) + } + fmt.Printf("DEBUG: 📋 AVAILABLE TOPICS: %v\n", availableTopics) + var topicsToReturn []string if len(requestedTopics) == 0 { // If no specific topics requested, return all existing topics for topicName := range h.topics { topicsToReturn = append(topicsToReturn, topicName) } - // fmt.Printf("DEBUG: Returning all existing topics: %v\n", topicsToReturn) + fmt.Printf("DEBUG: 📤 RETURNING all existing topics: %v\n", topicsToReturn) } else { // Return only requested topics that exist + fmt.Printf("DEBUG: 🔍 CHECKING requested topics: %v\n", requestedTopics) for _, topicName := range requestedTopics { if _, exists := h.topics[topicName]; exists { topicsToReturn = append(topicsToReturn, topicName) + fmt.Printf("DEBUG: ✅ Found requested topic: '%s'\n", topicName) + } else { + fmt.Printf("DEBUG: ❌ Topic NOT FOUND: '%s'\n", topicName) } } - fmt.Printf("DEBUG: Returning requested existing topics: %v\n", topicsToReturn) + fmt.Printf("DEBUG: 📤 RETURNING requested existing topics: %v\n", topicsToReturn) } h.topicsMu.RUnlock() @@ -448,11 +460,20 @@ func (h *Handler) handleMetadata(correlationID uint32, requestBody []byte) ([]by // fmt.Printf("DEBUG: Added partitions count: 1\n") // Partition 0: error_code(2) + partition_id(4) + leader_id(4) + replicas + isr - response = append(response, 0, 0) // no error - response = append(response, 0, 0, 0, 0) // partition_id = 0 - response = append(response, 0, 0, 0, 0) // leader_id = 0 (this broker) - response = append(response, 0, 0, 0, 1, 0, 0, 0, 0) // replicas = [0] - response = append(response, 0, 0, 0, 1, 0, 0, 0, 0) // isr = [0] + response = append(response, 0, 0) // no error + response = append(response, 0, 0, 0, 0) // partition_id = 0 + response = append(response, 0, 0, 0, 0) // leader_id = 0 (this broker) + + // Replicas array: length(4) + broker_ids + response = append(response, 0, 0, 0, 1) // replicas count = 1 + response = append(response, 0, 0, 0, 0) // replica broker_id = 0 + + // ISR (In-Sync Replicas) array: length(4) + broker_ids + response = append(response, 0, 0, 0, 1) // isr count = 1 + response = append(response, 0, 0, 0, 0) // isr broker_id = 0 + + // Debug: Show detailed partition info + fmt.Printf("DEBUG: Partition 0 - leader_id=0, replicas=[0], isr=[0]\n") // TEMP: Removed v7+ topic authorized operations for v1 compatibility }