fix: undo builder rename in config file

Author: grobinson-grafana

docs/sources/shared/configuration.md

@@ -1080,7 +1080,7 @@ kafka_config:
 
 dataobj:
   consumer:
-    builder:
+    builderconfig:
       # The target maximum amount of uncompressed data to hold in data pages
       # (for columnar sections). Uncompressed size is used for consistent I/O
       # and planning.

pkg/dataobj/consumer/config.go

@@ -13,7 +13,7 @@ import (
 )
 
 type Config struct {
-	BuilderConfig       logsobj.BuilderConfig `yaml:"builder,omitempty"`
+	logsobj.BuilderConfig
 	LifecyclerConfig    ring.LifecyclerConfig `yaml:"lifecycler,omitempty"`
 	PartitionRingConfig partitionring.Config  `yaml:"partition_ring" category:"experimental"`
 	UploaderConfig      uploader.Config       `yaml:"uploader"`

pkg/dataobj/consumer/mock_test.go

@@ -16,6 +16,7 @@ import (
 	"github.com/grafana/loki/v3/pkg/dataobj/consumer/logsobj"
 	"github.com/grafana/loki/v3/pkg/dataobj/metastore"
 	"github.com/grafana/loki/v3/pkg/dataobj/metastore/multitenancy"
+	"github.com/grafana/loki/v3/pkg/kafka/partition"
 	"github.com/grafana/loki/v3/pkg/logproto"
 )
 
@@ -133,13 +134,11 @@ func (m *mockBuilder) UnregisterMetrics(r prometheus.Registerer) {
 
 // A mockCommitter implements the committer interface for tests.
 type mockCommitter struct {
-	// We will need to change this when we add support for other methods like
-	// CommitOffsets and CommitOffsetsSync.
-	records []*kgo.Record
+	offsets []int64
 }
 
-func (m *mockCommitter) CommitRecords(_ context.Context, records ...*kgo.Record) error {
-	m.records = append(m.records, records...)
+func (m *mockCommitter) Commit(ctx context.Context, offset int64) error {
+	m.offsets = append(m.offsets, offset)
 	return nil
 }
 
@@ -200,11 +199,11 @@ func (m *mockKafka) ProduceSync(_ context.Context, rs ...*kgo.Record) kgo.Produc
 
 // mockPartitionProcessor mocks a [partitionProcessor].
 type mockPartitionProcessor struct {
-	records          []*kgo.Record
+	records          []partition.Record
 	started, stopped bool
 }
 
-func (m *mockPartitionProcessor) Append(records []*kgo.Record) bool {
+func (m *mockPartitionProcessor) Append(records []partition.Record) bool {
 	m.records = append(m.records, records...)
 	return true
 }

pkg/dataobj/consumer/partition_processor_test.go

@@ -10,11 +10,11 @@ import (
 	"github.com/go-kit/log"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/stretchr/testify/require"
-	"github.com/twmb/franz-go/pkg/kgo"
 
 	"github.com/grafana/loki/v3/pkg/dataobj/consumer/logsobj"
 	"github.com/grafana/loki/v3/pkg/dataobj/metastore"
 	"github.com/grafana/loki/v3/pkg/dataobj/uploader"
+	"github.com/grafana/loki/v3/pkg/kafka/partition"
 	"github.com/grafana/loki/v3/pkg/logproto"
 	"github.com/grafana/loki/v3/pkg/scratch"
 
@@ -49,9 +49,9 @@ func TestPartitionProcessor_Flush(t *testing.T) {
 		}
 		b, err := s.Marshal()
 		require.NoError(t, err)
-		p.processRecord(&kgo.Record{
-			Key:       []byte("test-tenant"),
-			Value:     b,
+		p.processRecord(partition.Record{
+			TenantID:  "test-tenant",
+			Content:   b,
 			Timestamp: now,
 		})
 
@@ -90,9 +90,9 @@ func TestPartitionProcessor_Flush(t *testing.T) {
 		}
 		b, err := s.Marshal()
 		require.NoError(t, err)
-		p.processRecord(&kgo.Record{
-			Key:       []byte("test-tenant"),
-			Value:     b,
+		p.processRecord(partition.Record{
+			TenantID:  "test-tenant",
+			Content:   b,
 			Timestamp: now,
 		})
 
@@ -144,9 +144,9 @@ func TestPartitionProcessor_IdleFlush(t *testing.T) {
 	}
 	b, err := s.Marshal()
 	require.NoError(t, err)
-	p.processRecord(&kgo.Record{
-		Key:       []byte("test-tenant"),
-		Value:     b,
+	p.processRecord(partition.Record{
+		TenantID:  "test-tenant",
+		Content:   b,
 		Timestamp: clock.Now(),
 	})
 	// A modification should have happened.
@@ -189,36 +189,36 @@ func TestPartitionProcessor_OffsetsCommitted(t *testing.T) {
 		}
 		b, err := s.Marshal()
 		require.NoError(t, err)
-		p.processRecord(&kgo.Record{
-			Key:       []byte("test-tenant"),
-			Value:     b,
+		p.processRecord(partition.Record{
+			TenantID:  "test-tenant",
+			Content:   b,
 			Timestamp: now1,
 			Offset:    1,
 		})
 
 		// No flush should have occurred and no offsets should be committed.
 		require.True(t, p.lastFlushed.IsZero())
-		require.Nil(t, committer.records)
+		require.Nil(t, committer.offsets)
 
 		// Mark the builder as full.
 		wrappedBuilder.nextErr = logsobj.ErrBuilderFull
 
 		// Append another record.
 		clock.Advance(time.Minute)
 		now2 := clock.Now()
-		p.processRecord(&kgo.Record{
-			Key:       []byte("test-tenant"),
-			Value:     b,
+		p.processRecord(partition.Record{
+			TenantID:  "test-tenant",
+			Content:   b,
 			Timestamp: now2,
 			Offset:    2,
 		})
 
 		// A flush should have occurred and offsets should be committed.
 		require.Equal(t, now2, p.lastFlushed)
-		require.Len(t, committer.records, 1)
+		require.Len(t, committer.offsets, 1)
 		// The offset committed should be the offset of the first record, as that
 		// was the record that was flushed.
-		require.Equal(t, int64(1), committer.records[0].Offset)
+		require.Equal(t, int64(1), committer.offsets[0])
 	})
 
 	t.Run("when idle timeout is exceeded", func(t *testing.T) {
@@ -240,16 +240,16 @@ func TestPartitionProcessor_OffsetsCommitted(t *testing.T) {
 		}
 		b, err := s.Marshal()
 		require.NoError(t, err)
-		p.processRecord(&kgo.Record{
-			Key:       []byte("test-tenant"),
-			Value:     b,
+		p.processRecord(partition.Record{
+			TenantID:  "test-tenant",
+			Content:   b,
 			Timestamp: now1,
 			Offset:    1,
 		})
 
 		// No flush should have occurred and no offsets should be committed.
 		require.True(t, p.lastFlushed.IsZero())
-		require.Nil(t, committer.records)
+		require.Nil(t, committer.offsets)
 
 		// Advance the clock past the idle timeout.
 		clock.Advance(61 * time.Minute)
@@ -260,11 +260,11 @@ func TestPartitionProcessor_OffsetsCommitted(t *testing.T) {
 
 		// A flush should have occurred and offsets should be committed.
 		require.Equal(t, now2, p.lastFlushed)
-		require.Len(t, committer.records, 1)
+		require.Len(t, committer.offsets, 1)
 
 		// The offset committed should be the offset of the first record, as that
 		// was the record that was flushed.
-		require.Equal(t, int64(1), committer.records[0].Offset)
+		require.Equal(t, int64(1), committer.offsets[0])
 	})
 }
 
@@ -286,9 +286,9 @@ func TestPartitionProcessor_ProcessRecord(t *testing.T) {
 	}
 	b, err := s.Marshal()
 	require.NoError(t, err)
-	p.processRecord(&kgo.Record{
-		Key:       []byte("test-tenant"),
-		Value:     b,
+	p.processRecord(partition.Record{
+		TenantID:  "test-tenant",
+		Content:   b,
 		Timestamp: clock.Now(),
 	})
 
@@ -300,16 +300,14 @@ func TestPartitionProcessor_ProcessRecord(t *testing.T) {
 func newTestPartitionProcessor(_ *testing.T, clock quartz.Clock) *partitionProcessor {
 	p := newPartitionProcessor(
 		context.Background(),
-		&kgo.Client{},
+		&mockCommitter{},
 		testBuilderConfig,
 		uploader.Config{},
 		metastore.Config{
 			PartitionRatio: 10,
 		},
 		newMockBucket(),
 		nil,
-		"topic",
-		0,
 		log.NewNopLogger(),
 		prometheus.NewRegistry(),
 		60*time.Minute,