Repository: liftbridge-io/liftbridge (go)

Issue:
api error: Failed to subscribe to partition
Subscribe() to a stream fails with the following error:

`Failed to lookup offset for timestamp: failed to find log segment for timestamp: EOF`

This happens only if I try to subscribe with StartAtTime(). However, subscribe with StartAtOffset() works OK.
The issue arises after a few days of the stream inactivity (no incoming events) and the subscriber is restarted.

This is a 3 node cluster running on kubernetes.
Image: `liftbridge/liftbridge-k8s:v1.4.1`

However, I did experience the same issue with a local docker container running a single node server based on the latest liftbridge/standalone-dev image (v.1.5.0).

These are kubernetes deployment details.

Config is pretty basic:
```
listen: 0.0.0.0:9292
logging.level: info

nats.servers:
  - "nats://user:secret@nats-client:4222"

clustering.min.insync.replicas: 2
```

The error is logged in liftbridge-1 log:

`time="2021-01-26 15:34:37" level=error msg="api: Failed to subscribe to partition [subject=ems.helb.sensors, stream=ems.helb.sensors, partition=0]: rpc error: code = Internal desc = Failed to lookup offset for timestamp: failed to find log segment for timestamp: EOF"`


liftbridge-0 log:
```
time="2021-01-19 05:59:31" level=warning msg="fsm: Removed replica cluster-liftbridge-0 from ISR for partition [subject=ems.helb.sensors, stream=ems.helb.sensors, partition=0]"
time="2021-01-19 05:59:31" level=error msg="ISR for partition [subject=ems.helb.sensors, stream=ems.helb.sensors, partition=0] has shrunk below minimum size 2, currently 1"
time="2021-01-19 05:59:31" level=warning msg="fsm: Removed replica cluster-liftbridge-2 from ISR for partition [subject=ems.helb.sensors, stream=ems.helb.sensors, partition=0]"
time="2021-01-19 05:59:46" level=info msg="ISR for partition [subject=ems.helb.sensors, stream=ems.helb.sensors, partition=0] has recovered from being below minimum size 2, currently 2"
time="2021-01-19 05:59:46" level=info msg="fsm: Added replica cluster-liftbridge-0 to ISR for partition [subject=ems.helb.sensors, stream=ems.helb.sensors, partition=0]"
time="2021-01-19 05:59:46" level=info msg="fsm: Added replica cluster-liftbridge-2 to ISR for partition [subject=ems.helb.sensors, stream=ems.helb.sensors, partition=0]"
```

liftbridge-1 log:
```
time="2021-01-19 05:59:08" level=error msg="Replica cluster-liftbridge-0 for partition [subject=ems.helb.sensors, stream=ems.helb.sensors, partition=0] exceeded max lag time (last seen: 15.000115667s, last caught up: 15.000115667s), removing from ISR"
time="2021-01-19 05:59:18" level=error msg="Replica cluster-liftbridge-2 for partition [subject=ems.helb.sensors, stream=ems.helb.sensors, partition=0] exceeded max lag time (last seen: 15.000100705s, last caught up: 15.000100705s), removing from ISR"
time="2021-01-19 05:59:31" level=warning msg="fsm: Removed replica cluster-liftbridge-0 from ISR for partition [subject=ems.helb.sensors, stream=ems.helb.sensors, partition=0]"
time="2021-01-19 05:59:31" level=error msg="ISR for partition [subject=ems.helb.sensors, stream=ems.helb.sensors, partition=0] has shrunk below minimum size 2, currently 1"
time="2021-01-19 05:59:31" level=warning msg="fsm: Removed replica cluster-liftbridge-2 from ISR for partition [subject=ems.helb.sensors, stream=ems.helb.sensors, partition=0]"
time="2021-01-19 05:59:33" level=error msg="Unable to commit messages for partition [subject=ems.helb.sensors, stream=ems.helb.sensors, partition=0], ISR size (1) below minimum (2)"
time="2021-01-19 05:59:46" level=info msg="Replica cluster-liftbridge-0 for partition [subject=ems.helb.sensors, stream=ems.helb.sensors, partition=0] caught back up with leader, rejoining ISR"
time="2021-01-19 05:59:46" level=info msg="Replica cluster-liftbridge-2 for partition [subject=ems.helb.sensors, stream=ems.helb.sensors, partition=0] caught back up with leader, rejoining ISR"
time="2021-01-19 05:59:46" level=info msg="ISR for partition [subject=ems.helb.sensors, stream=ems.helb.sensors, partition=0] has recovered from being below minimum size 2, currently 2"
time="2021-01-19 05:59:46" level=info msg="fsm: Added replica cluster-liftbridge-0 to ISR for partition [subject=ems.helb.sensors, stream=ems.helb.sensors, partition=0]"
time="2021-01-19 05:59:46" level=info msg="fsm: Added replica cluster-liftbridge-2 to ISR for partition [subject=ems.helb.sensors, stream=ems.helb.sensors, partition=0]"
time="2021-01-26 15:34:37" level=error msg="api: Failed to subscribe to partition [subject=ems.helb.sensors, stream=ems.helb.sensors, partition=0]: rpc error: code = Internal desc = Failed to lookup offset for timestamp: failed to find log segment for timestamp: EOF"
```

liftbridge-2 log:
```
time="2021-01-19 05:59:31" level=warning msg="fsm: Removed replica cluster-liftbridge-0 from ISR for partition [subject=ems.helb.sensors, stream=ems.helb.sensors, partition=0]"
time="2021-01-19 05:59:31" level=error msg="ISR for partition [subject=ems.helb.sensors, stream=ems.helb.sensors, partition=0] has shrunk below minimum size 2, currently 1"
time="2021-01-19 05:59:31" level=warning msg="fsm: Removed replica cluster-liftbridge-2 from ISR for partition [subject=ems.helb.sensors, stream=ems.helb.sensors, partition=0]"
time="2021-01-19 05:59:46" level=info msg="ISR for partition [subject=ems.helb.sensors, stream=ems.helb.sensors, partition=0] has recovered from being below minimum size 2, currently 2"
time="2021-01-19 05:59:46" level=info msg="fsm: Added replica cluster-liftbridge-0 to ISR for partition [subject=ems.helb.sensors, stream=ems.helb.sensors, partition=0]"
time="2021-01-19 05:59:46" level=info msg="fsm: Added replica cluster-liftbridge-2 to ISR for partition [subject=ems.helb.sensors, stream=ems.helb.sensors, partition=0]"
```

Pull request description:
Fix StartAtTime EOF error
Fix EOF error that occurs when using StartAtTime to subscribe to a
partition that is empty or otherwise has a timestamp that is beyond the
end of the log.

Fixes #323

Patch:
diff --git a/server/commitlog/commitlog.go b/server/commitlog/commitlog.go
index abafaac..e970f71 100644
--- a/server/commitlog/commitlog.go
+++ b/server/commitlog/commitlog.go
@@ -299,6 +299,11 @@ func (l *commitLog) EarliestOffsetAfterTimestamp(timestamp int64) (int64, error)
 	// Find the first segment whose base timestamp is greater than the given
 	// timestamp.
 	idx, err := findSegmentIndexByTimestamp(l.segments, timestamp)
+	if err == io.EOF {
+		// EOF indicates there is no such segment, meaning the timestamp is
+		// beyond the end of the log so return the next assignable offset.
+		return l.segments[len(l.segments)-1].NextOffset(), nil
+	}
 	if err != nil {
 		return 0, errors.Wrap(err, "failed to find log segment for timestamp")
 	}
@@ -321,8 +326,8 @@ func (l *commitLog) EarliestOffsetAfterTimestamp(timestamp int64) (int64, error)
 	// This indicates there are no entries in the segment whose timestamp
 	// is greater than or equal to the target timestamp. In this case, search
 	// the next segment if there is one. If there isn't, the timestamp is
-	// beyond the end of the log so return the next offset.
-	if idx < len(l.segments) {
+	// beyond the end of the log so return the next assignable offset.
+	if idx < len(l.segments)-1 {
 		seg = l.segments[idx]
 		entry, err := seg.findEntryByTimestamp(timestamp)
 		if err != nil {


Test patch:
diff --git a/server/commitlog/commitlog_test.go b/server/commitlog/commitlog_test.go
index 82cd760..e4786f0 100644
--- a/server/commitlog/commitlog_test.go
+++ b/server/commitlog/commitlog_test.go
@@ -372,7 +372,7 @@ func TestCleanerReplaceLeaderEpochOffsets(t *testing.T) {
 	require.Equal(t, int64(14), l.LastOffsetForLeaderEpoch(3))
 }
 
-// Ensures EarliestOffsetAfterTimestamp returns the earliest offset whose
+// Ensure EarliestOffsetAfterTimestamp returns the earliest offset whose
 // timestamp is greater than or equal to the given timestamp.
 func TestEarliestOffsetAfterTimestamp(t *testing.T) {
 	opts := Options{
@@ -425,7 +425,22 @@ func TestEarliestOffsetAfterTimestamp(t *testing.T) {
 	require.Equal(t, int64(3), offset)
 }
 
-// Ensures LatestOffsetBeforeTimestamp returns the latest offset whose
+// Ensure EarliestOffsetAfterTimestamp returns the next assignable offset
+// when the log is empty.
+func TestEarliestOffsetAfterTimestampEmptyLog(t *testing.T) {
+	opts := Options{
+		Path:            tempDir(t),
+		MaxSegmentBytes: 100,
+	}
+	l, cleanup := setupWithOptions(t, opts)
+	defer cleanup()
+
+	offset, err := l.EarliestOffsetAfterTimestamp(100)
+	require.NoError(t, err)
+	require.Equal(t, int64(0), offset)
+}
+
+// Ensure LatestOffsetBeforeTimestamp returns the latest offset whose
 // timestamp is less than or equal to the given timestamp.
 func TestLatestOffsetBeforeTimestamp(t *testing.T) {
 	opts := Options{