diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java index 5d6a1c2ccf..d4c4737978 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java @@ -217,6 +217,14 @@ private long findTotalTopicGroupLag(String topic, String group, Map endOffsets = metadataConsumer .endOffsets(topicPartitions); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsLoggingTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsLoggingTests.java new file mode 100644 index 0000000000..56c27cf246 --- /dev/null +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsLoggingTests.java @@ -0,0 +1,142 @@ +/* + * Copyright 2026-present the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder.kafka; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +import ch.qos.logback.classic.Level; +import ch.qos.logback.classic.Logger; +import ch.qos.logback.classic.spi.ILoggingEvent; +import ch.qos.logback.core.read.ListAppender; +import io.micrometer.core.instrument.MeterRegistry; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.ObjectProvider; +import org.springframework.boot.kafka.autoconfigure.KafkaProperties; + +import org.springframework.cloud.stream.binder.kafka.common.TopicInformation; +import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; +import org.springframework.kafka.test.EmbeddedKafkaBroker; +import org.springframework.kafka.test.condition.EmbeddedKafkaCondition; +import org.springframework.kafka.test.context.EmbeddedKafka; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** + * Verifies that offset-lag computations performed by {@link KafkaBinderMetrics} do + * not trigger kafka-clients' 'Not updating ... as it is no longer assigned' WARN + * logging (see GH-3208, KAFKA-20131, KAFKA-20449). + */ +@EmbeddedKafka(topics = KafkaBinderMetricsLoggingTests.TEST_TOPIC) +class KafkaBinderMetricsLoggingTests { + + static final String TEST_TOPIC = "metrics-logging-topic"; + + private static final String OFFSET_FETCHER_UTILS_LOGGER = + "org.apache.kafka.clients.consumer.internals.OffsetFetcherUtils"; + + private static EmbeddedKafkaBroker embeddedKafka; + + private final MeterRegistry meterRegistry = new SimpleMeterRegistry(); + + private Logger offsetFetcherUtilsLogger; + + private Level originalLevel; + + private ListAppender logAppender; + + @BeforeAll + static void startBroker() { + embeddedKafka = EmbeddedKafkaCondition.getBroker(); + } + + @AfterEach + void restoreLogging() { + if (this.offsetFetcherUtilsLogger != null) { + this.offsetFetcherUtilsLogger.detachAppender(this.logAppender); + this.offsetFetcherUtilsLogger.setLevel(this.originalLevel); + } + } + + @Test + void repeatedOffsetLagComputationsProduceNoNoLongerAssignedWarnings() { + captureKafkaClientWarnings(); + + KafkaBinderMetrics metrics = new KafkaBinderMetrics(binderWithTopicInUse(), + binderConfigurationProperties()); + metrics.bindTo(this.meterRegistry); + + // Two consecutive synchronous computations - each one calls endOffsets() + // on the metadata consumer and would emit a WARN per partition on + // affected kafka-clients versions (4.0.x - 4.3.x) without the fix. + this.meterRegistry.get(KafkaBinderMetrics.OFFSET_LAG_METRIC_NAME) + .tag("group", "metricsLoggingGroup").tag("topic", TEST_TOPIC).gauge().value(); + this.meterRegistry.get(KafkaBinderMetrics.OFFSET_LAG_METRIC_NAME) + .tag("group", "metricsLoggingGroup").tag("topic", TEST_TOPIC).gauge().value(); + + List warnings = this.logAppender.list.stream() + .filter(event -> event.getLevel() == Level.WARN) + .map(ILoggingEvent::getFormattedMessage) + .filter(message -> message.contains("no longer assigned")) + .collect(Collectors.toList()); + assertThat(warnings) + .as("kafka-clients must not log 'no longer assigned' warnings for " + + "binder offset-lag computations, but found: %s", warnings) + .isEmpty(); + } + + private KafkaMessageChannelBinder binderWithTopicInUse() { + Map topicsInUse = new HashMap<>(); + topicsInUse.put(TEST_TOPIC, + new TopicInformation("metricsLoggingGroup", null, true)); + KafkaMessageChannelBinder binder = mock(KafkaMessageChannelBinder.class); + when(binder.getTopicsInUse()).thenReturn(topicsInUse); + return binder; + } + + @SuppressWarnings("unchecked") + private KafkaBinderConfigurationProperties binderConfigurationProperties() { + KafkaProperties kafkaProperties = new KafkaProperties(); + KafkaBinderConfigurationProperties properties = + new KafkaBinderConfigurationProperties(kafkaProperties, + mock(ObjectProvider.class)); + properties.setBrokers(embeddedKafka.getBrokersAsString()); + // Mirrors the reporting application: read_committed selects the + // 'last stable offset' variant of the kafka-clients warning. + properties.getConfiguration().put("isolation.level", "read_committed"); + return properties; + } + + private void captureKafkaClientWarnings() { + this.offsetFetcherUtilsLogger = (Logger) LoggerFactory + .getLogger(OFFSET_FETCHER_UTILS_LOGGER); + this.originalLevel = this.offsetFetcherUtilsLogger.getLevel(); + this.offsetFetcherUtilsLogger.setLevel(Level.DEBUG); + this.logAppender = new ListAppender<>(); + this.logAppender.start(); + this.offsetFetcherUtilsLogger.addAppender(this.logAppender); + } + +} diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java index 80011fda77..8f706116c7 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java @@ -346,6 +346,24 @@ public void usesBeginningOffsetIfNoCommittedOffsetFound() { .isEqualTo(500.0); } + @Test + void metadataConsumerReceivesAssignBeforeEndOffsets() { + List partitions = partitions(new Node(0, null, 0)); + topicsInUse.put( + TEST_TOPIC, + new TopicInformation("group6-metrics", partitions, false) + ); + org.mockito.BDDMockito.given(consumer.partitionsFor(TEST_TOPIC)) + .willReturn(partitions); + metrics.bindTo(meterRegistry); + meterRegistry.get(KafkaBinderMetrics.OFFSET_LAG_METRIC_NAME) + .tag("group", "group6-metrics").tag("topic", TEST_TOPIC).gauge().value(); + + org.mockito.InOrder inOrder = Mockito.inOrder(consumer); + inOrder.verify(consumer).assign(ArgumentMatchers.anyCollection()); + inOrder.verify(consumer).endOffsets(ArgumentMatchers.anyCollection()); + } + @Test public void shouldShutdownSchedulerOnClose() { metrics.bindTo(meterRegistry);