- Replace hardcoded partition ranges with centralized kafka.CreateSMQPartition
- Use proper SMQ ranges instead of single partition numbers in storage paths
- All offset mapping tests now pass with consistent 32-slot ranges
- Remove unused pub_balancer import
- Created centralized PartitionMapper utility with consistent range size of 32
- Fixed inconsistent partition mapping between agent_client.go (78 range) and seaweedmq_handler.go (32 range)
- Updated smq_mapping.go to use centralized utility instead of dynamic calculation
- Standardized all partition mapping to use kafka.CreateSMQPartition() and related functions
- Added comprehensive tests for partition mapping consistency and round-trip conversion
- Achieves 99.05% ring utilization supporting 78 Kafka partitions
This fixes the high-priority issue where inconsistent partition mapping could cause
incorrect message routing between different components of the Kafka Gateway.
- Changed from strings.Split to strings.LastIndex to find the last colon
- This allows topic names like 'namespace:service:events' to be parsed correctly
- Updated test cases to verify topics with single and multiple colons work properly
- Kafka topic names are allowed to contain colons, so parsing must be robust
The previous logic would incorrectly parse 'my:topic:0' as topic='my', partition='topic:0'
Now it correctly parses as topic='my:topic', partition='0'
- Add unit tests for gateway connection/refusal in test/kafka/unit/gateway_test.go
- Add Schema Registry connectivity test in test/kafka/integration/docker_test.go
- Implement legacy offset key parsing in weed/mq/kafka/offset/smq_storage.go
- Fix leaderEpoch placeholder to return 0 instead of -1 in offset_management.go
- Add comprehensive test coverage for parseTopicPartitionKey function
All tests passing. Ready for Phase 0 Part 2 (API cleanup and logging).
- Remove old SMQIntegratedStorage implementation from persistence.go
- Update all integration modules to use SMQOffsetStorage instead
- Add delegation methods to PersistentLedger for backward compatibility
- Fix method signatures and compilation errors
- Maintain support for legacy offset operations through SeaweedMQStorage
- Add end-to-end flow tests for Kafka OffsetCommit to SMQ storage
- Test multiple consumer groups with independent offset tracking
- Validate SMQ file path and format compatibility
- Test error handling and edge cases (negative, zero, max offsets)
- Verify offset encoding/decoding matches SMQ broker format
- Ensure consumer group isolation and proper key generation
- Add SMQOffsetStorage that uses same filer locations and format as SMQ brokers
- Store offsets in <topic-dir>/<partition-dir>/<consumerGroup>.offset files
- Use 8-byte big-endian format matching SMQ broker implementation
- Include comprehensive test coverage for core functionality
- Maintain backward compatibility through legacy method support
Phase E2: Integrate Protobuf descriptor parser with decoder
- Update NewProtobufDecoder to use ProtobufDescriptorParser
- Add findFirstMessageName helper for automatic message detection
- Fix ParseBinaryDescriptor to return schema even on resolution failure
- Add comprehensive tests for protobuf decoder integration
- Improve error handling and caching behavior
This enables proper binary descriptor parsing in the protobuf decoder,
completing the integration between descriptor parsing and decoding.
Phase E3: Complete Protobuf message descriptor resolution
- Implement full protobuf descriptor resolution using protoreflect API
- Add buildFileDescriptor and findMessageInFileDescriptor methods
- Support nested message resolution with findNestedMessageDescriptor
- Add proper mutex protection for thread-safe cache access
- Update all test data to use proper field cardinality labels
- Update test expectations to handle successful descriptor resolution
- Enable full protobuf decoder creation from binary descriptors
Phase E (Protobuf Support) is now complete:
✅ E1: Binary descriptor parsing
✅ E2: Decoder integration
✅ E3: Full message descriptor resolution
Protobuf messages can now be fully parsed and decoded
Phase F: Implement Kafka record batch compression support
- Add comprehensive compression module supporting gzip/snappy/lz4/zstd
- Implement RecordBatchParser with full compression and CRC validation
- Support compression codec extraction from record batch attributes
- Add compression/decompression for all major Kafka codecs
- Integrate compression support into Produce and Fetch handlers
- Add extensive unit tests for all compression codecs
- Support round-trip compression/decompression with proper error handling
- Add performance benchmarks for compression operations
Key features:
✅ Gzip compression (ratio: 0.02)
✅ Snappy compression (ratio: 0.06, fastest)
✅ LZ4 compression (ratio: 0.02)
✅ Zstd compression (ratio: 0.01, best compression)
✅ CRC32 validation for record batch integrity
✅ Proper Kafka record batch format v2 parsing
✅ Backward compatibility with uncompressed records
Phase F (Compression Handling) is now complete.
Phase G: Implement advanced schema compatibility checking and migration
- Add comprehensive SchemaEvolutionChecker with full compatibility rules
- Support BACKWARD, FORWARD, FULL, and NONE compatibility levels
- Implement Avro schema compatibility checking with field analysis
- Add JSON Schema compatibility validation
- Support Protobuf compatibility checking (simplified implementation)
- Add type promotion rules (int->long, float->double, string<->bytes)
- Integrate schema evolution into Manager with validation methods
- Add schema evolution suggestions and migration guidance
- Support schema compatibility validation before evolution
- Add comprehensive unit tests for all compatibility scenarios
Key features:
✅ BACKWARD compatibility: New schema can read old data
✅ FORWARD compatibility: Old schema can read new data
✅ FULL compatibility: Both backward and forward compatible
✅ Type promotion support for safe schema evolution
✅ Field addition/removal validation with default value checks
✅ Schema evolution suggestions for incompatible changes
✅ Integration with schema registry for validation workflows
Phase G (Schema Evolution) is now complete.
fmt
chrislu