mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-08-16 04:06:44 +00:00
chrislu
592042e496
Commit e1a4bff79 applied Kafka client-side timeout to the entire produce
operation context, which breaks Schema Registry consumer initialization.
The bug:
- Schema Registry Produce request has 60000ms timeout
- This timeout was being applied to entire broker operation context
- Consumer initialization takes time (joins group, gets assignments, seeks, polls)
- If initialization isn't done before 60s, context times out
- Publish returns "context deadline exceeded" error
- Schema Registry times out
The fix:
- Remove context.WithTimeout() calls from produce handlers
- Revert to NOT applying client timeout to internal broker operations
- This allows consumer initialization to take as long as needed
- Kafka request will still timeout at protocol level naturally
NOTE: Consumer still not sending Fetch requests - there's likely a deeper
issue with consumer group coordination or partition assignment in the
gateway, separate from this timeout issue.
This removes the obvious timeout bug but may not completely fix SR init.
debug: Add instrumentation for Noop record timeout investigation
- Added critical debug logging to server.go connection acceptance
- Added handleProduce entry point logging
- Added 30+ debug statements to produce.go for Noop record tracing
- Created comprehensive investigation report
CRITICAL FINDING: Gateway accepts connections but requests hang in HandleConn()
request reading loop - no requests ever reach processRequestSync()
Files modified:
- weed/mq/kafka/gateway/server.go: Connection acceptance and HandleConn logging
- weed/mq/kafka/protocol/produce.go: Request entry logging and Noop tracing
See /tmp/INVESTIGATION_FINAL_REPORT.md for full analysis
Issue: Schema Registry Noop record write times out after 60 seconds
Root Cause: Kafka protocol request reading hangs in HandleConn loop
Status: Requires further debugging of request parsing logic in handler.go
debug: Add request reading loop instrumentation to handler.go
CRITICAL FINDING: Requests ARE being read and queued!
- Request header parsing works correctly
- Requests are successfully sent to data/control plane channels
- apiKey=3 (FindCoordinator) requests visible in logs
- Request queuing is NOT the bottleneck
Remaining issue: No Produce (apiKey=0) requests seen from Schema Registry
Hypothesis: Schema Registry stuck in metadata/coordinator discovery
Debug logs added to trace:
- Message size reading
- Message body reading
- API key/version/correlation ID parsing
- Request channel queuing
Next: Investigate why Produce requests not appearing
discovery: Add Fetch API logging - confirms consumer never initializes
SMOKING GUN CONFIRMED: Consumer NEVER sends Fetch requests!
Testing shows:
- Zero Fetch (apiKey=1) requests logged from Schema Registry
- Consumer never progresses past initialization
- This proves consumer group coordination is broken
Root Cause Confirmed:
The issue is NOT in Produce/Noop record handling.
The issue is NOT in message serialization.
The issue IS:
- Consumer cannot join group (JoinGroup/SyncGroup broken?)
- Consumer cannot assign partitions
- Consumer cannot begin fetching
This causes:
1. KafkaStoreReaderThread.doWork() hangs in consumer.poll()
2. Reader never signals initialization complete
3. Producer waiting for Noop ack times out
4. Schema Registry startup fails after 60 seconds
Next investigation:
- Add logging for JoinGroup (apiKey=11)
- Add logging for SyncGroup (apiKey=14)
- Add logging for Heartbeat (apiKey=12)
- Determine where in initialization the consumer gets stuck
Added Fetch API explicit logging that confirms it's never called.
Kafka Gateway Tests with SMQ Integration
This directory contains tests for the SeaweedFS Kafka Gateway with full SeaweedMQ (SMQ) integration.
Test Types
Unit Tests (./unit/)
- Basic gateway functionality
- Protocol compatibility
- No SeaweedFS backend required
- Uses mock handlers
Integration Tests (./integration/)
- Mock Mode (default): Uses in-memory handlers for protocol testing
- SMQ Mode (with
SEAWEEDFS_MASTERS): Uses real SeaweedFS backend for full integration
E2E Tests (./e2e/)
- End-to-end workflows
- Automatically detects SMQ availability
- Falls back to mock mode if SMQ unavailable
Running Tests Locally
Quick Protocol Testing (Mock Mode)
# Run all integration tests with mock backend
cd test/kafka
go test ./integration/...
# Run specific test
go test -v ./integration/ -run TestClientCompatibility
Full Integration Testing (SMQ Mode)
Requires running SeaweedFS instance:
- Start SeaweedFS with MQ support:
# Terminal 1: Start SeaweedFS server
weed server -ip="127.0.0.1" -ip.bind="0.0.0.0" -dir=/tmp/seaweedfs-data -master.port=9333 -volume.port=8081 -filer.port=8888 -filer=true
# Terminal 2: Start MQ broker
weed mq.broker -master="127.0.0.1:9333" -ip="127.0.0.1" -port=17777
- Run tests with SMQ backend:
cd test/kafka
SEAWEEDFS_MASTERS=127.0.0.1:9333 go test ./integration/...
# Run specific SMQ integration tests
SEAWEEDFS_MASTERS=127.0.0.1:9333 go test -v ./integration/ -run TestSMQIntegration
Test Broker Startup
If you're having broker startup issues:
# Debug broker startup locally
./scripts/test-broker-startup.sh
CI/CD Integration
GitHub Actions Jobs
- Unit Tests - Fast protocol tests with mock backend
- Integration Tests - Mock mode by default
- E2E Tests (with SMQ) - Full SeaweedFS + MQ broker stack
- Client Compatibility (with SMQ) - Tests different Kafka clients against real backend
- Consumer Group Tests (with SMQ) - Tests consumer group persistence
- SMQ Integration Tests - Dedicated SMQ-specific functionality tests
What Gets Tested with SMQ
When SEAWEEDFS_MASTERS is available, tests exercise:
- Real Message Persistence - Messages stored in SeaweedFS volumes
- Offset Persistence - Consumer group offsets stored in SeaweedFS filer
- Topic Persistence - Topic metadata persisted in SeaweedFS filer
- Consumer Group Coordination - Distributed coordinator assignment
- Cross-Client Compatibility - Sarama, kafka-go with real backend
- Broker Discovery - Gateway discovers MQ brokers via masters
Test Infrastructure
testutil.NewGatewayTestServerWithSMQ(t, mode)
Smart gateway creation that automatically:
- Detects SMQ availability via
SEAWEEDFS_MASTERS - Uses production handler when available
- Falls back to mock when unavailable
- Provides timeout protection against hanging
Modes:
SMQRequired- Skip test if SMQ unavailableSMQAvailable- Use SMQ if available, otherwise mockSMQUnavailable- Always use mock
Timeout Protection
Gateway creation includes timeout protection to prevent CI hanging:
- 20 second timeout for
SMQRequiredmode - 15 second timeout for
SMQAvailablemode - Clear error messages when broker discovery fails
Debugging Failed Tests
CI Logs to Check
- "SeaweedFS master is up" - Master started successfully
- "SeaweedFS filer is up" - Filer ready
- "SeaweedFS MQ broker is up" - Broker started successfully
- Broker/Server logs - Shown on broker startup failure
Local Debugging
- Run
./scripts/test-broker-startup.shto test broker startup - Check logs at
/tmp/weed-*.log - Test individual components:
# Test master curl http://127.0.0.1:9333/cluster/status # Test filer curl http://127.0.0.1:8888/status # Test broker nc -z 127.0.0.1 17777
Common Issues
- Broker fails to start: Check filer is ready before starting broker
- Gateway timeout: Broker discovery fails, check broker is accessible
- Test hangs: Timeout protection not working, reduce timeout values
Architecture
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ Kafka Client │───▶│ Kafka Gateway │───▶│ SeaweedMQ Broker│
│ (Sarama, │ │ (Protocol │ │ (Message │
│ kafka-go) │ │ Handler) │ │ Persistence) │
└─────────────────┘ └─────────────────┘ └─────────────────┘
│ │
▼ ▼
┌─────────────────┐ ┌─────────────────┐
│ SeaweedFS Filer │ │ SeaweedFS Master│
│ (Offset Storage)│ │ (Coordination) │
└─────────────────┘ └─────────────────┘
│ │
▼ ▼
┌─────────────────────────────────────────┐
│ SeaweedFS Volumes │
│ (Message Storage) │
└─────────────────────────────────────────┘
This architecture ensures full integration testing of the entire Kafka → SeaweedFS message path.