diff --git a/platform/consumer/consumer.go b/platform/consumer/consumer.go index f3dcdbb4d..c7888d264 100644 --- a/platform/consumer/consumer.go +++ b/platform/consumer/consumer.go @@ -225,7 +225,7 @@ func (m *consumer) subscribe(ctx context.Context, controller Controller) error { m.subscriptions[topicKey] = sub // Spawn consumption goroutine - go m.consumeLoop(controllerCtx, controller, deliveryChan, done, config.BatchSize) + go m.consumeLoop(controllerCtx, controller, topicName, deliveryChan, done, config.BatchSize) m.logger.Infow("controller started", "controller", controller.Name(), @@ -258,13 +258,14 @@ func (m *consumer) subscribe(ctx context.Context, controller Controller) error { // Any messages buffered in partition channels but not processed before ctx // cancellation are safe to drop — the queue's visibility timeout will make // them visible again for redelivery (at-least-once semantics). -func (m *consumer) consumeLoop(ctx context.Context, controller Controller, deliveryChan <-chan extqueue.Delivery, done chan struct{}, batchSize int) { +func (m *consumer) consumeLoop(ctx context.Context, controller Controller, topicName string, deliveryChan <-chan extqueue.Delivery, done chan struct{}, batchSize int) { defer close(done) topicKey := controller.TopicKey() controllerScope := m.metricsScope.Tagged(map[string]string{ "controller": controller.Name(), + "topic": topicName, "topic_key": topicKey.String(), }) diff --git a/platform/consumer/consumer_test.go b/platform/consumer/consumer_test.go index 6e2eb5d63..0480db1d7 100644 --- a/platform/consumer/consumer_test.go +++ b/platform/consumer/consumer_test.go @@ -642,6 +642,8 @@ func TestConsumer_Stop(t *testing.T) { } func TestConsumer_ObservabilityTags(t *testing.T) { + const topicName = "configured-test-topic" + tests := []struct { name string handlerError error @@ -712,7 +714,15 @@ func TestConsumer_ObservabilityTags(t *testing.T) { mockQ := queuemock.NewMockQueue(ctrl) mockQ.EXPECT().Subscriber().Return(mockSub) - reg := newRegistry(t, mockQ, testTopicKeyStart, "test-group") + reg, err := NewTopicRegistry([]TopicConfig{ + { + Key: testTopicKeyStart, + Name: topicName, + Queue: mockQ, + Subscription: extqueue.DefaultSubscriptionConfig("test-worker", "test-group"), + }, + }) + require.NoError(t, err) testC := New(logger, testScope, reg, tt.processor, consumergatenoop.New()) @@ -723,7 +733,7 @@ func TestConsumer_ObservabilityTags(t *testing.T) { }, ) - err := testC.Register(handler) + err = testC.Register(handler) require.NoError(t, err) ctx, cancel := context.WithCancel(context.Background()) @@ -753,6 +763,8 @@ func TestConsumer_ObservabilityTags(t *testing.T) { for key, value := range tt.expectedTags { assert.Equal(t, value, tags[key]) } + assert.Equal(t, topicName, tags["topic"]) + assert.Equal(t, testTopicKeyStart.String(), tags["topic_key"]) } } assert.True(t, foundLatency, "Should have process.finish metric")