Merge remote-tracking branch 'upstream/main' into block-scheduler

pkg/blockbuilder/builder/controller.go

@@ -5,12 +5,12 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/grafana/loki/v3/pkg/blockbuilder/types"
-
-	"github.com/prometheus/prometheus/model/labels"
-
+	"github.com/go-kit/log"
+	"github.com/go-kit/log/level"
 	"github.com/grafana/dskit/backoff"
+	"github.com/prometheus/prometheus/model/labels"
 
+	"github.com/grafana/loki/v3/pkg/blockbuilder/types"
 	"github.com/grafana/loki/v3/pkg/kafka"
 	"github.com/grafana/loki/v3/pkg/kafka/partition"
 
@@ -202,10 +202,11 @@ func (l *PartitionJobController) LoadJob(ctx context.Context) (bool, *types.Job,
 
 	if highestOffset < committedOffset {
 		level.Error(l.logger).Log("msg", "partition highest offset is less than committed offset", "highest", highestOffset, "committed", committedOffset)
-		return false, Job{}, fmt.Errorf("partition highest offset is less than committed offset")
+		return false, nil, fmt.Errorf("partition highest offset is less than committed offset")
 	}
 
 	if highestOffset == committedOffset {
+		level.Info(l.logger).Log("msg", "no pending records to process")
 		return false, nil, nil
 	}