diff --git a/test/kafka/schema_integration_test.go b/test/kafka/schema_integration_test.go new file mode 100644 index 000000000..641539b1e --- /dev/null +++ b/test/kafka/schema_integration_test.go @@ -0,0 +1,565 @@ +package kafka + +import ( + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/IBM/sarama" + "github.com/linkedin/goavro/v2" + "github.com/seaweedfs/seaweedfs/weed/mq/kafka/protocol" + "github.com/seaweedfs/seaweedfs/weed/mq/kafka/schema" +) + +// TestKafkaGateway_SchemaIntegration tests the full Kafka Gateway with schema support +func TestKafkaGateway_SchemaIntegration(t *testing.T) { + // Start mock schema registry + schemaRegistry := createTestSchemaRegistry(t) + defer schemaRegistry.Close() + + // Start Kafka Gateway with schema support + gateway := startKafkaGatewayWithSchema(t, schemaRegistry.URL) + defer gateway.Close() + + // Test with Sarama client + t.Run("Sarama_SchematizedProduceConsume", func(t *testing.T) { + testSaramaSchematizedWorkflow(t, gateway.URL) + }) + + // Test schema evolution + t.Run("Schema_Evolution", func(t *testing.T) { + testSchemaEvolution(t, gateway.URL, schemaRegistry.URL) + }) + + // Test error handling + t.Run("Schema_ErrorHandling", func(t *testing.T) { + testSchemaErrorHandling(t, gateway.URL) + }) +} + +// TestKafkaGateway_MultiFormatSupport tests multiple schema formats +func TestKafkaGateway_MultiFormatSupport(t *testing.T) { + schemaRegistry := createMultiFormatSchemaRegistry(t) + defer schemaRegistry.Close() + + gateway := startKafkaGatewayWithSchema(t, schemaRegistry.URL) + defer gateway.Close() + + formats := []struct { + name string + topic string + schemaID uint32 + format schema.Format + }{ + {"Avro", "avro-topic-value", 1, schema.FormatAvro}, + {"JSON_Schema", "json-topic-value", 3, schema.FormatJSONSchema}, + // Protobuf would be {"Protobuf", "proto-topic-value", 2, schema.FormatProtobuf}, + } + + for _, fmt := range formats { + t.Run(fmt.name, func(t *testing.T) { + testFormatSpecificWorkflow(t, gateway.URL, fmt.topic, fmt.schemaID, fmt.format) + }) + } +} + +// Helper functions + +func createTestSchemaRegistry(t *testing.T) *httptest.Server { + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/subjects": + subjects := []string{"user-value", "user-key", "product-value"} + json.NewEncoder(w).Encode(subjects) + + case "/schemas/ids/1": + // Avro user schema v1 + response := map[string]interface{}{ + "schema": `{ + "type": "record", + "name": "User", + "fields": [ + {"name": "id", "type": "int"}, + {"name": "name", "type": "string"}, + {"name": "email", "type": ["null", "string"], "default": null} + ] + }`, + "subject": "user-value", + "version": 1, + } + json.NewEncoder(w).Encode(response) + + case "/schemas/ids/2": + // Avro user schema v2 (evolved) + response := map[string]interface{}{ + "schema": `{ + "type": "record", + "name": "User", + "fields": [ + {"name": "id", "type": "int"}, + {"name": "name", "type": "string"}, + {"name": "email", "type": ["null", "string"], "default": null}, + {"name": "created_at", "type": ["null", "long"], "default": null} + ] + }`, + "subject": "user-value", + "version": 2, + } + json.NewEncoder(w).Encode(response) + + case "/subjects/user-value/versions/latest": + response := map[string]interface{}{ + "id": 2, + "version": 2, + "schema": `{ + "type": "record", + "name": "User", + "fields": [ + {"name": "id", "type": "int"}, + {"name": "name", "type": "string"}, + {"name": "email", "type": ["null", "string"], "default": null}, + {"name": "created_at", "type": ["null", "long"], "default": null} + ] + }`, + "subject": "user-value", + } + json.NewEncoder(w).Encode(response) + + default: + t.Logf("Schema registry: unhandled path %s", r.URL.Path) + w.WriteHeader(http.StatusNotFound) + } + })) +} + +func createMultiFormatSchemaRegistry(t *testing.T) *httptest.Server { + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/subjects": + subjects := []string{"avro-topic-value", "json-topic-value", "proto-topic-value"} + json.NewEncoder(w).Encode(subjects) + + case "/schemas/ids/1": + // Avro schema + response := map[string]interface{}{ + "schema": `{ + "type": "record", + "name": "AvroMessage", + "fields": [ + {"name": "id", "type": "int"}, + {"name": "message", "type": "string"} + ] + }`, + "subject": "avro-topic-value", + "version": 1, + } + json.NewEncoder(w).Encode(response) + + case "/schemas/ids/3": + // JSON Schema + response := map[string]interface{}{ + "schema": `{ + "$schema": "http://json-schema.org/draft-07/schema#", + "type": "object", + "properties": { + "id": {"type": "integer"}, + "message": {"type": "string"}, + "timestamp": {"type": "string", "format": "date-time"} + }, + "required": ["id", "message"] + }`, + "subject": "json-topic-value", + "version": 1, + } + json.NewEncoder(w).Encode(response) + + default: + w.WriteHeader(http.StatusNotFound) + } + })) +} + +func startKafkaGatewayWithSchema(t *testing.T, registryURL string) *TestServer { + // Create handler with schema support + handler := protocol.NewHandler() + + // Enable schema management + schemaConfig := schema.ManagerConfig{ + RegistryURL: registryURL, + ValidationMode: schema.ValidationPermissive, + EnableMirroring: false, + } + + if err := handler.EnableSchemaManagement(schemaConfig); err != nil { + t.Fatalf("Failed to enable schema management: %v", err) + } + + // Add test topics + handler.AddTopicForTesting("user-value", 1) + handler.AddTopicForTesting("avro-topic-value", 1) + handler.AddTopicForTesting("json-topic-value", 1) + + // Start server + server := &TestServer{ + Handler: handler, + URL: "localhost:9092", // Will be set by actual server start + } + + // In a real test, you would start the actual TCP server here + // For this integration test, we'll simulate the server behavior + + return server +} + +func testSaramaSchematizedWorkflow(t *testing.T, gatewayURL string) { + // This test would normally connect to the actual Kafka Gateway + // For demonstration, we'll test the schema processing logic directly + + t.Log("Testing Sarama schematized workflow") + + // Create test Avro message + avroSchema := `{ + "type": "record", + "name": "User", + "fields": [ + {"name": "id", "type": "int"}, + {"name": "name", "type": "string"}, + {"name": "email", "type": ["null", "string"], "default": null} + ] + }` + + codec, err := goavro.NewCodec(avroSchema) + if err != nil { + t.Fatalf("Failed to create Avro codec: %v", err) + } + + // Create user data + userData := map[string]interface{}{ + "id": int32(12345), + "name": "Integration Test User", + "email": map[string]interface{}{"string": "test@example.com"}, + } + + // Encode to Avro binary + avroBinary, err := codec.BinaryFromNative(nil, userData) + if err != nil { + t.Fatalf("Failed to encode Avro data: %v", err) + } + + // Create Confluent envelope (what would be sent by Sarama with schema registry) + confluentMsg := schema.CreateConfluentEnvelope(schema.FormatAvro, 1, nil, avroBinary) + + // Verify the message can be processed + envelope, ok := schema.ParseConfluentEnvelope(confluentMsg) + if !ok { + t.Fatal("Failed to parse Confluent envelope") + } + + if envelope.SchemaID != 1 { + t.Errorf("Expected schema ID 1, got %d", envelope.SchemaID) + } + + if len(envelope.Payload) != len(avroBinary) { + t.Errorf("Payload length mismatch: expected %d, got %d", len(avroBinary), len(envelope.Payload)) + } + + t.Logf("Successfully processed schematized message: %d bytes", len(confluentMsg)) +} + +func testSchemaEvolution(t *testing.T, gatewayURL, registryURL string) { + t.Log("Testing schema evolution") + + // Create manager for testing + config := schema.ManagerConfig{ + RegistryURL: registryURL, + ValidationMode: schema.ValidationPermissive, + } + + manager, err := schema.NewManager(config) + if err != nil { + t.Fatalf("Failed to create schema manager: %v", err) + } + + // Test v1 message + v1Data := map[string]interface{}{ + "id": int32(1), + "name": "User V1", + "email": map[string]interface{}{"string": "v1@example.com"}, + } + + v1Schema := `{ + "type": "record", + "name": "User", + "fields": [ + {"name": "id", "type": "int"}, + {"name": "name", "type": "string"}, + {"name": "email", "type": ["null", "string"], "default": null} + ] + }` + + codec1, _ := goavro.NewCodec(v1Schema) + v1Binary, _ := codec1.BinaryFromNative(nil, v1Data) + v1Msg := schema.CreateConfluentEnvelope(schema.FormatAvro, 1, nil, v1Binary) + + // Test v2 message (with additional field) + v2Data := map[string]interface{}{ + "id": int32(2), + "name": "User V2", + "email": map[string]interface{}{"string": "v2@example.com"}, + "created_at": map[string]interface{}{"long": time.Now().Unix()}, + } + + v2Schema := `{ + "type": "record", + "name": "User", + "fields": [ + {"name": "id", "type": "int"}, + {"name": "name", "type": "string"}, + {"name": "email", "type": ["null", "string"], "default": null}, + {"name": "created_at", "type": ["null", "long"], "default": null} + ] + }` + + codec2, _ := goavro.NewCodec(v2Schema) + v2Binary, _ := codec2.BinaryFromNative(nil, v2Data) + v2Msg := schema.CreateConfluentEnvelope(schema.FormatAvro, 2, nil, v2Binary) + + // Test that both versions can be processed + _, err = manager.DecodeMessage(v1Msg) + if err != nil { + t.Errorf("Failed to decode v1 message: %v", err) + } + + _, err = manager.DecodeMessage(v2Msg) + if err != nil { + t.Errorf("Failed to decode v2 message: %v", err) + } + + t.Log("Schema evolution test passed") +} + +func testSchemaErrorHandling(t *testing.T, gatewayURL string) { + t.Log("Testing schema error handling") + + // Test various error scenarios + errorCases := []struct { + name string + message []byte + desc string + }{ + { + name: "NonSchematizedMessage", + message: []byte("plain text message"), + desc: "Plain text should not be processed as schematized", + }, + { + name: "InvalidMagicByte", + message: []byte{0x01, 0x00, 0x00, 0x00, 0x01, 0x48, 0x65, 0x6c, 0x6c, 0x6f}, + desc: "Invalid magic byte should be rejected", + }, + { + name: "TooShortMessage", + message: []byte{0x00, 0x00, 0x00}, + desc: "Message too short for schema ID should be rejected", + }, + } + + for _, tc := range errorCases { + t.Run(tc.name, func(t *testing.T) { + envelope, ok := schema.ParseConfluentEnvelope(tc.message) + + switch tc.name { + case "NonSchematizedMessage", "InvalidMagicByte", "TooShortMessage": + if ok { + t.Errorf("Expected parsing to fail for %s, but it succeeded", tc.desc) + } else { + t.Logf("Correctly rejected: %s", tc.desc) + } + default: + if !ok { + t.Errorf("Expected parsing to succeed for %s, but it failed", tc.desc) + } + } + + _ = envelope // Use the variable to avoid unused warning + }) + } +} + +func testFormatSpecificWorkflow(t *testing.T, gatewayURL, topic string, schemaID uint32, format schema.Format) { + t.Logf("Testing %s format workflow for topic %s", format, topic) + + var testMessage []byte + var testData interface{} + + switch format { + case schema.FormatAvro: + // Create Avro test message + avroSchema := `{ + "type": "record", + "name": "AvroMessage", + "fields": [ + {"name": "id", "type": "int"}, + {"name": "message", "type": "string"} + ] + }` + + codec, _ := goavro.NewCodec(avroSchema) + testData = map[string]interface{}{ + "id": int32(123), + "message": "Avro test message", + } + avroBinary, _ := codec.BinaryFromNative(nil, testData) + testMessage = schema.CreateConfluentEnvelope(format, schemaID, nil, avroBinary) + + case schema.FormatJSONSchema: + // Create JSON Schema test message + testData = map[string]interface{}{ + "id": 456, + "message": "JSON test message", + "timestamp": time.Now().Format(time.RFC3339), + } + jsonData, _ := json.Marshal(testData) + testMessage = schema.CreateConfluentEnvelope(format, schemaID, nil, jsonData) + + case schema.FormatProtobuf: + // Protobuf would be implemented here + t.Skip("Protobuf format testing not fully implemented") + return + } + + // Verify message can be parsed + envelope, ok := schema.ParseConfluentEnvelope(testMessage) + if !ok { + t.Fatalf("Failed to parse %s message", format) + } + + if envelope.SchemaID != schemaID { + t.Errorf("Expected schema ID %d, got %d", schemaID, envelope.SchemaID) + } + + if len(envelope.Payload) == 0 { + t.Error("Expected non-empty payload") + } + + t.Logf("Successfully processed %s message: %d bytes", format, len(testMessage)) +} + +// TestServer represents a test Kafka Gateway server +type TestServer struct { + Handler *protocol.Handler + URL string +} + +func (ts *TestServer) Close() { + // In a real implementation, this would close the TCP server + if ts.Handler.IsSchemaEnabled() { + ts.Handler.DisableSchemaManagement() + } +} + +// Performance and load testing + +func TestKafkaGateway_SchemaPerformance(t *testing.T) { + if testing.Short() { + t.Skip("Skipping performance test in short mode") + } + + schemaRegistry := createTestSchemaRegistry(t) + defer schemaRegistry.Close() + + config := schema.ManagerConfig{ + RegistryURL: schemaRegistry.URL, + ValidationMode: schema.ValidationPermissive, + } + + manager, err := schema.NewManager(config) + if err != nil { + t.Fatalf("Failed to create schema manager: %v", err) + } + + // Create test message + avroSchema := `{ + "type": "record", + "name": "User", + "fields": [ + {"name": "id", "type": "int"}, + {"name": "name", "type": "string"} + ] + }` + + codec, _ := goavro.NewCodec(avroSchema) + testData := map[string]interface{}{ + "id": int32(1), + "name": "Performance Test", + } + avroBinary, _ := codec.BinaryFromNative(nil, testData) + testMsg := schema.CreateConfluentEnvelope(schema.FormatAvro, 1, nil, avroBinary) + + // Warm up cache + _, _ = manager.DecodeMessage(testMsg) + + // Performance test + start := time.Now() + iterations := 1000 + + for i := 0; i < iterations; i++ { + _, err := manager.DecodeMessage(testMsg) + if err != nil { + t.Fatalf("Decode failed at iteration %d: %v", i, err) + } + } + + duration := time.Since(start) + avgTime := duration / time.Duration(iterations) + + t.Logf("Performance test: %d iterations in %v (avg: %v per decode)", + iterations, duration, avgTime) + + // Verify reasonable performance (adjust threshold as needed) + if avgTime > time.Millisecond { + t.Logf("Warning: Average decode time %v may be too slow", avgTime) + } +} + +// Benchmark tests + +func BenchmarkKafkaGateway_AvroDecoding(b *testing.B) { + schemaRegistry := createTestSchemaRegistry(nil) + defer schemaRegistry.Close() + + config := schema.ManagerConfig{RegistryURL: schemaRegistry.URL} + manager, _ := schema.NewManager(config) + + // Create test message + avroSchema := `{"type": "record", "name": "User", "fields": [{"name": "id", "type": "int"}]}` + codec, _ := goavro.NewCodec(avroSchema) + testData := map[string]interface{}{"id": int32(1)} + avroBinary, _ := codec.BinaryFromNative(nil, testData) + testMsg := schema.CreateConfluentEnvelope(schema.FormatAvro, 1, nil, avroBinary) + + b.ResetTimer() + for i := 0; i < b.N; i++ { + _, _ = manager.DecodeMessage(testMsg) + } +} + +func BenchmarkKafkaGateway_JSONSchemaDecoding(b *testing.B) { + schemaRegistry := createMultiFormatSchemaRegistry(nil) + defer schemaRegistry.Close() + + config := schema.ManagerConfig{RegistryURL: schemaRegistry.URL} + manager, _ := schema.NewManager(config) + + // Create test message + jsonData := []byte(`{"id": 1, "message": "test"}`) + testMsg := schema.CreateConfluentEnvelope(schema.FormatJSONSchema, 3, nil, jsonData) + + b.ResetTimer() + for i := 0; i < b.N; i++ { + _, _ = manager.DecodeMessage(testMsg) + } +} diff --git a/weed/mq/kafka/schema/integration_test.go b/weed/mq/kafka/schema/integration_test.go new file mode 100644 index 000000000..cbfa68f2d --- /dev/null +++ b/weed/mq/kafka/schema/integration_test.go @@ -0,0 +1,629 @@ +package schema + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/linkedin/goavro/v2" +) + +// TestFullIntegration_AvroWorkflow tests the complete Avro workflow +func TestFullIntegration_AvroWorkflow(t *testing.T) { + // Create comprehensive mock schema registry + server := createMockSchemaRegistry(t) + defer server.Close() + + // Create manager with realistic configuration + config := ManagerConfig{ + RegistryURL: server.URL, + ValidationMode: ValidationPermissive, + EnableMirroring: false, + CacheTTL: "5m", + } + + manager, err := NewManager(config) + if err != nil { + t.Fatalf("Failed to create manager: %v", err) + } + + // Test 1: Producer workflow - encode schematized message + t.Run("Producer_Workflow", func(t *testing.T) { + // Create realistic user data (with proper Avro union handling) + userData := map[string]interface{}{ + "id": int32(12345), + "name": "Alice Johnson", + "email": map[string]interface{}{"string": "alice@example.com"}, // Avro union + "age": map[string]interface{}{"int": int32(28)}, // Avro union + "preferences": map[string]interface{}{ + "Preferences": map[string]interface{}{ // Avro union with record type + "notifications": true, + "theme": "dark", + }, + }, + } + + // Create Avro message (simulate what a Kafka producer would send) + avroSchema := getUserAvroSchema() + codec, err := goavro.NewCodec(avroSchema) + if err != nil { + t.Fatalf("Failed to create Avro codec: %v", err) + } + + avroBinary, err := codec.BinaryFromNative(nil, userData) + if err != nil { + t.Fatalf("Failed to encode Avro data: %v", err) + } + + // Create Confluent envelope (what Kafka Gateway receives) + confluentMsg := CreateConfluentEnvelope(FormatAvro, 1, nil, avroBinary) + + // Decode message (Produce path processing) + decodedMsg, err := manager.DecodeMessage(confluentMsg) + if err != nil { + t.Fatalf("Failed to decode message: %v", err) + } + + // Verify decoded data + if decodedMsg.SchemaID != 1 { + t.Errorf("Expected schema ID 1, got %d", decodedMsg.SchemaID) + } + + if decodedMsg.SchemaFormat != FormatAvro { + t.Errorf("Expected Avro format, got %v", decodedMsg.SchemaFormat) + } + + // Verify field values + fields := decodedMsg.RecordValue.Fields + if fields["id"].GetInt32Value() != 12345 { + t.Errorf("Expected id=12345, got %v", fields["id"].GetInt32Value()) + } + + if fields["name"].GetStringValue() != "Alice Johnson" { + t.Errorf("Expected name='Alice Johnson', got %v", fields["name"].GetStringValue()) + } + + t.Logf("Successfully processed producer message with %d fields", len(fields)) + }) + + // Test 2: Consumer workflow - reconstruct original message + t.Run("Consumer_Workflow", func(t *testing.T) { + // Create test RecordValue (simulate what's stored in SeaweedMQ) + testData := map[string]interface{}{ + "id": int32(67890), + "name": "Bob Smith", + "email": map[string]interface{}{"string": "bob@example.com"}, + "age": map[string]interface{}{"int": int32(35)}, // Avro union + } + recordValue := MapToRecordValue(testData) + + // Reconstruct message (Fetch path processing) + reconstructedMsg, err := manager.EncodeMessage(recordValue, 1, FormatAvro) + if err != nil { + t.Fatalf("Failed to reconstruct message: %v", err) + } + + // Verify reconstructed message can be parsed + envelope, ok := ParseConfluentEnvelope(reconstructedMsg) + if !ok { + t.Fatal("Failed to parse reconstructed envelope") + } + + if envelope.SchemaID != 1 { + t.Errorf("Expected schema ID 1, got %d", envelope.SchemaID) + } + + // Verify the payload can be decoded by Avro + avroSchema := getUserAvroSchema() + codec, err := goavro.NewCodec(avroSchema) + if err != nil { + t.Fatalf("Failed to create Avro codec: %v", err) + } + + decodedData, _, err := codec.NativeFromBinary(envelope.Payload) + if err != nil { + t.Fatalf("Failed to decode reconstructed Avro data: %v", err) + } + + // Verify data integrity + decodedMap := decodedData.(map[string]interface{}) + if decodedMap["id"] != int32(67890) { + t.Errorf("Expected id=67890, got %v", decodedMap["id"]) + } + + if decodedMap["name"] != "Bob Smith" { + t.Errorf("Expected name='Bob Smith', got %v", decodedMap["name"]) + } + + t.Logf("Successfully reconstructed consumer message: %d bytes", len(reconstructedMsg)) + }) + + // Test 3: Round-trip integrity + t.Run("Round_Trip_Integrity", func(t *testing.T) { + originalData := map[string]interface{}{ + "id": int32(99999), + "name": "Charlie Brown", + "email": map[string]interface{}{"string": "charlie@example.com"}, + "age": map[string]interface{}{"int": int32(42)}, // Avro union + } + + // Encode -> Decode -> Encode -> Decode + avroSchema := getUserAvroSchema() + codec, _ := goavro.NewCodec(avroSchema) + + // Step 1: Original -> Confluent + avroBinary, _ := codec.BinaryFromNative(nil, originalData) + confluentMsg := CreateConfluentEnvelope(FormatAvro, 1, nil, avroBinary) + + // Step 2: Confluent -> RecordValue + decodedMsg, _ := manager.DecodeMessage(confluentMsg) + + // Step 3: RecordValue -> Confluent + reconstructedMsg, _ := manager.EncodeMessage(decodedMsg.RecordValue, 1, FormatAvro) + + // Step 4: Confluent -> Verify + finalDecodedMsg, err := manager.DecodeMessage(reconstructedMsg) + if err != nil { + // Debug: Check if the reconstructed message is properly formatted + envelope, ok := ParseConfluentEnvelope(reconstructedMsg) + if !ok { + t.Fatalf("Round-trip failed: reconstructed message is not a valid Confluent envelope") + } + t.Logf("Debug: Envelope SchemaID=%d, Format=%v, PayloadLen=%d", + envelope.SchemaID, envelope.Format, len(envelope.Payload)) + t.Fatalf("Round-trip failed: %v", err) + } + + // Verify data integrity through complete round-trip + finalFields := finalDecodedMsg.RecordValue.Fields + if finalFields["id"].GetInt32Value() != 99999 { + t.Error("Round-trip failed for id field") + } + + if finalFields["name"].GetStringValue() != "Charlie Brown" { + t.Error("Round-trip failed for name field") + } + + t.Log("Round-trip integrity test passed") + }) +} + +// TestFullIntegration_MultiFormatSupport tests all schema formats together +func TestFullIntegration_MultiFormatSupport(t *testing.T) { + server := createMockSchemaRegistry(t) + defer server.Close() + + config := ManagerConfig{ + RegistryURL: server.URL, + ValidationMode: ValidationPermissive, + } + + manager, err := NewManager(config) + if err != nil { + t.Fatalf("Failed to create manager: %v", err) + } + + testCases := []struct { + name string + format Format + schemaID uint32 + testData interface{} + }{ + { + name: "Avro_Format", + format: FormatAvro, + schemaID: 1, + testData: map[string]interface{}{ + "id": int32(123), + "name": "Avro User", + }, + }, + { + name: "JSON_Schema_Format", + format: FormatJSONSchema, + schemaID: 3, + testData: map[string]interface{}{ + "id": float64(456), // JSON numbers are float64 + "name": "JSON User", + "active": true, + }, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + // Create RecordValue from test data + recordValue := MapToRecordValue(tc.testData.(map[string]interface{})) + + // Test encoding + encoded, err := manager.EncodeMessage(recordValue, tc.schemaID, tc.format) + if err != nil { + if tc.format == FormatProtobuf { + // Protobuf encoding may fail due to incomplete implementation + t.Skipf("Protobuf encoding not fully implemented: %v", err) + } else { + t.Fatalf("Failed to encode %s message: %v", tc.name, err) + } + } + + // Test decoding + decoded, err := manager.DecodeMessage(encoded) + if err != nil { + t.Fatalf("Failed to decode %s message: %v", tc.name, err) + } + + // Verify format + if decoded.SchemaFormat != tc.format { + t.Errorf("Expected format %v, got %v", tc.format, decoded.SchemaFormat) + } + + // Verify schema ID + if decoded.SchemaID != tc.schemaID { + t.Errorf("Expected schema ID %d, got %d", tc.schemaID, decoded.SchemaID) + } + + t.Logf("Successfully processed %s format", tc.name) + }) + } +} + +// TestIntegration_CachePerformance tests caching behavior under load +func TestIntegration_CachePerformance(t *testing.T) { + server := createMockSchemaRegistry(t) + defer server.Close() + + config := ManagerConfig{ + RegistryURL: server.URL, + ValidationMode: ValidationPermissive, + } + + manager, err := NewManager(config) + if err != nil { + t.Fatalf("Failed to create manager: %v", err) + } + + // Create test message + testData := map[string]interface{}{ + "id": int32(1), + "name": "Cache Test", + } + + avroSchema := getUserAvroSchema() + codec, _ := goavro.NewCodec(avroSchema) + avroBinary, _ := codec.BinaryFromNative(nil, testData) + testMsg := CreateConfluentEnvelope(FormatAvro, 1, nil, avroBinary) + + // First decode (should hit registry) + start := time.Now() + _, err = manager.DecodeMessage(testMsg) + if err != nil { + t.Fatalf("First decode failed: %v", err) + } + firstDuration := time.Since(start) + + // Subsequent decodes (should hit cache) + start = time.Now() + for i := 0; i < 100; i++ { + _, err = manager.DecodeMessage(testMsg) + if err != nil { + t.Fatalf("Cached decode failed: %v", err) + } + } + cachedDuration := time.Since(start) + + // Verify cache performance improvement + avgCachedTime := cachedDuration / 100 + if avgCachedTime >= firstDuration { + t.Logf("Warning: Cache may not be effective. First: %v, Avg Cached: %v", + firstDuration, avgCachedTime) + } + + // Check cache stats + decoders, schemas, subjects := manager.GetCacheStats() + if decoders == 0 || schemas == 0 { + t.Error("Expected non-zero cache stats") + } + + t.Logf("Cache performance: First decode: %v, Average cached: %v", + firstDuration, avgCachedTime) + t.Logf("Cache stats: %d decoders, %d schemas, %d subjects", + decoders, schemas, subjects) +} + +// TestIntegration_ErrorHandling tests error scenarios +func TestIntegration_ErrorHandling(t *testing.T) { + server := createMockSchemaRegistry(t) + defer server.Close() + + config := ManagerConfig{ + RegistryURL: server.URL, + ValidationMode: ValidationStrict, + } + + manager, err := NewManager(config) + if err != nil { + t.Fatalf("Failed to create manager: %v", err) + } + + testCases := []struct { + name string + message []byte + expectError bool + errorType string + }{ + { + name: "Non_Schematized_Message", + message: []byte("plain text message"), + expectError: true, + errorType: "not schematized", + }, + { + name: "Invalid_Schema_ID", + message: CreateConfluentEnvelope(FormatAvro, 999, nil, []byte("payload")), + expectError: true, + errorType: "schema not found", + }, + { + name: "Empty_Payload", + message: CreateConfluentEnvelope(FormatAvro, 1, nil, []byte{}), + expectError: true, + errorType: "empty payload", + }, + { + name: "Corrupted_Avro_Data", + message: CreateConfluentEnvelope(FormatAvro, 1, nil, []byte("invalid avro")), + expectError: true, + errorType: "decode failed", + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + _, err := manager.DecodeMessage(tc.message) + + if (err != nil) != tc.expectError { + t.Errorf("Expected error: %v, got error: %v", tc.expectError, err != nil) + } + + if tc.expectError && err != nil { + t.Logf("Expected error occurred: %v", err) + } + }) + } +} + +// TestIntegration_SchemaEvolution tests schema evolution scenarios +func TestIntegration_SchemaEvolution(t *testing.T) { + server := createMockSchemaRegistryWithEvolution(t) + defer server.Close() + + config := ManagerConfig{ + RegistryURL: server.URL, + ValidationMode: ValidationPermissive, + } + + manager, err := NewManager(config) + if err != nil { + t.Fatalf("Failed to create manager: %v", err) + } + + // Test decoding messages with different schema versions + t.Run("Schema_V1_Message", func(t *testing.T) { + // Create message with schema v1 (basic user) + userData := map[string]interface{}{ + "id": int32(1), + "name": "User V1", + } + + avroSchema := getUserAvroSchemaV1() + codec, _ := goavro.NewCodec(avroSchema) + avroBinary, _ := codec.BinaryFromNative(nil, userData) + msg := CreateConfluentEnvelope(FormatAvro, 1, nil, avroBinary) + + decoded, err := manager.DecodeMessage(msg) + if err != nil { + t.Fatalf("Failed to decode v1 message: %v", err) + } + + if decoded.Version != 1 { + t.Errorf("Expected version 1, got %d", decoded.Version) + } + }) + + t.Run("Schema_V2_Message", func(t *testing.T) { + // Create message with schema v2 (user with email) + userData := map[string]interface{}{ + "id": int32(2), + "name": "User V2", + "email": map[string]interface{}{"string": "user@example.com"}, + } + + avroSchema := getUserAvroSchemaV2() + codec, _ := goavro.NewCodec(avroSchema) + avroBinary, _ := codec.BinaryFromNative(nil, userData) + msg := CreateConfluentEnvelope(FormatAvro, 2, nil, avroBinary) + + decoded, err := manager.DecodeMessage(msg) + if err != nil { + t.Fatalf("Failed to decode v2 message: %v", err) + } + + if decoded.Version != 2 { + t.Errorf("Expected version 2, got %d", decoded.Version) + } + }) +} + +// Helper functions for creating mock schema registries + +func createMockSchemaRegistry(t *testing.T) *httptest.Server { + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/subjects": + // List subjects + subjects := []string{"user-value", "product-value", "order-value"} + json.NewEncoder(w).Encode(subjects) + + case "/schemas/ids/1": + // Avro user schema + response := map[string]interface{}{ + "schema": getUserAvroSchema(), + "subject": "user-value", + "version": 1, + } + json.NewEncoder(w).Encode(response) + + case "/schemas/ids/2": + // Protobuf schema (simplified) + response := map[string]interface{}{ + "schema": "syntax = \"proto3\"; message User { int32 id = 1; string name = 2; }", + "subject": "user-value", + "version": 2, + } + json.NewEncoder(w).Encode(response) + + case "/schemas/ids/3": + // JSON Schema + response := map[string]interface{}{ + "schema": getUserJSONSchema(), + "subject": "user-value", + "version": 3, + } + json.NewEncoder(w).Encode(response) + + default: + w.WriteHeader(http.StatusNotFound) + } + })) +} + +func createMockSchemaRegistryWithEvolution(t *testing.T) *httptest.Server { + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/schemas/ids/1": + // Schema v1 + response := map[string]interface{}{ + "schema": getUserAvroSchemaV1(), + "subject": "user-value", + "version": 1, + } + json.NewEncoder(w).Encode(response) + + case "/schemas/ids/2": + // Schema v2 (evolved) + response := map[string]interface{}{ + "schema": getUserAvroSchemaV2(), + "subject": "user-value", + "version": 2, + } + json.NewEncoder(w).Encode(response) + + default: + w.WriteHeader(http.StatusNotFound) + } + })) +} + +// Schema definitions for testing + +func getUserAvroSchema() string { + return `{ + "type": "record", + "name": "User", + "fields": [ + {"name": "id", "type": "int"}, + {"name": "name", "type": "string"}, + {"name": "email", "type": ["null", "string"], "default": null}, + {"name": "age", "type": ["null", "int"], "default": null}, + {"name": "preferences", "type": ["null", { + "type": "record", + "name": "Preferences", + "fields": [ + {"name": "notifications", "type": "boolean", "default": true}, + {"name": "theme", "type": "string", "default": "light"} + ] + }], "default": null} + ] + }` +} + +func getUserAvroSchemaV1() string { + return `{ + "type": "record", + "name": "User", + "fields": [ + {"name": "id", "type": "int"}, + {"name": "name", "type": "string"} + ] + }` +} + +func getUserAvroSchemaV2() string { + return `{ + "type": "record", + "name": "User", + "fields": [ + {"name": "id", "type": "int"}, + {"name": "name", "type": "string"}, + {"name": "email", "type": ["null", "string"], "default": null} + ] + }` +} + +func getUserJSONSchema() string { + return `{ + "$schema": "http://json-schema.org/draft-07/schema#", + "type": "object", + "properties": { + "id": {"type": "integer"}, + "name": {"type": "string"}, + "active": {"type": "boolean"} + }, + "required": ["id", "name"] + }` +} + +// Benchmark tests for integration scenarios + +func BenchmarkIntegration_AvroDecoding(b *testing.B) { + server := createMockSchemaRegistry(nil) + defer server.Close() + + config := ManagerConfig{RegistryURL: server.URL} + manager, _ := NewManager(config) + + // Create test message + testData := map[string]interface{}{ + "id": int32(1), + "name": "Benchmark User", + } + + avroSchema := getUserAvroSchema() + codec, _ := goavro.NewCodec(avroSchema) + avroBinary, _ := codec.BinaryFromNative(nil, testData) + testMsg := CreateConfluentEnvelope(FormatAvro, 1, nil, avroBinary) + + b.ResetTimer() + for i := 0; i < b.N; i++ { + _, _ = manager.DecodeMessage(testMsg) + } +} + +func BenchmarkIntegration_JSONSchemaDecoding(b *testing.B) { + server := createMockSchemaRegistry(nil) + defer server.Close() + + config := ManagerConfig{RegistryURL: server.URL} + manager, _ := NewManager(config) + + // Create test message + jsonData := []byte(`{"id": 1, "name": "Benchmark User", "active": true}`) + testMsg := CreateConfluentEnvelope(FormatJSONSchema, 3, nil, jsonData) + + b.ResetTimer() + for i := 0; i < b.N; i++ { + _, _ = manager.DecodeMessage(testMsg) + } +}