This is an automated email from the ASF dual-hosted git repository.
Aias00 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shenyu.git
The following commit(s) were added to refs/heads/master by this push:
new 08e92ab1b2 fix: avoid consuming logs after client init failure (#6959)
08e92ab1b2 is described below
commit 08e92ab1b26942950d7bfab6802af1e25f20460c
Author: hengyuss <[email protected]>
AuthorDate: Sun Aug 23 15:01:01 2026 +0800
fix: avoid consuming logs after client init failure (#6959)
* fix: avoid consuming logs after client init failure
* fix: fix tests
* trigger ci
---------
Co-authored-by: aias00 <[email protected]>
---
.../sls/client/AliyunSlsLogCollectClient.java | 12 ++-
.../client/ClickHouseLogCollectClient.java | 6 +-
.../client/ClickHouseLogCollectClientTest.java | 8 +-
.../common/client/AbstractLogConsumeClient.java | 13 ++-
.../client/AbstractLogConsumeClientTest.java | 112 +++++++++++++++++++++
.../client/ElasticSearchLogCollectClient.java | 4 +-
.../lts/client/HuaweiLtsLogCollectClient.java | 6 +-
.../kafka/client/KafkaLogCollectClient.java | 15 ++-
.../pulsar/client/PulsarLogCollectClient.java | 22 ++--
.../pulsar/client/PulsarLogCollectClientTest.java | 9 +-
.../rabbitmq/client/RabbitmqLogCollectClient.java | 28 +++++-
.../rocketmq/client/RocketMQLogCollectClient.java | 12 ++-
.../cls/client/TencentClsLogCollectClient.java | 12 ++-
13 files changed, 217 insertions(+), 42 deletions(-)
diff --git
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-aliyun-sls/src/main/java/org/apache/shenyu/plugin/aliyun/sls/client/AliyunSlsLogCollectClient.java
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-aliyun-sls/src/main/java/org/apache/shenyu/plugin/aliyun/sls/client/AliyunSlsLogCollectClient.java
index 2f03ae7444..412b509e66 100644
---
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-aliyun-sls/src/main/java/org/apache/shenyu/plugin/aliyun/sls/client/AliyunSlsLogCollectClient.java
+++
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-aliyun-sls/src/main/java/org/apache/shenyu/plugin/aliyun/sls/client/AliyunSlsLogCollectClient.java
@@ -72,15 +72,16 @@ public class AliyunSlsLogCollectClient extends
AbstractLogConsumeClient<AliyunLo
* init aliyun sls client.
*
* @param config config
+ * @return true if the client was initialized successfully
*/
@Override
- public void initClient0(@NonNull final
AliyunLogCollectConfig.AliyunSlsLogConfig config) {
+ public boolean initClient0(@NonNull final
AliyunLogCollectConfig.AliyunSlsLogConfig config) {
String accessId = config.getAccessId();
String accessKey = config.getAccessKey();
String host = config.getHost();
if (StringUtils.isBlank(accessId) || StringUtils.isBlank(accessKey) ||
StringUtils.isBlank(host)) {
LOG.error("init aliyun sls client error, please check accessId,
accessKey or host");
- return;
+ return false;
}
client = new Client(host, accessId, accessKey);
// create LogStore, if you don't create logStore, shenyu will do it.
@@ -98,7 +99,14 @@ public class AliyunSlsLogCollectClient extends
AbstractLogConsumeClient<AliyunLo
client.CreateLogStore(projectName, store);
} catch (LogException e) {
LOG.warn("error code:{}, error message:{}", e.GetErrorCode(),
e.GetErrorMessage());
+ try {
+ close0();
+ } catch (Exception closeException) {
+ LOG.error("failed to clean up Aliyun SLS resources after
initialization failure", closeException);
+ }
+ return false;
}
+ return true;
}
/**
diff --git
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-clickhouse/src/main/java/org/apache/shenyu/plugin/logging/clickhouse/client/ClickHouseLogCollectClient.java
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-clickhouse/src/main/java/org/apache/shenyu/plugin/logging/clickhouse/client/ClickHouseLogCollectClient.java
index 7b6d0e3e51..823ae373c1 100644
---
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-clickhouse/src/main/java/org/apache/shenyu/plugin/logging/clickhouse/client/ClickHouseLogCollectClient.java
+++
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-clickhouse/src/main/java/org/apache/shenyu/plugin/logging/clickhouse/client/ClickHouseLogCollectClient.java
@@ -121,9 +121,10 @@ public class ClickHouseLogCollectClient extends
AbstractLogConsumeClient<ClickHo
* init client .
*
* @param config properties.
+ * @return true if the client was initialized successfully
*/
@Override
- public void initClient0(@NonNull final
ClickHouseLogCollectConfig.ClickHouseLogConfig config) {
+ public boolean initClient0(@NonNull final
ClickHouseLogCollectConfig.ClickHouseLogConfig config) {
final String username = config.getUsername();
final String password = config.getPassword();
final String ttl = StringUtils.defaultIfBlank(config.getTtl(), "30");
@@ -141,6 +142,9 @@ public class ClickHouseLogCollectClient extends
AbstractLogConsumeClient<ClickHo
request.query(String.format(ClickHouseLoggingConstant.CREATE_DISTRIBUTED_TABLE_SQL,
database, database, config.getClusterName(), database)).executeAndWait();
} catch (Exception e) {
LOG.error("inti ClickHouseLogClient error", e);
+ close0();
+ return false;
}
+ return true;
}
}
diff --git
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-clickhouse/src/test/java/org/apache/shenyu/plugin/logging/clickhouse/client/ClickHouseLogCollectClientTest.java
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-clickhouse/src/test/java/org/apache/shenyu/plugin/logging/clickhouse/client/ClickHouseLogCollectClientTest.java
index 8d6079d126..2a9ce4c196 100644
---
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-clickhouse/src/test/java/org/apache/shenyu/plugin/logging/clickhouse/client/ClickHouseLogCollectClientTest.java
+++
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-clickhouse/src/test/java/org/apache/shenyu/plugin/logging/clickhouse/client/ClickHouseLogCollectClientTest.java
@@ -75,15 +75,9 @@ public class ClickHouseLogCollectClientTest {
@Test
public void testConsume() {
- String msg = "";
ClickHouseLogCollectConfig.INSTANCE.setClickHouseLogConfig(clickHouseLogConfig);
clickHouseLogCollectClient.initClient(clickHouseLogConfig);
- try {
- clickHouseLogCollectClient.consume(logs);
- } catch (Exception e) {
- msg = "false";
- }
- Assertions.assertEquals(msg, "false");
+ Assertions.assertDoesNotThrow(() ->
clickHouseLogCollectClient.consume(logs));
}
@Test
diff --git
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/main/java/org/apache/shenyu/plugin/logging/common/client/AbstractLogConsumeClient.java
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/main/java/org/apache/shenyu/plugin/logging/common/client/AbstractLogConsumeClient.java
index aba6fb6edc..b5dafa3fae 100644
---
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/main/java/org/apache/shenyu/plugin/logging/common/client/AbstractLogConsumeClient.java
+++
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/main/java/org/apache/shenyu/plugin/logging/common/client/AbstractLogConsumeClient.java
@@ -45,8 +45,9 @@ public abstract class AbstractLogConsumeClient<T extends
GenericGlobalConfig, L
* initClient0.
*
* @param config config
+ * @return true if the client was initialized successfully
*/
- public abstract void initClient0(@NonNull T config);
+ public abstract boolean initClient0(@NonNull T config);
/**
* consume0.
@@ -72,10 +73,12 @@ public abstract class AbstractLogConsumeClient<T extends
GenericGlobalConfig, L
LOG.error("{} config is null, client not init.",
this.getClass().getSimpleName());
return;
}
- this.initClient0(config);
- isStarted.set(true);
- closeThread.set(new Thread(this::close));
- Runtime.getRuntime().addShutdownHook(closeThread.get());
+ boolean initialized = this.initClient0(config);
+ isStarted.set(initialized);
+ if (initialized) {
+ closeThread.set(new Thread(this::close));
+ Runtime.getRuntime().addShutdownHook(closeThread.get());
+ }
}
@Override
diff --git
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/test/java/org/apache/shenyu/plugin/logging/common/client/AbstractLogConsumeClientTest.java
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/test/java/org/apache/shenyu/plugin/logging/common/client/AbstractLogConsumeClientTest.java
new file mode 100644
index 0000000000..67089e3844
--- /dev/null
+++
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/test/java/org/apache/shenyu/plugin/logging/common/client/AbstractLogConsumeClientTest.java
@@ -0,0 +1,112 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.shenyu.plugin.logging.common.client;
+
+import org.apache.shenyu.plugin.logging.common.config.GenericGlobalConfig;
+import org.apache.shenyu.plugin.logging.common.entity.ShenyuRequestLog;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Test cases for {@link AbstractLogConsumeClient}.
+ */
+public final class AbstractLogConsumeClientTest {
+
+ @Test
+ public void testConsumeIsSkippedWhenInitializationFails() throws Exception
{
+ TestLogConsumeClient client = new TestLogConsumeClient(false);
+
+ client.initClient(new GenericGlobalConfig());
+ client.consume(Collections.singletonList(new ShenyuRequestLog()));
+
+ assertEquals(0, client.getConsumeCount());
+ assertEquals(0, client.getCloseCount());
+ }
+
+ @Test
+ public void testConsumeRunsWhenInitializationSucceeds() throws Exception {
+ TestLogConsumeClient client = new TestLogConsumeClient(true);
+
+ try {
+ client.initClient(new GenericGlobalConfig());
+ client.consume(Collections.singletonList(new ShenyuRequestLog()));
+
+ assertEquals(1, client.getConsumeCount());
+ } finally {
+ client.close();
+ }
+ assertEquals(1, client.getCloseCount());
+ }
+
+ @Test
+ public void testFailedReinitializationLeavesClientStopped() throws
Exception {
+ TestLogConsumeClient client = new TestLogConsumeClient(true);
+ client.initClient(new GenericGlobalConfig());
+ client.setInitializationResult(false);
+
+ client.initClient(new GenericGlobalConfig());
+ client.consume(Collections.singletonList(new ShenyuRequestLog()));
+
+ assertEquals(0, client.getConsumeCount());
+ assertEquals(1, client.getCloseCount());
+ }
+
+ private static final class TestLogConsumeClient extends
AbstractLogConsumeClient<GenericGlobalConfig, ShenyuRequestLog> {
+
+ private boolean initializationResult;
+
+ private int consumeCount;
+
+ private int closeCount;
+
+ private TestLogConsumeClient(final boolean initializationResult) {
+ this.initializationResult = initializationResult;
+ }
+
+ @Override
+ public boolean initClient0(final GenericGlobalConfig config) {
+ return initializationResult;
+ }
+
+ @Override
+ public void consume0(final List<ShenyuRequestLog> logs) {
+ consumeCount++;
+ }
+
+ @Override
+ public void close0() {
+ closeCount++;
+ }
+
+ private void setInitializationResult(final boolean
initializationResult) {
+ this.initializationResult = initializationResult;
+ }
+
+ private int getConsumeCount() {
+ return consumeCount;
+ }
+
+ private int getCloseCount() {
+ return closeCount;
+ }
+ }
+}
diff --git
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-elasticsearch/src/main/java/org/apache/shenyu/plugin/logging/elasticsearch/client/ElasticSearchLogCollectClient.java
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-elasticsearch/src/main/java/org/apache/shenyu/plugin/logging/elasticsearch/client/ElasticSearchLogCollectClient.java
index 97a9ac728a..ffd41861f9 100644
---
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-elasticsearch/src/main/java/org/apache/shenyu/plugin/logging/elasticsearch/client/ElasticSearchLogCollectClient.java
+++
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-elasticsearch/src/main/java/org/apache/shenyu/plugin/logging/elasticsearch/client/ElasticSearchLogCollectClient.java
@@ -68,9 +68,10 @@ public class ElasticSearchLogCollectClient extends
AbstractLogConsumeClient<Elas
* init elasticsearch client.
*
* @param config elasticsearch client config
+ * @return true if the client was initialized successfully
*/
@Override
- public void initClient0(@NonNull final
ElasticSearchLogCollectConfig.ElasticSearchLogConfig config) {
+ public boolean initClient0(@NonNull final
ElasticSearchLogCollectConfig.ElasticSearchLogConfig config) {
RestClientBuilder builder = RestClient
.builder(new HttpHost(config.getHost(),
Integer.parseInt(config.getPort())));
@@ -94,6 +95,7 @@ public class ElasticSearchLogCollectClient extends
AbstractLogConsumeClient<Elas
LogUtils.info(LOG, "init ElasticSearchLogCollectClient success");
createOrUpdateIndexAlias(indexName);
+ return true;
}
/**
diff --git
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-huawei-lts/src/main/java/org/apache/shenyu/plugin/huawei/lts/client/HuaweiLtsLogCollectClient.java
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-huawei-lts/src/main/java/org/apache/shenyu/plugin/huawei/lts/client/HuaweiLtsLogCollectClient.java
index ac6eb63d9f..05e1656777 100644
---
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-huawei-lts/src/main/java/org/apache/shenyu/plugin/huawei/lts/client/HuaweiLtsLogCollectClient.java
+++
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-huawei-lts/src/main/java/org/apache/shenyu/plugin/huawei/lts/client/HuaweiLtsLogCollectClient.java
@@ -68,9 +68,10 @@ public class HuaweiLtsLogCollectClient extends
AbstractLogConsumeClient<HuaweiLo
* init Huawei lts client.
*
* @param huaweiLtsLogConfig shenyu log config
+ * @return true if the client was initialized successfully
*/
@Override
- public void initClient0(@NonNull final
HuaweiLogCollectConfig.HuaweiLtsLogConfig huaweiLtsLogConfig) {
+ public boolean initClient0(@NonNull final
HuaweiLogCollectConfig.HuaweiLtsLogConfig huaweiLtsLogConfig) {
final String accessKeyId = huaweiLtsLogConfig.getAccessKeyId();
final String accessKeySecret = huaweiLtsLogConfig.getAccessKeySecret();
final String regionName = huaweiLtsLogConfig.getRegionName();
@@ -80,7 +81,7 @@ public class HuaweiLtsLogCollectClient extends
AbstractLogConsumeClient<HuaweiLo
if (StringUtils.isBlank(accessKeyId) ||
StringUtils.isBlank(accessKeySecret) || StringUtils.isBlank(projectId)
|| StringUtils.isBlank(regionName) ||
StringUtils.isBlank(logGroupId) || StringUtils.isBlank(logStreamId)) {
LOG.error("init Huawei lts client error, please check projectId,
accessKeyId, accessKeySecret, regionName, logGroupId or logStreamId");
- return;
+ return false;
}
JavaSDKAppender appender = JavaSDKAppender.custom()
.setProjectId(projectId)
@@ -101,6 +102,7 @@ public class HuaweiLtsLogCollectClient extends
AbstractLogConsumeClient<HuaweiLo
this.producer = appender.getProducer();
threadExecutor =
createThreadPoolExecutor(huaweiLtsLogConfig.getIoThreadCount());
+ return true;
}
/**
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 42c3bf33a4..3cbb1ecbab 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
@@ -63,12 +63,13 @@ public class KafkaLogCollectClient extends
AbstractLogConsumeClient<KafkaLogColl
* init producer.
*
* @param config kafka props
+ * @return true if the client was initialized successfully
*/
@Override
- public void initClient0(@NonNull final
KafkaLogCollectConfig.KafkaLogConfig config) {
+ public boolean initClient0(@NonNull final
KafkaLogCollectConfig.KafkaLogConfig config) {
if (StringUtils.isBlank(config.getBootstrapServer()) ||
StringUtils.isBlank(config.getTopic())) {
LOG.error("kafka props is empty. failed init kafka producer");
- return;
+ return false;
}
LOG.info("initClient0:{}", GsonUtils.getInstance().toJson(config));
@@ -78,7 +79,7 @@ public class KafkaLogCollectClient extends
AbstractLogConsumeClient<KafkaLogColl
if (StringUtils.isBlank(topic) ||
StringUtils.isBlank(nameserverAddress)) {
LOG.error("init kafkaLogCollectClient error, please check topic or
nameserverAddress");
- return;
+ return false;
}
this.topic = topic;
@@ -104,12 +105,16 @@ public class KafkaLogCollectClient extends
AbstractLogConsumeClient<KafkaLogColl
// 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;
} 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();
+ return false;
}
+ return true;
}
/**
@@ -119,6 +124,10 @@ public class KafkaLogCollectClient extends
AbstractLogConsumeClient<KafkaLogColl
*/
@Override
public void consume0(@NonNull final List<ShenyuRequestLog> logs) {
+ if (Objects.isNull(producer)) {
+ LOG.warn("Kafka producer is not initialized.");
+ return;
+ }
logs.forEach(log -> {
String logTopic =
Optional.ofNullable(LoggingKafkaPluginDataHandler.getSelectApiConfigMap().get(log.getSelectorId()))
.map(apiConfig ->
StringUtils.defaultIfBlank(apiConfig.getTopic(), topic)
diff --git
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-pulsar/src/main/java/org/apache/shenyu/plugin/logging/pulsar/client/PulsarLogCollectClient.java
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-pulsar/src/main/java/org/apache/shenyu/plugin/logging/pulsar/client/PulsarLogCollectClient.java
index c58a4d13df..2c4235bb02 100644
---
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-pulsar/src/main/java/org/apache/shenyu/plugin/logging/pulsar/client/PulsarLogCollectClient.java
+++
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-pulsar/src/main/java/org/apache/shenyu/plugin/logging/pulsar/client/PulsarLogCollectClient.java
@@ -50,14 +50,15 @@ public class PulsarLogCollectClient extends
AbstractLogConsumeClient<PulsarLogCo
* init producer.
*
* @param config pulsar props
+ * @return true if the client was initialized successfully
*/
@Override
- public void initClient0(@NonNull final
PulsarLogCollectConfig.PulsarLogConfig config) {
+ public boolean initClient0(@NonNull final
PulsarLogCollectConfig.PulsarLogConfig config) {
String topic = config.getTopic();
String serviceUrl = config.getServiceUrl();
if (StringUtils.isBlank(topic) || StringUtils.isBlank(serviceUrl)) {
LOG.error("init PulsarLogCollectClient error, please check topic
or serviceUrl.");
- return;
+ return false;
}
try {
client = PulsarClient.builder().serviceUrl(serviceUrl).build();
@@ -66,11 +67,18 @@ public class PulsarLogCollectClient extends
AbstractLogConsumeClient<PulsarLogCo
} catch (PulsarClientException e) {
LOG.error("init PulsarLogCollectClient error, ", e);
+ close0();
+ return false;
}
+ return true;
}
@Override
public void consume0(@NonNull final List<ShenyuRequestLog> logs) {
+ if (Objects.isNull(producer)) {
+ LOG.warn("Pulsar producer is not initialized.");
+ return;
+ }
logs.forEach(log -> producer.sendAsync(toBytes(log)));
}
@@ -93,13 +101,15 @@ public class PulsarLogCollectClient extends
AbstractLogConsumeClient<PulsarLogCo
@Override
public void close0() {
- if (Objects.nonNull(producer)) {
- try {
+ try {
+ if (Objects.nonNull(producer)) {
producer.close();
+ }
+ if (Objects.nonNull(client)) {
client.close();
- } catch (PulsarClientException e) {
- LOG.error("fail to close PulsarLogCollectClient, e", e);
}
+ } catch (PulsarClientException e) {
+ LOG.error("fail to close PulsarLogCollectClient, e", e);
}
}
}
diff --git
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-pulsar/src/test/java/org/apache/shenyu/plugin/logging/pulsar/client/PulsarLogCollectClientTest.java
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-pulsar/src/test/java/org/apache/shenyu/plugin/logging/pulsar/client/PulsarLogCollectClientTest.java
index 2f4837a0a9..1962cbec96 100644
---
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-pulsar/src/test/java/org/apache/shenyu/plugin/logging/pulsar/client/PulsarLogCollectClientTest.java
+++
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-pulsar/src/test/java/org/apache/shenyu/plugin/logging/pulsar/client/PulsarLogCollectClientTest.java
@@ -53,16 +53,9 @@ public class PulsarLogCollectClientTest {
@Test
public void testConsume() {
- String msg = "";
PulsarLogCollectConfig.INSTANCE.setPulsarLogConfig(pulsarLogConfig);
pulsarLogCollectClient.initClient(pulsarLogConfig);
- try {
- pulsarLogCollectClient.consume(logs);
- } catch (Exception e) {
- msg = "false";
- }
- Assertions.assertEquals(msg, "false");
+ Assertions.assertDoesNotThrow(() ->
pulsarLogCollectClient.consume(logs));
pulsarLogCollectClient.close();
}
}
-
diff --git
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-rabbitmq/src/main/java/org/apache/shenyu/plugin/logging/rabbitmq/client/RabbitmqLogCollectClient.java
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-rabbitmq/src/main/java/org/apache/shenyu/plugin/logging/rabbitmq/client/RabbitmqLogCollectClient.java
index d652688ea2..b599be56e2 100644
---
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-rabbitmq/src/main/java/org/apache/shenyu/plugin/logging/rabbitmq/client/RabbitmqLogCollectClient.java
+++
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-rabbitmq/src/main/java/org/apache/shenyu/plugin/logging/rabbitmq/client/RabbitmqLogCollectClient.java
@@ -50,15 +50,21 @@ public class RabbitmqLogCollectClient extends
AbstractLogConsumeClient<RabbitmqL
private String routingKey;
+ /**
+ * Initialize RabbitMQ client.
+ *
+ * @param config RabbitMQ configuration
+ * @return true if the client was initialized successfully
+ */
@Override
- public void initClient0(@NonNull final
RabbitmqLogCollectConfig.RabbitmqLogConfig config) {
+ public boolean initClient0(@NonNull final
RabbitmqLogCollectConfig.RabbitmqLogConfig config) {
if (StringUtils.isBlank(config.getHost())
|| Objects.isNull(config.getPort())
|| StringUtils.isBlank(config.getExchangeName())
|| StringUtils.isBlank(config.getQueueName())
|| StringUtils.isBlank(config.getExchangeType())) {
LOG.error("rabbitmq prop is empty. failed init rabbit producer");
- return;
+ return false;
}
String queueName = config.getQueueName();
@@ -82,14 +88,22 @@ public class RabbitmqLogCollectClient extends
AbstractLogConsumeClient<RabbitmqL
LOG.info("init rabbitmqLogCollectClient success");
} catch (IOException e) {
LOG.error("failed to initialize Rabbitmq connection", e);
+ closeAfterFailedInitialization();
+ return false;
} catch (TimeoutException e) {
LOG.error("failed to connect rabbitmq, connect timeout", e);
+ closeAfterFailedInitialization();
+ return false;
}
-
+ return true;
}
@Override
public void consume0(@NonNull final List<ShenyuRequestLog> logs) {
+ if (Objects.isNull(channel)) {
+ LOG.warn("RabbitMQ channel is not initialized.");
+ return;
+ }
logs.forEach(log -> {
try {
channel.basicPublish(exchangeName, routingKey,
MessageProperties.PERSISTENT_TEXT_PLAIN, buildLogMessageBytes(log));
@@ -136,4 +150,12 @@ public class RabbitmqLogCollectClient extends
AbstractLogConsumeClient<RabbitmqL
LOG.error("failed to close RabbitMQ connection", e);
}
}
+
+ private void closeAfterFailedInitialization() {
+ try {
+ close0();
+ } catch (Exception e) {
+ LOG.error("failed to clean up RabbitMQ resources after
initialization failure", e);
+ }
+ }
}
diff --git
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-rocketmq/src/main/java/org/apache/shenyu/plugin/logging/rocketmq/client/RocketMQLogCollectClient.java
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-rocketmq/src/main/java/org/apache/shenyu/plugin/logging/rocketmq/client/RocketMQLogCollectClient.java
index 73b22f3184..9240808c8e 100644
---
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-rocketmq/src/main/java/org/apache/shenyu/plugin/logging/rocketmq/client/RocketMQLogCollectClient.java
+++
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-rocketmq/src/main/java/org/apache/shenyu/plugin/logging/rocketmq/client/RocketMQLogCollectClient.java
@@ -61,16 +61,17 @@ public class RocketMQLogCollectClient extends
AbstractLogConsumeClient<RocketMQL
* init producer.
*
* @param config rocketmq props
+ * @return true if the client was initialized successfully
*/
@Override
- public void initClient0(@NonNull final
RocketMQLogCollectConfig.RocketMQLogConfig config) {
+ public boolean initClient0(@NonNull final
RocketMQLogCollectConfig.RocketMQLogConfig config) {
String topic = config.getTopic();
String nameserverAddress = config.getNamesrvAddr();
String producerGroup = config.getProducerGroup();
producerGroup =
Optional.ofNullable(producerGroup).orElse(DEFAULT_PRODUCER_GROUP);
if (StringUtils.isBlank(topic) ||
StringUtils.isBlank(nameserverAddress)) {
LOG.error("init RocketMQLogCollectClient error, please check topic
or nameserverAddress");
- return;
+ return false;
}
this.topic = topic;
producer = new DefaultMQProducer(producerGroup, getAclRPCHook(config));
@@ -82,7 +83,10 @@ public class RocketMQLogCollectClient extends
AbstractLogConsumeClient<RocketMQL
LOG.info("init RocketMQLogCollectClient success");
} catch (Exception e) {
LOG.error("init RocketMQLogCollectClient error", e);
+ producer.shutdown();
+ return false;
}
+ return true;
}
/**
@@ -105,6 +109,10 @@ public class RocketMQLogCollectClient extends
AbstractLogConsumeClient<RocketMQL
*/
@Override
public void consume0(@NonNull final List<ShenyuRequestLog> logs) {
+ if (Objects.isNull(producer)) {
+ LOG.warn("RocketMQ producer is not initialized.");
+ return;
+ }
logs.forEach(log -> {
String logTopic =
Optional.ofNullable(LoggingRocketMQPluginDataHandler.getSelectApiConfigMap().get(log.getSelectorId()))
.map(apiConfig ->
StringUtils.defaultIfBlank(apiConfig.getTopic(), topic)
diff --git
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-tencent-cls/src/main/java/org/apache/shenyu/plugin/tencent/cls/client/TencentClsLogCollectClient.java
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-tencent-cls/src/main/java/org/apache/shenyu/plugin/tencent/cls/client/TencentClsLogCollectClient.java
index c76b3a2091..d4d071c2b0 100644
---
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-tencent-cls/src/main/java/org/apache/shenyu/plugin/tencent/cls/client/TencentClsLogCollectClient.java
+++
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-tencent-cls/src/main/java/org/apache/shenyu/plugin/tencent/cls/client/TencentClsLogCollectClient.java
@@ -64,16 +64,17 @@ public class TencentClsLogCollectClient extends
AbstractLogConsumeClient<Tencent
* init Tencent cls client.
*
* @param tencentClsLogConfig shenyu log config
+ * @return true if the client was initialized successfully
*/
@Override
- public void initClient0(@NonNull final
TencentLogCollectConfig.TencentClsLogConfig tencentClsLogConfig) {
+ public boolean initClient0(@NonNull final
TencentLogCollectConfig.TencentClsLogConfig tencentClsLogConfig) {
String secretId = tencentClsLogConfig.getSecretId();
String secretKey = tencentClsLogConfig.getSecretKey();
String endpoint = tencentClsLogConfig.getEndpoint();
this.topic = tencentClsLogConfig.getTopic();
if (StringUtils.isBlank(secretId) || StringUtils.isBlank(secretKey) ||
StringUtils.isBlank(topic) || StringUtils.isBlank(endpoint)) {
LOG.error("init Tencent cls client error, please check secretId,
secretKey, topic or host");
- return;
+ return false;
}
// init AsyncProducerConfig, AsyncProducerClient
@@ -96,7 +97,10 @@ public class TencentClsLogCollectClient extends
AbstractLogConsumeClient<Tencent
client = new AsyncProducerClient(config);
} catch (Exception e) {
LOG.warn("TencentClsLogCollectClient initClient error message:{}",
e.getMessage());
+ threadExecutor.shutdownNow();
+ return false;
}
+ return true;
}
/**
@@ -106,6 +110,10 @@ public class TencentClsLogCollectClient extends
AbstractLogConsumeClient<Tencent
*/
@Override
public void consume0(@NonNull final List<ShenyuRequestLog> logs) {
+ if (Objects.isNull(client)) {
+ LOG.warn("Tencent CLS client is not initialized.");
+ return;
+ }
logs.forEach(this::sendLog);
}
