From 1822363aae6ccbf52c0d9df3a34edbac25461526 Mon Sep 17 00:00:00 2001 From: kruskall <99559985+kruskall@users.noreply.github.com> Date: Thu, 16 Jan 2025 14:27:06 +0100 Subject: [PATCH 1/2] fix: better protect against nil topicAttributeFunc (#618) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## ❓ Why is this being changed systemtests are failing because of a nil topic attribute func. The topic manager does not handle nil func. See https://github.com/elastic/apm-queue/actions/runs/12741942829/job/35509307227 ## 🧑‍💻 What is being changed set the default func in the common config instead of the metrics hook ## ✅ How to validate the change run systemtests --- kafka/common.go | 7 +++++++ kafka/common_test.go | 1 + kafka/consumer_test.go | 36 ++++++++++++++++-------------------- kafka/metrics.go | 6 ------ 4 files changed, 24 insertions(+), 26 deletions(-) diff --git a/kafka/common.go b/kafka/common.go index d2284b3f..ae2ff716 100644 --- a/kafka/common.go +++ b/kafka/common.go @@ -34,6 +34,7 @@ import ( "github.com/twmb/franz-go/pkg/sasl/plain" "github.com/twmb/franz-go/plugin/kzap" "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/metric" "go.opentelemetry.io/otel/trace" "go.uber.org/zap" @@ -247,6 +248,12 @@ func (cfg *CommonConfig) finalize() error { if cfg.TopicLogFieldFunc != nil { cfg.TopicLogFieldFunc = topicFieldFunc(cfg.TopicLogFieldFunc) } + if cfg.TopicAttributeFunc == nil { + cfg.TopicAttributeFunc = func(topic string) attribute.KeyValue { + return attribute.KeyValue{} + } + } + return errors.Join(errs...) } diff --git a/kafka/common_test.go b/kafka/common_test.go index 31c68d0e..7cddc7f4 100644 --- a/kafka/common_test.go +++ b/kafka/common_test.go @@ -47,6 +47,7 @@ func TestCommonConfig(t *testing.T) { t.Helper() err := in.finalize() require.NoError(t, err) + in.TopicAttributeFunc = nil in.TopicLogFieldFunc = nil in.hooks = nil assert.Equal(t, expected, in) diff --git a/kafka/consumer_test.go b/kafka/consumer_test.go index eeb08516..1dc613cf 100644 --- a/kafka/consumer_test.go +++ b/kafka/consumer_test.go @@ -780,10 +780,7 @@ func TestConsumerConfigFinalizer(t *testing.T) { } err := cfg.finalize() require.NoError(t, err) - assert.NotNil(t, cfg.Processor) - cfg.Processor = nil - assert.NotNil(t, cfg.Logger) - cfg.Logger = nil + assertNotNilOptions(t, &cfg) assert.Equal(t, ConsumerConfig{ CommonConfig: CommonConfig{Brokers: []string{"localhost:9092"}}, @@ -805,10 +802,7 @@ func TestConsumerConfigFinalizer(t *testing.T) { } err := cfg.finalize() require.NoError(t, err) - assert.NotNil(t, cfg.Processor) - cfg.Processor = nil - assert.NotNil(t, cfg.Logger) - cfg.Logger = nil + assertNotNilOptions(t, &cfg) assert.Equal(t, ConsumerConfig{ CommonConfig: CommonConfig{Brokers: []string{"localhost:9092"}}, @@ -830,10 +824,7 @@ func TestConsumerConfigFinalizer(t *testing.T) { } err := cfg.finalize() require.NoError(t, err) - assert.NotNil(t, cfg.Processor) - cfg.Processor = nil - assert.NotNil(t, cfg.Logger) - cfg.Logger = nil + assertNotNilOptions(t, &cfg) assert.Equal(t, ConsumerConfig{ CommonConfig: CommonConfig{Brokers: []string{"localhost:9092"}}, @@ -855,10 +846,7 @@ func TestConsumerConfigFinalizer(t *testing.T) { } err := cfg.finalize() require.NoError(t, err) - assert.NotNil(t, cfg.Processor) - cfg.Processor = nil - assert.NotNil(t, cfg.Logger) - cfg.Logger = nil + assertNotNilOptions(t, &cfg) assert.Equal(t, ConsumerConfig{ CommonConfig: CommonConfig{Brokers: []string{"localhost:9092"}}, @@ -880,10 +868,7 @@ func TestConsumerConfigFinalizer(t *testing.T) { } err := cfg.finalize() require.NoError(t, err) - assert.NotNil(t, cfg.Processor) - cfg.Processor = nil - assert.NotNil(t, cfg.Logger) - cfg.Logger = nil + assertNotNilOptions(t, &cfg) assert.Equal(t, ConsumerConfig{ CommonConfig: CommonConfig{Brokers: []string{"localhost:9092"}}, @@ -909,6 +894,17 @@ func TestConsumerConfigFinalizer(t *testing.T) { }) } +func assertNotNilOptions(t testing.TB, cfg *ConsumerConfig) { + t.Helper() + + assert.NotNil(t, cfg.Processor) + cfg.Processor = nil + assert.NotNil(t, cfg.Logger) + cfg.Logger = nil + assert.NotNil(t, cfg.TopicAttributeFunc) + cfg.TopicAttributeFunc = nil +} + func newConsumer(t testing.TB, cfg ConsumerConfig) *Consumer { if cfg.MaxPollWait <= 0 { // Lower MaxPollWait, ShutdownGracePeriod to speed up execution. diff --git a/kafka/metrics.go b/kafka/metrics.go index 2e883456..09f4fb89 100644 --- a/kafka/metrics.go +++ b/kafka/metrics.go @@ -299,12 +299,6 @@ func newKgoHooks(mp metric.MeterProvider, namespace, topicPrefix string, return nil, formatMetricError(throttlingDurationKey, err) } - if topicAttributeFunc == nil { - topicAttributeFunc = func(topic string) attribute.KeyValue { - return attribute.KeyValue{} - } - } - return &metricHooks{ namespace: namespace, topicPrefix: topicPrefix, From e452a84f10653132a6d857d914fc80314cf4b763 Mon Sep 17 00:00:00 2001 From: "dependabot[bot]" <49699333+dependabot[bot]@users.noreply.github.com> Date: Thu, 16 Jan 2025 13:29:49 +0000 Subject: [PATCH 2/2] build(deps): bump golang.org/x/tools from 0.28.0 to 0.29.0 in /tools (#611) Bumps [golang.org/x/tools](https://github.com/golang/tools) from 0.28.0 to 0.29.0.
Commits

[![Dependabot compatibility score](https://dependabot-badges.githubapp.com/badges/compatibility_score?dependency-name=golang.org/x/tools&package-manager=go_modules&previous-version=0.28.0&new-version=0.29.0)](https://docs.github.com/en/github/managing-security-vulnerabilities/about-dependabot-security-updates#about-compatibility-scores) Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting `@dependabot rebase`. [//]: # (dependabot-automerge-start) [//]: # (dependabot-automerge-end) ---
Dependabot commands and options
You can trigger Dependabot actions by commenting on this PR: - `@dependabot rebase` will rebase this PR - `@dependabot recreate` will recreate this PR, overwriting any edits that have been made to it - `@dependabot merge` will merge this PR after your CI passes on it - `@dependabot squash and merge` will squash and merge this PR after your CI passes on it - `@dependabot cancel merge` will cancel a previously requested merge and block automerging - `@dependabot reopen` will reopen this PR if it is closed - `@dependabot close` will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually - `@dependabot show ignore conditions` will show all of the ignore conditions of the specified dependency - `@dependabot ignore this major version` will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself) - `@dependabot ignore this minor version` will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself) - `@dependabot ignore this dependency` will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)
Signed-off-by: dependabot[bot] Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: kruskall <99559985+kruskall@users.noreply.github.com> --- tools/go.mod | 2 +- tools/go.sum | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/tools/go.mod b/tools/go.mod index 4e459de8..1dda5bdb 100644 --- a/tools/go.mod +++ b/tools/go.mod @@ -5,7 +5,7 @@ go 1.22.1 require ( github.com/elastic/go-licenser v0.4.2 go.elastic.co/go-licence-detector v0.7.0 - golang.org/x/tools v0.28.0 + golang.org/x/tools v0.29.0 honnef.co/go/tools v0.5.1 ) diff --git a/tools/go.sum b/tools/go.sum index dd144551..a229b1fa 100644 --- a/tools/go.sum +++ b/tools/go.sum @@ -33,8 +33,8 @@ golang.org/x/mod v0.22.0 h1:D4nJWe9zXqHOmWqj4VMOJhvzj7bEZg4wEYa759z1pH4= golang.org/x/mod v0.22.0/go.mod h1:6SkKJ3Xj0I0BrPOZoBy3bdMptDDU9oJrpohJ3eWZ1fY= golang.org/x/sync v0.10.0 h1:3NQrjDixjgGwUOCaF8w2+VYHv0Ve/vGYSbdkTa98gmQ= golang.org/x/sync v0.10.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= -golang.org/x/tools v0.28.0 h1:WuB6qZ4RPCQo5aP3WdKZS7i595EdWqWR8vqJTlwTVK8= -golang.org/x/tools v0.28.0/go.mod h1:dcIOrVd3mfQKTgrDVQHqCPMWy6lnhfhtX3hLXYVLfRw= +golang.org/x/tools v0.29.0 h1:Xx0h3TtM9rzQpQuR4dKLrdglAmCEN5Oi+P74JdhdzXE= +golang.org/x/tools v0.29.0/go.mod h1:KMQVMRsVxU6nHCFXrBPhDB8XncLNLM0lIy/F14RP588= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=