-
Notifications
You must be signed in to change notification settings - Fork 107
/
interface.go
98 lines (77 loc) · 3.99 KB
/
interface.go
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
package commitlog
// CommitLog is the durable write-ahead log interface used to back each stream.
type CommitLog interface {
// Delete closes the log and removes all data associated with it from the
// filesystem.
Delete() error
// NewReader creates a new Reader starting at the given offset. If
// uncommitted is true, the Reader will read uncommitted messages from the
// log. Otherwise, it will only return committed messages.
NewReader(offset int64, uncommitted bool) (*Reader, error)
// Truncate removes all messages from the log starting at the given offset.
Truncate(offset int64) error
// NewestOffset returns the offset of the last message in the log or -1 if
// empty.
NewestOffset() int64
// OldestOffset returns the offset of the first message in the log or -1 if
// empty.
OldestOffset() int64
// EarliestOffsetAfterTimestamp returns the earliest offset whose timestamp
// is greater than or equal to the given timestamp.
EarliestOffsetAfterTimestamp(timestamp int64) (int64, error)
// LatestOffsetBeforeTimestamp returns the latest offset whose timestamp is
// less than or equal to the given timestamp.
LatestOffsetBeforeTimestamp(timestamp int64) (int64, error)
// SetHighWatermark sets the high watermark on the log. All messages up to
// and including the high watermark are considered committed.
SetHighWatermark(hw int64)
// OverrideHighWatermark sets the high watermark on the log using the given
// value, even if the value is less than the current HW. This is used for
// unit testing purposes.
OverrideHighWatermark(hw int64)
// HighWatermark returns the high watermark for the log.
HighWatermark() int64
// NewLeaderEpoch indicates the log is entering a new leader epoch.
NewLeaderEpoch(epoch uint64) error
// LastOffsetForLeaderEpoch returns the start offset of the first leader
// epoch larger than the provided one or the log end offset if the current
// epoch equals the provided one.
LastOffsetForLeaderEpoch(epoch uint64) int64
// LastLeaderEpoch returns the latest leader epoch for the log.
LastLeaderEpoch() uint64
// Append writes the given batch of messages to the log and returns their
// corresponding offsets in the log. This will return ErrCommitLogReadonly
// if the log is in readonly mode.
Append(msg []*Message) ([]int64, error)
// AppendMessageSet writes the given message set data to the log and
// returns the corresponding offsets in the log. This can be called even if
// the log is in readonly mode to allow for reconciliation, e.g. when
// replicating from another log.
AppendMessageSet(ms []byte) ([]int64, error)
// Clean applies retention and compaction rules against the log, if
// applicable.
Clean() error
// NotifyLEO registers and returns a channel which is closed when messages
// past the given log end offset are added to the log. If the given offset
// is no longer the log end offset, the channel is closed immediately.
// Waiter is an opaque value that uniquely identifies the entity waiting
// for data.
NotifyLEO(waiter interface{}, leo int64) <-chan struct{}
// SetReadonly marks the log as readonly. When in readonly mode, new
// messages cannot be added to the log with Append and committed readers
// will read up to the log end offset (LEO), if the HW allows so, and then
// will receive an ErrCommitLogReadonly error. This will unblock committed
// readers waiting for data if they are at the LEO. Readers will continue
// to block if the HW is less than the LEO. This does not affect
// uncommitted readers. Messages can still be written to the log with
// AppendMessageSet for reconciliation purposes, e.g. when replicating from
// another log.
SetReadonly(readonly bool)
// IsReadonly indicates if the log is in readonly mode.
IsReadonly() bool
// IsConcurrencyControlEnabled indicates if the log should check for concurrency before appending messages
IsConcurrencyControlEnabled() bool
// Close closes each log segment file and stops the background goroutine
// checkpointing the high watermark to disk.
Close() error
}