GH-3208 Assign partitions to KafkaBinderMetrics metadata consumer to avoid 'no longer assigned' WARN flood

[akenra]

Problem

Applications using the Kafka binder's offset-lag metrics (spring.cloud.stream.binder.kafka.offset gauge) get their logs flooded on kafka-clients 4.0.x-4.3.x:

Not updating last stable offset for partition <partition> as it is no longer assigned

(with read_uncommitted, the variant is ... high watermark ...). A reproducer was attached to the issue.

Root cause

The warning is emitted by kafka-clients itself in OffsetFetcherUtils.updateSubscriptionState() whenever a LIST_OFFSETS response arrives for a partition that is not in that consumer's current assignment (introduced by KAFKA-20131; downgraded to DEBUG only in kafka-clients 4.4.0 by KAFKA-20449).

KafkaBinderMetrics.findTotalTopicGroupLag uses a long-lived metadata Consumer per group that never subscribes or assigns any partitions. Every lag computation - scheduled every offsetLagMetricsInterval and additionally on every gauge scrape when defaultOffsetLagMetricsEnabled=true - calls endOffsets(), so every partition hits the "no longer assigned" path every time. For this usage it is structural, not a rebalance race.

Fix

Manually assign the discovered partitions to the metadata consumer before querying offsets:

metadataConsumer.assign(topicPartitions);

This keeps the client-side subscription state consistent with our usage so response processing takes the silent trace-level path. Computed lag values are unchanged (all existing metric-value assertions pass untouched).

Testing

• New unit test verifies assign() happens before endOffsets() on every computation.
• New KafkaBinderMetricsLoggingTests runs against an embedded broker with isolation.level=read_committed (mirroring the reporter's setup) and asserts zero such WARN events across repeated computations: it captures 8 warnings pre-fix, zero post-fix.
• Existing KafkaBinderMetricsTest (13 tests), KafkaMultiBinderCustomConfigurationTests (custom KafkaBinderMetrics subclass) and KafkaBinderUnitTests all green.

Resolves #3208

Follow-up ideas (out of scope here)

• Metadata consumers are keyed by group; two topics sharing a group touch one non-thread-safe Consumer from different scheduler threads (pre-existing).
• An AdminClient.listOffsets() / listConsumerGroupOffsets() based computation would remove per-group consumer threads entirely and decouple metrics from the app's isolation.level.