ensure montonically increasing tsNs
This commit is contained in:
@@ -34,6 +34,7 @@ type LogBuffer struct {
|
|||||||
notifyFn func()
|
notifyFn func()
|
||||||
isStopping bool
|
isStopping bool
|
||||||
flushChan chan *dataToFlush
|
flushChan chan *dataToFlush
|
||||||
|
lastTsNs int64
|
||||||
sync.RWMutex
|
sync.RWMutex
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -64,8 +65,15 @@ func (m *LogBuffer) AddToBuffer(partitionKey, data []byte) {
|
|||||||
|
|
||||||
// need to put the timestamp inside the lock
|
// need to put the timestamp inside the lock
|
||||||
ts := time.Now()
|
ts := time.Now()
|
||||||
|
tsNs := ts.UnixNano()
|
||||||
|
if m.lastTsNs >= tsNs {
|
||||||
|
// this is unlikely to happen, but just in case
|
||||||
|
tsNs = m.lastTsNs + 1
|
||||||
|
ts = time.Unix(0, tsNs)
|
||||||
|
}
|
||||||
|
m.lastTsNs = tsNs
|
||||||
logEntry := &filer_pb.LogEntry{
|
logEntry := &filer_pb.LogEntry{
|
||||||
TsNs: ts.UnixNano(),
|
TsNs: tsNs,
|
||||||
PartitionKeyHash: util.HashToInt32(partitionKey),
|
PartitionKeyHash: util.HashToInt32(partitionKey),
|
||||||
Data: data,
|
Data: data,
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user