chrislu
7 days ago
5 changed files with 1373 additions and 296 deletions
-
148weed/mq/broker/broker_grpc_sub.go
-
872weed/mq/broker/broker_grpc_sub_seek_test.go
-
337weed/mq/kafka/integration/broker_client_subscribe.go
-
5weed/pb/mq_broker.proto
-
307weed/pb/mq_pb/mq_broker.pb.go
@ -0,0 +1,872 @@ |
|||
package broker |
|||
|
|||
import ( |
|||
"context" |
|||
"fmt" |
|||
"io" |
|||
"sync" |
|||
"testing" |
|||
"time" |
|||
|
|||
"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb" |
|||
"github.com/seaweedfs/seaweedfs/weed/pb/schema_pb" |
|||
"github.com/seaweedfs/seaweedfs/weed/util/log_buffer" |
|||
"google.golang.org/grpc/metadata" |
|||
) |
|||
|
|||
// TestGetRequestPositionFromSeek tests the helper function that converts seek requests to message positions
|
|||
func TestGetRequestPositionFromSeek(t *testing.T) { |
|||
broker := &MessageQueueBroker{} |
|||
|
|||
tests := []struct { |
|||
name string |
|||
offsetType schema_pb.OffsetType |
|||
offset int64 |
|||
expectedBatch int64 |
|||
expectZeroTime bool |
|||
}{ |
|||
{ |
|||
name: "reset to earliest", |
|||
offsetType: schema_pb.OffsetType_RESET_TO_EARLIEST, |
|||
offset: 0, |
|||
expectedBatch: -3, |
|||
expectZeroTime: false, |
|||
}, |
|||
{ |
|||
name: "reset to latest", |
|||
offsetType: schema_pb.OffsetType_RESET_TO_LATEST, |
|||
offset: 0, |
|||
expectedBatch: -4, |
|||
expectZeroTime: false, |
|||
}, |
|||
{ |
|||
name: "exact offset zero", |
|||
offsetType: schema_pb.OffsetType_EXACT_OFFSET, |
|||
offset: 0, |
|||
expectedBatch: 0, |
|||
expectZeroTime: true, |
|||
}, |
|||
{ |
|||
name: "exact offset 100", |
|||
offsetType: schema_pb.OffsetType_EXACT_OFFSET, |
|||
offset: 100, |
|||
expectedBatch: 100, |
|||
expectZeroTime: true, |
|||
}, |
|||
{ |
|||
name: "exact offset 1000", |
|||
offsetType: schema_pb.OffsetType_EXACT_OFFSET, |
|||
offset: 1000, |
|||
expectedBatch: 1000, |
|||
expectZeroTime: true, |
|||
}, |
|||
{ |
|||
name: "exact timestamp", |
|||
offsetType: schema_pb.OffsetType_EXACT_TS_NS, |
|||
offset: 1234567890123456789, |
|||
expectedBatch: -2, |
|||
expectZeroTime: false, |
|||
}, |
|||
{ |
|||
name: "reset to offset", |
|||
offsetType: schema_pb.OffsetType_RESET_TO_OFFSET, |
|||
offset: 42, |
|||
expectedBatch: 42, |
|||
expectZeroTime: true, |
|||
}, |
|||
} |
|||
|
|||
for _, tt := range tests { |
|||
t.Run(tt.name, func(t *testing.T) { |
|||
seekMsg := &mq_pb.SubscribeMessageRequest_SeekMessage{ |
|||
Offset: tt.offset, |
|||
OffsetType: tt.offsetType, |
|||
} |
|||
|
|||
position := broker.getRequestPositionFromSeek(seekMsg) |
|||
|
|||
if position.Offset != tt.expectedBatch { |
|||
t.Errorf("Expected batch index %d, got %d", tt.expectedBatch, position.Offset) |
|||
} |
|||
|
|||
// Verify time handling
|
|||
if tt.expectZeroTime && !position.Time.IsZero() { |
|||
t.Errorf("Expected zero time for offset-based seek, got %v", position.Time) |
|||
} |
|||
|
|||
if !tt.expectZeroTime && position.Time.IsZero() && tt.offsetType != schema_pb.OffsetType_RESET_TO_EARLIEST { |
|||
t.Errorf("Expected non-zero time, got zero time") |
|||
} |
|||
}) |
|||
} |
|||
} |
|||
|
|||
// TestGetRequestPositionFromSeek_NilSafety tests that the function handles nil input gracefully
|
|||
func TestGetRequestPositionFromSeek_NilSafety(t *testing.T) { |
|||
broker := &MessageQueueBroker{} |
|||
|
|||
position := broker.getRequestPositionFromSeek(nil) |
|||
|
|||
// Should return zero-value position without panicking
|
|||
if position.Offset != 0 { |
|||
t.Errorf("Expected zero offset for nil input, got %d", position.Offset) |
|||
} |
|||
} |
|||
|
|||
// TestGetRequestPositionFromSeek_ConsistentResults verifies that multiple calls with same input produce same output
|
|||
func TestGetRequestPositionFromSeek_ConsistentResults(t *testing.T) { |
|||
broker := &MessageQueueBroker{} |
|||
|
|||
seekMsg := &mq_pb.SubscribeMessageRequest_SeekMessage{ |
|||
Offset: 42, |
|||
OffsetType: schema_pb.OffsetType_EXACT_OFFSET, |
|||
} |
|||
|
|||
// Call multiple times
|
|||
positions := make([]log_buffer.MessagePosition, 5) |
|||
for i := 0; i < 5; i++ { |
|||
positions[i] = broker.getRequestPositionFromSeek(seekMsg) |
|||
time.Sleep(1 * time.Millisecond) // Small delay
|
|||
} |
|||
|
|||
// All positions should be identical
|
|||
for i := 1; i < len(positions); i++ { |
|||
if positions[i].Offset != positions[0].Offset { |
|||
t.Errorf("Inconsistent Offset: %d vs %d", positions[0].Offset, positions[i].Offset) |
|||
} |
|||
if !positions[i].Time.Equal(positions[0].Time) { |
|||
t.Errorf("Inconsistent Time: %v vs %v", positions[0].Time, positions[i].Time) |
|||
} |
|||
if positions[i].IsOffsetBased != positions[0].IsOffsetBased { |
|||
t.Errorf("Inconsistent IsOffsetBased: %v vs %v", positions[0].IsOffsetBased, positions[i].IsOffsetBased) |
|||
} |
|||
} |
|||
} |
|||
|
|||
// TestGetRequestPositionFromSeek_OffsetExtraction verifies offset can be correctly extracted
|
|||
func TestGetRequestPositionFromSeek_OffsetExtraction(t *testing.T) { |
|||
broker := &MessageQueueBroker{} |
|||
|
|||
testOffsets := []int64{0, 1, 10, 100, 1000, 9999} |
|||
|
|||
for _, offset := range testOffsets { |
|||
t.Run(fmt.Sprintf("offset_%d", offset), func(t *testing.T) { |
|||
seekMsg := &mq_pb.SubscribeMessageRequest_SeekMessage{ |
|||
Offset: offset, |
|||
OffsetType: schema_pb.OffsetType_EXACT_OFFSET, |
|||
} |
|||
|
|||
position := broker.getRequestPositionFromSeek(seekMsg) |
|||
|
|||
if !position.IsOffsetBased { |
|||
t.Error("Position should be detected as offset-based") |
|||
} |
|||
|
|||
if extractedOffset := position.GetOffset(); extractedOffset != offset { |
|||
t.Errorf("Expected extracted offset %d, got %d", offset, extractedOffset) |
|||
} |
|||
}) |
|||
} |
|||
} |
|||
|
|||
// MockSubscribeMessageStream is a mock implementation of the gRPC stream for testing
|
|||
type MockSubscribeMessageStream struct { |
|||
ctx context.Context |
|||
recvChan chan *mq_pb.SubscribeMessageRequest |
|||
sentMessages []*mq_pb.SubscribeMessageResponse |
|||
mu sync.Mutex |
|||
recvErr error |
|||
} |
|||
|
|||
func NewMockSubscribeMessageStream(ctx context.Context) *MockSubscribeMessageStream { |
|||
return &MockSubscribeMessageStream{ |
|||
ctx: ctx, |
|||
recvChan: make(chan *mq_pb.SubscribeMessageRequest, 10), |
|||
sentMessages: make([]*mq_pb.SubscribeMessageResponse, 0), |
|||
} |
|||
} |
|||
|
|||
func (m *MockSubscribeMessageStream) Send(msg *mq_pb.SubscribeMessageResponse) error { |
|||
m.mu.Lock() |
|||
defer m.mu.Unlock() |
|||
m.sentMessages = append(m.sentMessages, msg) |
|||
return nil |
|||
} |
|||
|
|||
func (m *MockSubscribeMessageStream) Recv() (*mq_pb.SubscribeMessageRequest, error) { |
|||
if m.recvErr != nil { |
|||
return nil, m.recvErr |
|||
} |
|||
|
|||
select { |
|||
case msg := <-m.recvChan: |
|||
return msg, nil |
|||
case <-m.ctx.Done(): |
|||
return nil, io.EOF |
|||
} |
|||
} |
|||
|
|||
func (m *MockSubscribeMessageStream) SetHeader(metadata.MD) error { |
|||
return nil |
|||
} |
|||
|
|||
func (m *MockSubscribeMessageStream) SendHeader(metadata.MD) error { |
|||
return nil |
|||
} |
|||
|
|||
func (m *MockSubscribeMessageStream) SetTrailer(metadata.MD) {} |
|||
|
|||
func (m *MockSubscribeMessageStream) Context() context.Context { |
|||
return m.ctx |
|||
} |
|||
|
|||
func (m *MockSubscribeMessageStream) SendMsg(interface{}) error { |
|||
return nil |
|||
} |
|||
|
|||
func (m *MockSubscribeMessageStream) RecvMsg(interface{}) error { |
|||
return nil |
|||
} |
|||
|
|||
func (m *MockSubscribeMessageStream) QueueMessage(msg *mq_pb.SubscribeMessageRequest) { |
|||
m.recvChan <- msg |
|||
} |
|||
|
|||
func (m *MockSubscribeMessageStream) SetRecvError(err error) { |
|||
m.recvErr = err |
|||
} |
|||
|
|||
func (m *MockSubscribeMessageStream) GetSentMessages() []*mq_pb.SubscribeMessageResponse { |
|||
m.mu.Lock() |
|||
defer m.mu.Unlock() |
|||
return append([]*mq_pb.SubscribeMessageResponse{}, m.sentMessages...) |
|||
} |
|||
|
|||
// TestSeekMessageHandling_BasicSeek tests that seek messages are properly received and acknowledged
|
|||
func TestSeekMessageHandling_BasicSeek(t *testing.T) { |
|||
// Create seek message
|
|||
seekMsg := &mq_pb.SubscribeMessageRequest{ |
|||
Message: &mq_pb.SubscribeMessageRequest_Seek{ |
|||
Seek: &mq_pb.SubscribeMessageRequest_SeekMessage{ |
|||
Offset: 100, |
|||
OffsetType: schema_pb.OffsetType_EXACT_OFFSET, |
|||
}, |
|||
}, |
|||
} |
|||
|
|||
// Verify message structure
|
|||
if seekReq := seekMsg.GetSeek(); seekReq == nil { |
|||
t.Fatal("Failed to create seek message") |
|||
} else { |
|||
if seekReq.Offset != 100 { |
|||
t.Errorf("Expected offset 100, got %d", seekReq.Offset) |
|||
} |
|||
if seekReq.OffsetType != schema_pb.OffsetType_EXACT_OFFSET { |
|||
t.Errorf("Expected EXACT_OFFSET, got %v", seekReq.OffsetType) |
|||
} |
|||
} |
|||
} |
|||
|
|||
// TestSeekMessageHandling_MultipleSeekTypes tests different seek offset types
|
|||
func TestSeekMessageHandling_MultipleSeekTypes(t *testing.T) { |
|||
testCases := []struct { |
|||
name string |
|||
offset int64 |
|||
offsetType schema_pb.OffsetType |
|||
}{ |
|||
{ |
|||
name: "seek to earliest", |
|||
offset: 0, |
|||
offsetType: schema_pb.OffsetType_RESET_TO_EARLIEST, |
|||
}, |
|||
{ |
|||
name: "seek to latest", |
|||
offset: 0, |
|||
offsetType: schema_pb.OffsetType_RESET_TO_LATEST, |
|||
}, |
|||
{ |
|||
name: "seek to exact offset", |
|||
offset: 42, |
|||
offsetType: schema_pb.OffsetType_EXACT_OFFSET, |
|||
}, |
|||
{ |
|||
name: "seek to timestamp", |
|||
offset: time.Now().UnixNano(), |
|||
offsetType: schema_pb.OffsetType_EXACT_TS_NS, |
|||
}, |
|||
{ |
|||
name: "reset to offset", |
|||
offset: 1000, |
|||
offsetType: schema_pb.OffsetType_RESET_TO_OFFSET, |
|||
}, |
|||
} |
|||
|
|||
for _, tc := range testCases { |
|||
t.Run(tc.name, func(t *testing.T) { |
|||
seekMsg := &mq_pb.SubscribeMessageRequest{ |
|||
Message: &mq_pb.SubscribeMessageRequest_Seek{ |
|||
Seek: &mq_pb.SubscribeMessageRequest_SeekMessage{ |
|||
Offset: tc.offset, |
|||
OffsetType: tc.offsetType, |
|||
}, |
|||
}, |
|||
} |
|||
|
|||
seekReq := seekMsg.GetSeek() |
|||
if seekReq == nil { |
|||
t.Fatal("Failed to get seek message") |
|||
} |
|||
|
|||
if seekReq.Offset != tc.offset { |
|||
t.Errorf("Expected offset %d, got %d", tc.offset, seekReq.Offset) |
|||
} |
|||
|
|||
if seekReq.OffsetType != tc.offsetType { |
|||
t.Errorf("Expected offset type %v, got %v", tc.offsetType, seekReq.OffsetType) |
|||
} |
|||
}) |
|||
} |
|||
} |
|||
|
|||
// TestSeekMessageHandling_AckVsSeekDistinction tests that we can distinguish between ack and seek messages
|
|||
func TestSeekMessageHandling_AckVsSeekDistinction(t *testing.T) { |
|||
// Create ack message
|
|||
ackMsg := &mq_pb.SubscribeMessageRequest{ |
|||
Message: &mq_pb.SubscribeMessageRequest_Ack{ |
|||
Ack: &mq_pb.SubscribeMessageRequest_AckMessage{ |
|||
Key: []byte("test-key"), |
|||
TsNs: time.Now().UnixNano(), |
|||
}, |
|||
}, |
|||
} |
|||
|
|||
// Create seek message
|
|||
seekMsg := &mq_pb.SubscribeMessageRequest{ |
|||
Message: &mq_pb.SubscribeMessageRequest_Seek{ |
|||
Seek: &mq_pb.SubscribeMessageRequest_SeekMessage{ |
|||
Offset: 100, |
|||
OffsetType: schema_pb.OffsetType_EXACT_OFFSET, |
|||
}, |
|||
}, |
|||
} |
|||
|
|||
// Verify ack message doesn't match seek
|
|||
if ackMsg.GetSeek() != nil { |
|||
t.Error("Ack message should not be detected as seek") |
|||
} |
|||
if ackMsg.GetAck() == nil { |
|||
t.Error("Ack message should be detected as ack") |
|||
} |
|||
|
|||
// Verify seek message doesn't match ack
|
|||
if seekMsg.GetAck() != nil { |
|||
t.Error("Seek message should not be detected as ack") |
|||
} |
|||
if seekMsg.GetSeek() == nil { |
|||
t.Error("Seek message should be detected as seek") |
|||
} |
|||
} |
|||
|
|||
// TestSeekMessageResponse_SuccessFormat tests the response format for successful seek
|
|||
func TestSeekMessageResponse_SuccessFormat(t *testing.T) { |
|||
// Create success response (empty error string = success)
|
|||
successResponse := &mq_pb.SubscribeMessageResponse{ |
|||
Message: &mq_pb.SubscribeMessageResponse_Ctrl{ |
|||
Ctrl: &mq_pb.SubscribeMessageResponse_SubscribeCtrlMessage{ |
|||
Error: "", // Empty error means success
|
|||
}, |
|||
}, |
|||
} |
|||
|
|||
ctrlMsg := successResponse.GetCtrl() |
|||
if ctrlMsg == nil { |
|||
t.Fatal("Failed to get control message") |
|||
} |
|||
|
|||
// Empty error string indicates success
|
|||
if ctrlMsg.Error != "" { |
|||
t.Errorf("Expected empty error for success, got: %s", ctrlMsg.Error) |
|||
} |
|||
} |
|||
|
|||
// TestSeekMessageResponse_ErrorFormat tests the response format for failed seek
|
|||
func TestSeekMessageResponse_ErrorFormat(t *testing.T) { |
|||
// Create error response
|
|||
errorResponse := &mq_pb.SubscribeMessageResponse{ |
|||
Message: &mq_pb.SubscribeMessageResponse_Ctrl{ |
|||
Ctrl: &mq_pb.SubscribeMessageResponse_SubscribeCtrlMessage{ |
|||
Error: "Seek not implemented", |
|||
}, |
|||
}, |
|||
} |
|||
|
|||
ctrlMsg := errorResponse.GetCtrl() |
|||
if ctrlMsg == nil { |
|||
t.Fatal("Failed to get control message") |
|||
} |
|||
|
|||
// Non-empty error string indicates failure
|
|||
if ctrlMsg.Error == "" { |
|||
t.Error("Expected non-empty error for failure") |
|||
} |
|||
|
|||
if ctrlMsg.Error != "Seek not implemented" { |
|||
t.Errorf("Expected specific error message, got: %s", ctrlMsg.Error) |
|||
} |
|||
} |
|||
|
|||
// TestSeekMessageHandling_BackwardSeek tests backward seeking scenarios
|
|||
func TestSeekMessageHandling_BackwardSeek(t *testing.T) { |
|||
testCases := []struct { |
|||
name string |
|||
currentPos int64 |
|||
seekOffset int64 |
|||
expectedGap int64 |
|||
}{ |
|||
{ |
|||
name: "small backward gap", |
|||
currentPos: 100, |
|||
seekOffset: 90, |
|||
expectedGap: 10, |
|||
}, |
|||
{ |
|||
name: "medium backward gap", |
|||
currentPos: 1000, |
|||
seekOffset: 500, |
|||
expectedGap: 500, |
|||
}, |
|||
{ |
|||
name: "large backward gap", |
|||
currentPos: 1000000, |
|||
seekOffset: 1, |
|||
expectedGap: 999999, |
|||
}, |
|||
{ |
|||
name: "seek to zero", |
|||
currentPos: 100, |
|||
seekOffset: 0, |
|||
expectedGap: 100, |
|||
}, |
|||
} |
|||
|
|||
for _, tc := range testCases { |
|||
t.Run(tc.name, func(t *testing.T) { |
|||
// Verify gap calculation
|
|||
gap := tc.currentPos - tc.seekOffset |
|||
if gap != tc.expectedGap { |
|||
t.Errorf("Expected gap %d, got %d", tc.expectedGap, gap) |
|||
} |
|||
|
|||
// Create seek message for backward seek
|
|||
seekMsg := &mq_pb.SubscribeMessageRequest{ |
|||
Message: &mq_pb.SubscribeMessageRequest_Seek{ |
|||
Seek: &mq_pb.SubscribeMessageRequest_SeekMessage{ |
|||
Offset: tc.seekOffset, |
|||
OffsetType: schema_pb.OffsetType_EXACT_OFFSET, |
|||
}, |
|||
}, |
|||
} |
|||
|
|||
seekReq := seekMsg.GetSeek() |
|||
if seekReq == nil { |
|||
t.Fatal("Failed to create seek message") |
|||
} |
|||
|
|||
if seekReq.Offset != tc.seekOffset { |
|||
t.Errorf("Expected offset %d, got %d", tc.seekOffset, seekReq.Offset) |
|||
} |
|||
}) |
|||
} |
|||
} |
|||
|
|||
// TestSeekMessageHandling_ForwardSeek tests forward seeking scenarios
|
|||
func TestSeekMessageHandling_ForwardSeek(t *testing.T) { |
|||
testCases := []struct { |
|||
name string |
|||
currentPos int64 |
|||
seekOffset int64 |
|||
shouldSeek bool |
|||
}{ |
|||
{ |
|||
name: "small forward gap", |
|||
currentPos: 100, |
|||
seekOffset: 110, |
|||
shouldSeek: false, // Forward seeks don't need special handling
|
|||
}, |
|||
{ |
|||
name: "same position", |
|||
currentPos: 100, |
|||
seekOffset: 100, |
|||
shouldSeek: false, |
|||
}, |
|||
{ |
|||
name: "large forward gap", |
|||
currentPos: 100, |
|||
seekOffset: 10000, |
|||
shouldSeek: false, |
|||
}, |
|||
} |
|||
|
|||
for _, tc := range testCases { |
|||
t.Run(tc.name, func(t *testing.T) { |
|||
// For forward seeks, gateway typically just continues reading
|
|||
// No special seek message needed
|
|||
isBackward := tc.seekOffset < tc.currentPos |
|||
|
|||
if isBackward && !tc.shouldSeek { |
|||
t.Error("Backward seek should require seek message") |
|||
} |
|||
}) |
|||
} |
|||
} |
|||
|
|||
// TestSeekIntegration_PositionConversion tests the complete flow from seek message to position
|
|||
func TestSeekIntegration_PositionConversion(t *testing.T) { |
|||
broker := &MessageQueueBroker{} |
|||
|
|||
testCases := []struct { |
|||
name string |
|||
offset int64 |
|||
offsetType schema_pb.OffsetType |
|||
verifyFunc func(t *testing.T, pos log_buffer.MessagePosition) |
|||
}{ |
|||
{ |
|||
name: "exact offset conversion", |
|||
offset: 42, |
|||
offsetType: schema_pb.OffsetType_EXACT_OFFSET, |
|||
verifyFunc: func(t *testing.T, pos log_buffer.MessagePosition) { |
|||
if !pos.IsOffsetBased { |
|||
t.Error("Expected offset-based position") |
|||
} |
|||
if pos.GetOffset() != 42 { |
|||
t.Errorf("Expected offset 42, got %d", pos.GetOffset()) |
|||
} |
|||
}, |
|||
}, |
|||
{ |
|||
name: "earliest offset conversion", |
|||
offset: 0, |
|||
offsetType: schema_pb.OffsetType_RESET_TO_EARLIEST, |
|||
verifyFunc: func(t *testing.T, pos log_buffer.MessagePosition) { |
|||
if pos.Offset != -3 { |
|||
t.Errorf("Expected batch -3 for earliest, got %d", pos.Offset) |
|||
} |
|||
}, |
|||
}, |
|||
{ |
|||
name: "latest offset conversion", |
|||
offset: 0, |
|||
offsetType: schema_pb.OffsetType_RESET_TO_LATEST, |
|||
verifyFunc: func(t *testing.T, pos log_buffer.MessagePosition) { |
|||
if pos.Offset != -4 { |
|||
t.Errorf("Expected batch -4 for latest, got %d", pos.Offset) |
|||
} |
|||
}, |
|||
}, |
|||
} |
|||
|
|||
for _, tc := range testCases { |
|||
t.Run(tc.name, func(t *testing.T) { |
|||
// Create seek message
|
|||
seekMsg := &mq_pb.SubscribeMessageRequest_SeekMessage{ |
|||
Offset: tc.offset, |
|||
OffsetType: tc.offsetType, |
|||
} |
|||
|
|||
// Convert to position
|
|||
position := broker.getRequestPositionFromSeek(seekMsg) |
|||
|
|||
// Verify result
|
|||
tc.verifyFunc(t, position) |
|||
}) |
|||
} |
|||
} |
|||
|
|||
// TestSeekMessageHandling_ConcurrentSeeks tests handling multiple seek requests
|
|||
func TestSeekMessageHandling_ConcurrentSeeks(t *testing.T) { |
|||
broker := &MessageQueueBroker{} |
|||
|
|||
// Simulate multiple concurrent seek requests
|
|||
seekOffsets := []int64{10, 20, 30, 40, 50} |
|||
|
|||
var wg sync.WaitGroup |
|||
results := make([]log_buffer.MessagePosition, len(seekOffsets)) |
|||
|
|||
for i, offset := range seekOffsets { |
|||
wg.Add(1) |
|||
go func(idx int, off int64) { |
|||
defer wg.Done() |
|||
|
|||
seekMsg := &mq_pb.SubscribeMessageRequest_SeekMessage{ |
|||
Offset: off, |
|||
OffsetType: schema_pb.OffsetType_EXACT_OFFSET, |
|||
} |
|||
|
|||
results[idx] = broker.getRequestPositionFromSeek(seekMsg) |
|||
}(i, offset) |
|||
} |
|||
|
|||
wg.Wait() |
|||
|
|||
// Verify all results are correct
|
|||
for i, offset := range seekOffsets { |
|||
if results[i].GetOffset() != offset { |
|||
t.Errorf("Expected offset %d at index %d, got %d", offset, i, results[i].GetOffset()) |
|||
} |
|||
} |
|||
} |
|||
|
|||
// TestSeekMessageProtocol_WireFormat verifies the protobuf message structure
|
|||
func TestSeekMessageProtocol_WireFormat(t *testing.T) { |
|||
// Test that SeekMessage is properly defined in the oneof
|
|||
req := &mq_pb.SubscribeMessageRequest{ |
|||
Message: &mq_pb.SubscribeMessageRequest_Seek{ |
|||
Seek: &mq_pb.SubscribeMessageRequest_SeekMessage{ |
|||
Offset: 100, |
|||
OffsetType: schema_pb.OffsetType_EXACT_OFFSET, |
|||
}, |
|||
}, |
|||
} |
|||
|
|||
// Verify oneof is set correctly
|
|||
switch msg := req.Message.(type) { |
|||
case *mq_pb.SubscribeMessageRequest_Seek: |
|||
if msg.Seek.Offset != 100 { |
|||
t.Errorf("Expected offset 100, got %d", msg.Seek.Offset) |
|||
} |
|||
default: |
|||
t.Errorf("Expected Seek message, got %T", msg) |
|||
} |
|||
|
|||
// Verify other message types are nil
|
|||
if req.GetAck() != nil { |
|||
t.Error("Seek message should not have Ack") |
|||
} |
|||
if req.GetInit() != nil { |
|||
t.Error("Seek message should not have Init") |
|||
} |
|||
} |
|||
|
|||
// TestSeekByTimestamp tests timestamp-based seek operations
|
|||
func TestSeekByTimestamp(t *testing.T) { |
|||
broker := &MessageQueueBroker{} |
|||
|
|||
testCases := []struct { |
|||
name string |
|||
timestampNs int64 |
|||
offsetType schema_pb.OffsetType |
|||
}{ |
|||
{ |
|||
name: "seek to specific timestamp", |
|||
timestampNs: 1234567890123456789, |
|||
offsetType: schema_pb.OffsetType_EXACT_TS_NS, |
|||
}, |
|||
{ |
|||
name: "seek to current timestamp", |
|||
timestampNs: time.Now().UnixNano(), |
|||
offsetType: schema_pb.OffsetType_EXACT_TS_NS, |
|||
}, |
|||
{ |
|||
name: "seek to past timestamp", |
|||
timestampNs: time.Now().Add(-24 * time.Hour).UnixNano(), |
|||
offsetType: schema_pb.OffsetType_EXACT_TS_NS, |
|||
}, |
|||
} |
|||
|
|||
for _, tc := range testCases { |
|||
t.Run(tc.name, func(t *testing.T) { |
|||
seekMsg := &mq_pb.SubscribeMessageRequest_SeekMessage{ |
|||
Offset: tc.timestampNs, |
|||
OffsetType: tc.offsetType, |
|||
} |
|||
|
|||
position := broker.getRequestPositionFromSeek(seekMsg) |
|||
|
|||
// For timestamp-based seeks, Time should be set to the timestamp
|
|||
expectedTime := time.Unix(0, tc.timestampNs) |
|||
if !position.Time.Equal(expectedTime) { |
|||
t.Errorf("Expected time %v, got %v", expectedTime, position.Time) |
|||
} |
|||
|
|||
// Batch should be -2 for EXACT_TS_NS
|
|||
if position.Offset != -2 { |
|||
t.Errorf("Expected batch -2 for timestamp seek, got %d", position.Offset) |
|||
} |
|||
}) |
|||
} |
|||
} |
|||
|
|||
// TestSeekByTimestamp_Ordering tests that timestamp seeks preserve ordering
|
|||
func TestSeekByTimestamp_Ordering(t *testing.T) { |
|||
broker := &MessageQueueBroker{} |
|||
|
|||
// Create timestamps in chronological order
|
|||
baseTime := time.Now().Add(-1 * time.Hour) |
|||
timestamps := []int64{ |
|||
baseTime.UnixNano(), |
|||
baseTime.Add(10 * time.Minute).UnixNano(), |
|||
baseTime.Add(20 * time.Minute).UnixNano(), |
|||
baseTime.Add(30 * time.Minute).UnixNano(), |
|||
} |
|||
|
|||
positions := make([]log_buffer.MessagePosition, len(timestamps)) |
|||
|
|||
for i, ts := range timestamps { |
|||
seekMsg := &mq_pb.SubscribeMessageRequest_SeekMessage{ |
|||
Offset: ts, |
|||
OffsetType: schema_pb.OffsetType_EXACT_TS_NS, |
|||
} |
|||
positions[i] = broker.getRequestPositionFromSeek(seekMsg) |
|||
} |
|||
|
|||
// Verify positions are in chronological order
|
|||
for i := 1; i < len(positions); i++ { |
|||
if !positions[i].Time.After(positions[i-1].Time) { |
|||
t.Errorf("Timestamp ordering violated: position[%d].Time (%v) should be after position[%d].Time (%v)", |
|||
i, positions[i].Time, i-1, positions[i-1].Time) |
|||
} |
|||
} |
|||
} |
|||
|
|||
// TestSeekByTimestamp_EdgeCases tests edge cases for timestamp seeks
|
|||
func TestSeekByTimestamp_EdgeCases(t *testing.T) { |
|||
broker := &MessageQueueBroker{} |
|||
|
|||
testCases := []struct { |
|||
name string |
|||
timestampNs int64 |
|||
expectValid bool |
|||
}{ |
|||
{ |
|||
name: "zero timestamp", |
|||
timestampNs: 0, |
|||
expectValid: true, // Valid - means Unix epoch
|
|||
}, |
|||
{ |
|||
name: "negative timestamp", |
|||
timestampNs: -1, |
|||
expectValid: true, // Valid in Go (before Unix epoch)
|
|||
}, |
|||
{ |
|||
name: "far future timestamp", |
|||
timestampNs: time.Now().Add(100 * 365 * 24 * time.Hour).UnixNano(), |
|||
expectValid: true, |
|||
}, |
|||
} |
|||
|
|||
for _, tc := range testCases { |
|||
t.Run(tc.name, func(t *testing.T) { |
|||
seekMsg := &mq_pb.SubscribeMessageRequest_SeekMessage{ |
|||
Offset: tc.timestampNs, |
|||
OffsetType: schema_pb.OffsetType_EXACT_TS_NS, |
|||
} |
|||
|
|||
position := broker.getRequestPositionFromSeek(seekMsg) |
|||
|
|||
if tc.expectValid { |
|||
expectedTime := time.Unix(0, tc.timestampNs) |
|||
if !position.Time.Equal(expectedTime) { |
|||
t.Errorf("Expected time %v, got %v", expectedTime, position.Time) |
|||
} |
|||
} |
|||
}) |
|||
} |
|||
} |
|||
|
|||
// TestSeekByTimestamp_VsOffset tests that timestamp and offset seeks are independent
|
|||
func TestSeekByTimestamp_VsOffset(t *testing.T) { |
|||
broker := &MessageQueueBroker{} |
|||
|
|||
timestampSeek := &mq_pb.SubscribeMessageRequest_SeekMessage{ |
|||
Offset: time.Now().UnixNano(), |
|||
OffsetType: schema_pb.OffsetType_EXACT_TS_NS, |
|||
} |
|||
|
|||
offsetSeek := &mq_pb.SubscribeMessageRequest_SeekMessage{ |
|||
Offset: 100, |
|||
OffsetType: schema_pb.OffsetType_EXACT_OFFSET, |
|||
} |
|||
|
|||
timestampPos := broker.getRequestPositionFromSeek(timestampSeek) |
|||
offsetPos := broker.getRequestPositionFromSeek(offsetSeek) |
|||
|
|||
// Timestamp-based position should have batch -2
|
|||
if timestampPos.Offset != -2 { |
|||
t.Errorf("Timestamp seek should have batch -2, got %d", timestampPos.Offset) |
|||
} |
|||
|
|||
// Offset-based position should have the exact offset in Offset field
|
|||
if offsetPos.GetOffset() != 100 { |
|||
t.Errorf("Offset seek should have offset 100, got %d", offsetPos.GetOffset()) |
|||
} |
|||
|
|||
// They should use different positioning mechanisms
|
|||
if timestampPos.IsOffsetBased { |
|||
t.Error("Timestamp seek should not be offset-based") |
|||
} |
|||
|
|||
if !offsetPos.IsOffsetBased { |
|||
t.Error("Offset seek should be offset-based") |
|||
} |
|||
} |
|||
|
|||
// TestSeekOptimization_SkipRedundantSeek tests that seeking to the same offset is optimized
|
|||
func TestSeekOptimization_SkipRedundantSeek(t *testing.T) { |
|||
broker := &MessageQueueBroker{} |
|||
|
|||
// Test that seeking to the same offset multiple times produces the same result
|
|||
seekMsg := &mq_pb.SubscribeMessageRequest_SeekMessage{ |
|||
Offset: 100, |
|||
OffsetType: schema_pb.OffsetType_EXACT_OFFSET, |
|||
} |
|||
|
|||
// First seek
|
|||
pos1 := broker.getRequestPositionFromSeek(seekMsg) |
|||
|
|||
// Second seek to same offset
|
|||
pos2 := broker.getRequestPositionFromSeek(seekMsg) |
|||
|
|||
// Third seek to same offset
|
|||
pos3 := broker.getRequestPositionFromSeek(seekMsg) |
|||
|
|||
// All positions should be identical
|
|||
if pos1.GetOffset() != pos2.GetOffset() || pos2.GetOffset() != pos3.GetOffset() { |
|||
t.Errorf("Multiple seeks to same offset should produce identical results: %d, %d, %d", |
|||
pos1.GetOffset(), pos2.GetOffset(), pos3.GetOffset()) |
|||
} |
|||
|
|||
// Verify the offset is correct
|
|||
if pos1.GetOffset() != 100 { |
|||
t.Errorf("Expected offset 100, got %d", pos1.GetOffset()) |
|||
} |
|||
} |
|||
|
|||
// TestSeekOptimization_DifferentOffsets tests that different offsets produce different positions
|
|||
func TestSeekOptimization_DifferentOffsets(t *testing.T) { |
|||
broker := &MessageQueueBroker{} |
|||
|
|||
offsets := []int64{0, 50, 100, 150, 200} |
|||
positions := make([]log_buffer.MessagePosition, len(offsets)) |
|||
|
|||
for i, offset := range offsets { |
|||
seekMsg := &mq_pb.SubscribeMessageRequest_SeekMessage{ |
|||
Offset: offset, |
|||
OffsetType: schema_pb.OffsetType_EXACT_OFFSET, |
|||
} |
|||
positions[i] = broker.getRequestPositionFromSeek(seekMsg) |
|||
} |
|||
|
|||
// Verify each position has the correct offset
|
|||
for i, offset := range offsets { |
|||
if positions[i].GetOffset() != offset { |
|||
t.Errorf("Position %d: expected offset %d, got %d", i, offset, positions[i].GetOffset()) |
|||
} |
|||
} |
|||
|
|||
// Verify all positions are different
|
|||
for i := 1; i < len(positions); i++ { |
|||
if positions[i].GetOffset() == positions[i-1].GetOffset() { |
|||
t.Errorf("Positions %d and %d should be different", i-1, i) |
|||
} |
|||
} |
|||
} |
|||
Write
Preview
Loading…
Cancel
Save
Reference in new issue