1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
|
package integration
import (
"testing"
"time"
)
// MockSeaweedClient provides a mock implementation for testing
type MockSeaweedClient struct {
records map[string]map[int32][]*SeaweedRecord // topic -> partition -> records
}
func NewMockSeaweedClient() *MockSeaweedClient {
return &MockSeaweedClient{
records: make(map[string]map[int32][]*SeaweedRecord),
}
}
func (m *MockSeaweedClient) AddRecord(topic string, partition int32, key []byte, value []byte, timestamp int64) {
if m.records[topic] == nil {
m.records[topic] = make(map[int32][]*SeaweedRecord)
}
if m.records[topic][partition] == nil {
m.records[topic][partition] = make([]*SeaweedRecord, 0)
}
record := &SeaweedRecord{
Key: key,
Value: value,
Timestamp: timestamp,
Offset: int64(len(m.records[topic][partition])), // Simple offset numbering
}
m.records[topic][partition] = append(m.records[topic][partition], record)
}
func (m *MockSeaweedClient) GetRecords(topic string, partition int32, fromOffset int64, maxRecords int) ([]*SeaweedRecord, error) {
if m.records[topic] == nil || m.records[topic][partition] == nil {
return nil, nil
}
allRecords := m.records[topic][partition]
if fromOffset < 0 || fromOffset >= int64(len(allRecords)) {
return nil, nil
}
endOffset := fromOffset + int64(maxRecords)
if endOffset > int64(len(allRecords)) {
endOffset = int64(len(allRecords))
}
return allRecords[fromOffset:endOffset], nil
}
func TestSeaweedSMQRecord_Interface(t *testing.T) {
// Test that SeaweedSMQRecord properly implements SMQRecord interface
key := []byte("test-key")
value := []byte("test-value")
timestamp := time.Now().UnixNano()
kafkaOffset := int64(42)
record := &SeaweedSMQRecord{
key: key,
value: value,
timestamp: timestamp,
offset: kafkaOffset,
}
// Test interface compliance
var smqRecord SMQRecord = record
// Test GetKey
if string(smqRecord.GetKey()) != string(key) {
t.Errorf("Expected key %s, got %s", string(key), string(smqRecord.GetKey()))
}
// Test GetValue
if string(smqRecord.GetValue()) != string(value) {
t.Errorf("Expected value %s, got %s", string(value), string(smqRecord.GetValue()))
}
// Test GetTimestamp
if smqRecord.GetTimestamp() != timestamp {
t.Errorf("Expected timestamp %d, got %d", timestamp, smqRecord.GetTimestamp())
}
// Test GetOffset
if smqRecord.GetOffset() != kafkaOffset {
t.Errorf("Expected offset %d, got %d", kafkaOffset, smqRecord.GetOffset())
}
}
func TestSeaweedMQHandler_GetStoredRecords_EmptyTopic(t *testing.T) {
// Note: Ledgers have been removed - SMQ broker handles all offset management directly
// This test is now obsolete as GetStoredRecords requires a real broker connection
t.Skip("Test obsolete: ledgers removed, SMQ broker handles offset management")
}
func TestSeaweedMQHandler_GetStoredRecords_EmptyPartition(t *testing.T) {
// Note: Ledgers have been removed - SMQ broker handles all offset management directly
// This test is now obsolete as GetStoredRecords requires a real broker connection
t.Skip("Test obsolete: ledgers removed, SMQ broker handles offset management")
}
func TestSeaweedMQHandler_GetStoredRecords_OffsetBeyondHighWaterMark(t *testing.T) {
// Note: Ledgers have been removed - SMQ broker handles all offset management directly
// This test is now obsolete as GetStoredRecords requires a real broker connection
t.Skip("Test obsolete: ledgers removed, SMQ broker handles offset management")
}
func TestSeaweedMQHandler_GetStoredRecords_MaxRecordsLimit(t *testing.T) {
// Note: Ledgers have been removed - SMQ broker handles all offset management directly
// This test is now obsolete as GetStoredRecords requires a real broker connection
t.Skip("Test obsolete: ledgers removed, SMQ broker handles offset management")
}
// Integration test helpers and benchmarks
func BenchmarkSeaweedSMQRecord_GetMethods(b *testing.B) {
record := &SeaweedSMQRecord{
key: []byte("benchmark-key"),
value: []byte("benchmark-value-with-some-longer-content"),
timestamp: time.Now().UnixNano(),
offset: 12345,
}
b.ResetTimer()
b.Run("GetKey", func(b *testing.B) {
for i := 0; i < b.N; i++ {
_ = record.GetKey()
}
})
b.Run("GetValue", func(b *testing.B) {
for i := 0; i < b.N; i++ {
_ = record.GetValue()
}
})
b.Run("GetTimestamp", func(b *testing.B) {
for i := 0; i < b.N; i++ {
_ = record.GetTimestamp()
}
})
b.Run("GetOffset", func(b *testing.B) {
for i := 0; i < b.N; i++ {
_ = record.GetOffset()
}
})
}
|
