Deduplicate metrics created by loadAndRun

[danieljbingham]

Is your feature request related to a problem? Please describe.
We've been noticing logs similar to the below, logging constantly for different kafka metrics, reporting multiple values for the same attributes:

[otel.javaagent 2024-11-20 10:01:38:390 +0000] [prometheus-http-1-1] WARN io.opentelemetry.sdk.metrics.internal.state.AsynchronousMetricStorage - Instrument kafka.consumer.connection_close_rate has recorded multiple values for the same attributes: {client-id="recently-watched-ingest"}

This duplication is caused by loadAndRun creating 2 KafkaConsumers with identical client-ids. We validated this by updating the clientId for one of the consumers, and the warning logs disappeared.

Describe the solution you'd like
We should be able to specify different consumerSettings or clientIds per KafkaConsumer to avoid these clashes.

Describe alternatives you've considered

• We could disable our own logs for this, but this would just hide the problem
• Only one consumer could be used, but two are required for re-assigning offsets after the initial load

Additional context
loadAndRun method

[bcarter97]

I think because the preloadConsumer is still alive in scope of the postLoad function it is never closed.

fs2-kafka-topic-loader/src/main/scala/uk/sky/fs2/kafka/topicloader/TopicLoader.scala

Lines 85 to 91 in 47572c0

	for {
	given Logger[F] <- Stream.eval(LoggerFactory[F].create)
	preloadConsumer <- KafkaConsumer.stream(consumerSettings)
	logOffsets <- Stream.eval(logOffsetsForTopics(topics, LoadAll, preloadConsumer)).flatMap(Stream.fromOption(_))
	_ <- Stream.eval(info"log offsets: ${logOffsets.show}")
	record <- load(logOffsets, preloadConsumer).onFinalizeCase(onLoad) ++ postLoad(logOffsets)
	} yield record

Because it's needed to get the logOffsets we have two options as I can see it:

1. Create a 3rd consumer to get the LogOffsets and destroy this one properly
2. Destroy the preloadConsumer before calling postLoad