fix build & tests.

pkg/blockbuilder/blockbuilder.go

@@ -108,7 +108,7 @@ func (b *BlockBuilder) makeSchedulerClient() (schedulerpb.SchedulerClient, error
 		b.logger,
 		b.cfg.SchedulerConfig.UpdateInterval,
 		b.cfg.SchedulerConfig.MaxUpdateAge,
-	), nil
+	)
 }
 
 func (b *BlockBuilder) starting(context.Context) (err error) {

pkg/blockbuilder/scheduler/scheduler.go

@@ -133,6 +133,7 @@ func (s *BlockBuilderScheduler) running(ctx context.Context) error {
 	}
 }
 
+// completeObservationMode transitions the scheduler from observation mode to normal operation.
 func (s *BlockBuilderScheduler) completeObservationMode() {
 	s.mu.Lock()
 	defer s.mu.Unlock()
@@ -331,7 +332,7 @@ func (s *BlockBuilderScheduler) updateJob(key jobKey, workerID string, complete
 	}
 
 	if c, ok := s.committed.Lookup(s.cfg.Kafka.Topic, j.Partition); ok {
-		if j.StartOffset <= c.At {
+		if j.EndOffset <= c.At {
 			// Update of a completed/committed job. Ignore.
 			level.Debug(logger).Log("msg", "ignored historical job")
 			return nil

pkg/blockbuilder/scheduler/scheduler_test.go

@@ -316,12 +316,12 @@ func TestOffsetMovement(t *testing.T) {
 	}
 	sched.completeObservationMode()
 
-	spec := jobSpec{
-		topic:       "ingest",
-		partition:   1,
-		commitRecTs: time.Unix(200, 0),
-		startOffset: 5000,
-		endOffset:   6000,
+	spec := schedulerpb.JobSpec{
+		Topic:       "ingest",
+		Partition:   1,
+		CommitRecTs: time.Unix(200, 0),
+		StartOffset: 5000,
+		EndOffset:   6000,
 	}
 
 	sched.jobs.addOrUpdate("ingest/1/5524", spec)