Delete old code for consuming multiple topics

Author: grobinson-grafana

pkg/dataobj/consumer/consumer.go

@@ -3,7 +3,6 @@ package consumer
 import (
 	"context"
 	"errors"
-	"fmt"
 	"sync"
 	"time"
 
@@ -18,47 +17,26 @@ type kafkaConsumer interface {
 	PollFetches(context.Context) kgo.Fetches
 }
 
+// processor allows mocking of [partitionProcessor] in tests.
+type processor interface {
+	Append(records []*kgo.Record) bool
+}
+
 // consumer polls records from the Kafka topic and passes each record to
 // its indended processor.
 type consumer struct {
-	client     kafkaConsumer
-	logger     log.Logger
-	processors map[string]map[int32]processor
-	mtx        sync.RWMutex
+	client    kafkaConsumer
+	logger    log.Logger
+	processor processor
+	mtx       sync.RWMutex
 }
 
 // newConsumer returns a new consumer.
-func newConsumer(client kafkaConsumer, logger log.Logger) *consumer {
+func newConsumer(client kafkaConsumer, processor processor, logger log.Logger) *consumer {
 	return &consumer{
-		client:     client,
-		logger:     logger,
-		processors: make(map[string]map[int32]processor),
-	}
-}
-
-// OnRegister implements the [partitionProcessorListener] interface.
-func (c *consumer) OnRegister(topic string, partition int32, p processor) {
-	c.mtx.Lock()
-	defer c.mtx.Unlock()
-	processorsByTopic, ok := c.processors[topic]
-	if !ok {
-		processorsByTopic = make(map[int32]processor)
-		c.processors[topic] = processorsByTopic
-	}
-	processorsByTopic[partition] = p
-}
-
-// OnDeregister implements the [partitionProcessorListener] interface.
-func (c *consumer) OnDeregister(topic string, partition int32) {
-	c.mtx.Lock()
-	defer c.mtx.Unlock()
-	processorsByTopic, ok := c.processors[topic]
-	if !ok {
-		return
-	}
-	delete(processorsByTopic, partition)
-	if len(processorsByTopic) == 0 {
-		delete(c.processors, topic)
+		client:    client,
+		logger:    logger,
+		processor: processor,
 	}
 }
 
@@ -96,56 +74,7 @@ func (c *consumer) pollFetches(ctx context.Context) error {
 		if errors.Is(err, kgo.ErrClientClosed) || errors.Is(err, context.Canceled) {
 			return err
 		}
-		// Some other error occurred. We will check it in
-		// [processFetchTopicPartition] instead.
 	}
-	fetches.EachPartition(c.processFetchTopicPartition(ctx))
+	c.processor.Append(fetches.Records())
 	return nil
 }
-
-func (c *consumer) processFetchTopicPartition(_ context.Context) func(kgo.FetchTopicPartition) {
-	return func(fetch kgo.FetchTopicPartition) {
-		if err := fetch.Err; err != nil {
-			level.Error(c.logger).Log("msg", "failed to fetch records for topic partition", "topic", fetch.Topic, "partition", fetch.Partition, "err", err.Error())
-			return
-		}
-		// If there are no records for this partition then skip it.
-		if len(fetch.Records) == 0 {
-			return
-		}
-		processor, err := c.processorForTopicPartition(fetch.Topic, fetch.Partition)
-		if err != nil {
-			// It should never happen that we fetch records for a newly
-			// assigned partition before the lifecycler has registered a
-			// processor for it. This is because [kgo.OnPartitionsAssigned]
-			// guarantees to return before the client starts fetching records
-			// for new partitions.
-			//
-			// However, it can happen the client has fetched records for a
-			// partition that has just been reassigned to another consumer.
-			// If this happens, we will attempt to process those records, but
-			// may not have a processor for them as the processor would have
-			// been deregistered via [kgo.OnPartitionsRevoked], and the
-			// following log line will be emitted.
-			level.Error(c.logger).Log("msg", "failed to get processor", "error", err.Error())
-			return
-		}
-		_ = processor.Append(fetch.Records)
-	}
-}
-
-// processorForTopicPartition returns the processor for the topic and partition.
-// It returns an error if one does not exist.
-func (c *consumer) processorForTopicPartition(topic string, partition int32) (processor, error) {
-	c.mtx.RLock()
-	defer c.mtx.RUnlock()
-	processorsByTopic, ok := c.processors[topic]
-	if !ok {
-		return nil, fmt.Errorf("unknown topic %s", topic)
-	}
-	p, ok := processorsByTopic[partition]
-	if !ok {
-		return nil, fmt.Errorf("unknown partition %d for topic %s", partition, topic)
-	}
-	return p, nil
-}

pkg/dataobj/consumer/consumer_test.go

@@ -1,64 +1,3 @@
 package consumer
 
-import (
-	"testing"
-
-	"github.com/go-kit/log"
-	"github.com/stretchr/testify/require"
-)
-
-func TestConsumer_OnRegister(t *testing.T) {
-	c := newConsumer(&mockKafka{}, log.NewNopLogger())
-	require.Empty(t, c.processors)
-	// Register a processor for the topic.
-	p1 := &partitionProcessor{}
-	c.OnRegister("topic1", 1, p1)
-	require.NotNil(t, c.processors["topic1"])
-	require.Same(t, p1, c.processors["topic1"][1])
-	// Register another processor for the same topic.
-	p2 := &partitionProcessor{}
-	c.OnRegister("topic1", 2, p2)
-	require.Len(t, c.processors["topic1"], 2)
-	require.Same(t, p1, c.processors["topic1"][1])
-	require.Same(t, p2, c.processors["topic1"][2])
-	// Register another processor for a different topic.
-	p3 := &partitionProcessor{}
-	c.OnRegister("topic2", 1, p3)
-	require.NotNil(t, c.processors["topic1"])
-	require.Len(t, c.processors["topic1"], 2)
-	require.Same(t, p1, c.processors["topic1"][1])
-	require.Same(t, p2, c.processors["topic1"][2])
-	require.Len(t, c.processors["topic2"], 1)
-	require.Same(t, p3, c.processors["topic2"][1])
-}
-
-func TestConsumer_OnDeregister(t *testing.T) {
-	c := newConsumer(&mockKafka{}, log.NewNopLogger())
-	require.Empty(t, c.processors)
-	// Register a bunch of processors for different topics.
-	p1 := &partitionProcessor{}
-	c.OnRegister("topic1", 1, p1)
-	p2 := &partitionProcessor{}
-	c.OnRegister("topic1", 2, p2)
-	require.Len(t, c.processors["topic1"], 2)
-	p3 := &partitionProcessor{}
-	c.OnRegister("topic2", 1, p3)
-	// Check all the processors are as expected.
-	require.NotNil(t, c.processors["topic1"])
-	require.Len(t, c.processors["topic1"], 2)
-	require.Same(t, p1, c.processors["topic1"][1])
-	require.Same(t, p2, c.processors["topic1"][2])
-	require.Len(t, c.processors["topic2"], 1)
-	require.Same(t, p3, c.processors["topic2"][1])
-	// Now deregister one and check the remaining processors.
-	c.OnDeregister("topic1", 1)
-	require.Len(t, c.processors["topic1"], 1)
-	require.Nil(t, c.processors["topic1"][1])
-	require.Same(t, p2, c.processors["topic1"][2])
-	// Now deregister the last one and check the topic is removed.
-	c.OnDeregister("topic1", 2)
-	require.Nil(t, c.processors["topic1"])
-	// The other topic should be unmodified.
-	require.Len(t, c.processors["topic2"], 1)
-	require.Same(t, p3, c.processors["topic2"][1])
-}
+// TODO(grobinson): Write tests for the consumer.

pkg/dataobj/consumer/mock_test.go

@@ -227,63 +227,6 @@ func (m *mockPartitionProcessorFactory) New(_ context.Context, _ *kgo.Client, _
 	return &mockPartitionProcessor{}
 }
 
-type mockPartitionProcessorListener struct {
-	processors map[string]map[int32]processor
-}
-
-func (m *mockPartitionProcessorListener) OnRegister(topic string, partition int32, p processor) {
-	if m.processors == nil {
-		m.processors = make(map[string]map[int32]processor)
-	}
-	processorsByTopic, ok := m.processors[topic]
-	if !ok {
-		processorsByTopic = make(map[int32]processor)
-		m.processors[topic] = processorsByTopic
-	}
-	processorsByTopic[partition] = p
-}
-func (m *mockPartitionProcessorListener) OnDeregister(topic string, partition int32) {
-	processorsByTopic, ok := m.processors[topic]
-	if !ok {
-		return
-	}
-	delete(processorsByTopic, partition)
-	if len(processorsByTopic) == 0 {
-		delete(m.processors, topic)
-	}
-}
-
-// mockPartitionProcessorLifecycler mocks a [partitionProcessorLifecycler].
-type mockPartitionProcessorLifecycler struct {
-	processors map[string]map[int32]struct{}
-}
-
-func (m *mockPartitionProcessorLifecycler) Register(_ context.Context, _ *kgo.Client, topic string, partition int32) {
-	if m.processors == nil {
-		m.processors = make(map[string]map[int32]struct{})
-	}
-	processorsByTopic, ok := m.processors[topic]
-	if !ok {
-		processorsByTopic = make(map[int32]struct{})
-		m.processors[topic] = processorsByTopic
-	}
-	processorsByTopic[partition] = struct{}{}
-}
-func (m *mockPartitionProcessorLifecycler) Deregister(_ context.Context, topic string, partition int32) {
-	if m.processors == nil {
-		return
-	}
-	processorsByTopic, ok := m.processors[topic]
-	if !ok {
-		return
-	}
-	delete(processorsByTopic, partition)
-	if len(processorsByTopic) == 0 {
-		delete(m.processors, topic)
-	}
-}
-func (m *mockPartitionProcessorLifecycler) Stop(_ context.Context) {}
-
 type recordedTocEntry struct {
 	DataObjectPath string
 	MinTimestamp   time.Time