From ada45448367271a02b16b84c0bf35c7dab9648ba Mon Sep 17 00:00:00 2001 From: jerry <1394367234@qq.com> Date: Mon, 21 Sep 2026 00:04:20 +0800 Subject: [PATCH] fix logging-kafka : avoid sending a record during initialization. --- .../kafka/client/KafkaLogCollectClient.java | 27 ++++++---------- .../kafka/KafkaLogCollectClientTest.java | 31 +++++++++++++------ 2 files changed, 30 insertions(+), 28 deletions(-) diff --git a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-kafka/src/main/java/org/apache/shenyu/plugin/logging/kafka/client/KafkaLogCollectClient.java b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-kafka/src/main/java/org/apache/shenyu/plugin/logging/kafka/client/KafkaLogCollectClient.java index 3cbb1ecbab7f..e919111dab14 100644 --- a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-kafka/src/main/java/org/apache/shenyu/plugin/logging/kafka/client/KafkaLogCollectClient.java +++ b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-kafka/src/main/java/org/apache/shenyu/plugin/logging/kafka/client/KafkaLogCollectClient.java @@ -26,9 +26,6 @@ import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.KafkaException; import org.apache.kafka.common.config.SaslConfigs; -import org.apache.kafka.common.errors.AuthorizationException; -import org.apache.kafka.common.errors.OutOfOrderSequenceException; -import org.apache.kafka.common.errors.ProducerFencedException; import org.apache.kafka.common.serialization.StringSerializer; import org.apache.shenyu.common.utils.GsonUtils; import org.apache.shenyu.common.utils.JsonUtils; @@ -96,25 +93,19 @@ public boolean initClient0(@NonNull final KafkaLogCollectConfig.KafkaLogConfig c .format("org.apache.kafka.common.security.scram.ScramLoginModule required username=\"{0}\" password=\"{1}\";", config.getUserName(), config.getPassWord())); } - producer = new KafkaProducer<>(props); - ProducerRecord record = new ProducerRecord<>(this.topic, StringSerializer.class.getName(), StringSerializer.class.getName()); try { - producer.send(record); - LOG.info("init kafkaLogCollectClient success"); - } catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) { - // We can't recover from these exceptions, so our only option is to close the producer and exit. - LOG.error("Init kafkaLogCollectClient error, We can't recover from these exceptions, so our only option is to close the producer and exit", e); - producer.close(); - return false; + producer = new KafkaProducer<>(props); + producer.partitionsFor(this.topic); + LOG.info("kafka topic metadata fetched successfully"); + return true; } catch (KafkaException e) { - // For all other exceptions, just abort the transaction and try again. - LOG.error( - "init kafkaLogCollectClient error，Exceptions other than ProducerFencedException or OutOfOrderSequenceException or AuthorizationException" - + ", just abort the transaction and try again", e); - producer.close(); + LOG.error("Failed to initialize kafka producer", e); + if (Objects.nonNull(producer)) { + producer.close(); + producer = null; + } return false; } - return true; } /** diff --git a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-kafka/src/test/java/org/apache/shenyu/plugin/logging/kafka/kafka/KafkaLogCollectClientTest.java b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-kafka/src/test/java/org/apache/shenyu/plugin/logging/kafka/kafka/KafkaLogCollectClientTest.java index e0b5cd6b8303..a307fc3e4b86 100644 --- a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-kafka/src/test/java/org/apache/shenyu/plugin/logging/kafka/kafka/KafkaLogCollectClientTest.java +++ b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-kafka/src/test/java/org/apache/shenyu/plugin/logging/kafka/kafka/KafkaLogCollectClientTest.java @@ -18,6 +18,7 @@ package org.apache.shenyu.plugin.logging.kafka.kafka; import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.common.errors.TimeoutException; import org.apache.shenyu.common.dto.PluginData; import org.apache.shenyu.common.utils.GsonUtils; import org.apache.shenyu.plugin.logging.common.entity.ShenyuRequestLog; @@ -25,13 +26,14 @@ import org.apache.shenyu.plugin.logging.kafka.config.KafkaLogCollectConfig; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; import org.mockito.MockedConstruction; -import java.lang.reflect.Field; - +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.mockConstruction; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; /** * The Test Case For KafkaLogCollectClient. @@ -58,14 +60,23 @@ public void setUp() { } @Test - @Disabled - public void testInitClient() throws NoSuchFieldException, IllegalAccessException { + public void testInitClientDoesNotSendRecord() { try (MockedConstruction construction = mockConstruction(KafkaProducer.class)) { - kafkaLogCollectClient.initClient(globalLogConfig); - Field field = kafkaLogCollectClient.getClass().getDeclaredField("topic"); - field.setAccessible(true); - Assertions.assertEquals(field.get(kafkaLogCollectClient), "shenyu-access-logging"); - kafkaLogCollectClient.close(); + Assertions.assertTrue(kafkaLogCollectClient.initClient0(globalLogConfig)); + Assertions.assertEquals(1, construction.constructed().size()); + verify(construction.constructed().get(0)).partitionsFor("shenyu-access-logging"); + verify(construction.constructed().get(0), never()).send(any()); + } + kafkaLogCollectClient.close0(); + } + + @Test + public void testInitClientFailsWhenTopicMetadataIsUnavailable() { + try (MockedConstruction construction = mockConstruction(KafkaProducer.class, + (mock, context) -> when(mock.partitionsFor("shenyu-access-logging")).thenThrow(new TimeoutException("metadata unavailable")))) { + Assertions.assertFalse(kafkaLogCollectClient.initClient0(globalLogConfig)); + verify(construction.constructed().get(0)).close(); + verify(construction.constructed().get(0), never()).send(any()); } } }