You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@seata.apache.org by ji...@apache.org on 2026/09/28 07:04:42 UTC
(incubator-seata) branch 2.x updated: test: verify Logstash appender delivery through a local TCP sink (#8271)
This is an automated email from the ASF dual-hosted git repository.
funky-eyes pushed a commit to branch 2.x
in repository https://gitbox.apache.org/repos/asf/incubator-seata.git
The following commit(s) were added to refs/heads/2.x by this push:
new 69bde83bc8 test: verify Logstash appender delivery through a local TCP sink (#8271)
69bde83bc8 is described below
commit 69bde83bc890f3279e50e67a867cc913b0bcd9ac
Author: jimin <sl...@163.com>
AuthorDate: Mon Sep 28 15:04:36 2026 +0800
test: verify Logstash appender delivery through a local TCP sink (#8271)
---
changes/en-us/2.x.md | 2 +-
changes/zh-cn/2.x.md | 4 +-
.../apache/seata/server/logging/AppenderTest.java | 103 ++++++++++++++++++++-
3 files changed, 105 insertions(+), 4 deletions(-)
diff --git a/changes/en-us/2.x.md b/changes/en-us/2.x.md
index fd4a3873e8..4d16d4cdb2 100644
--- a/changes/en-us/2.x.md
+++ b/changes/en-us/2.x.md
@@ -52,8 +52,8 @@ Add changes here for all PR submitted to the 2.x branch.
### test:
+- [[#8271](https://github.com/apache/incubator-seata/pull/8271)] verify Logstash appender delivery with a local TCP test sink
- [[#8261](https://github.com/apache/incubator-seata/pull/8261)] isolate TM bootstrap option tests from external server connections
-
- [[#8146](https://github.com/apache/incubator-seata/pull/8146)] test: cover rm-datasource resource id initializers
- [[#8175](https://github.com/apache/incubator-seata/pull/8175)] fix flaky testCachedConfigurationChangeListener due to async race condition
- [[#8181](https://github.com/apache/incubator-seata/pull/8181)] fix AuthControllerWithRandomPasswordTest flaky failure caused by Spring Boot test context caching
diff --git a/changes/zh-cn/2.x.md b/changes/zh-cn/2.x.md
index cbc3911b82..38c3c962ac 100644
--- a/changes/zh-cn/2.x.md
+++ b/changes/zh-cn/2.x.md
@@ -30,7 +30,7 @@
- [[#8278](https://github.com/apache/incubator-seata/pull/8278)] 保护 Fastjson2 JSONB 并发解码，避免 RPC 与 UndoLog 引用字段丢失
- [[#8138](https://github.com/apache/incubator-seata/pull/8138)] 修复 TCC fence 清理时误删同一全局事务中仍在进行(TRIED)或未过期的分支记录的问题
- [[#8191](https://github.com/apache/incubator-seata/pull/8191)] 修复默认本地地址缓存后 NetUtil 参数化查询绕过 ignoredInterfaces 和 preferredNetworks 的问题
-- [#8145](https://github.com/apache/incubator-seata/pull/8145) 修复达梦(DM)数据库下全局锁批量获取被误判为失败的问题
+- [[#8145](https://github.com/apache/incubator-seata/pull/8145)] 修复达梦(DM)数据库下全局锁批量获取被误判为失败的问题
- [[#8157](https://github.com/apache/incubator-seata/pull/8157)] 修复 Spring Boot 4.x 下由于硬编码 `DataSourceAutoConfiguration` 类引用导致 Saga 自动配置被跳过的问题
- [[#8160](https://github.com/apache/incubator-seata/pull/8160)] 修复 fastjson2 JSONB 并发反序列化问题
- [[#8179](https://github.com/apache/incubator-seata/pull/8179)] 将 console 模块的 java.version 设置为 17 以修复编译错误
@@ -51,8 +51,8 @@
### test:
+- [[#8271](https://github.com/apache/incubator-seata/pull/8271)] 使用本地 TCP 接收端校验 Logstash appender 的实际日志投递
- [[#8261](https://github.com/apache/incubator-seata/pull/8261)] 隔离 TM 启动参数测试对外部服务连接的依赖
-
- [[#8146](https://github.com/apache/incubator-seata/pull/8146)] test: 覆盖 rm-datasource resource id initializers
- [[#8175](https://github.com/apache/incubator-seata/pull/8175)] 修复 testCachedConfigurationChangeListener 因异步竞态条件导致的偶发失败
- [[#8181](https://github.com/apache/incubator-seata/pull/8181)] 修复 AuthControllerWithRandomPasswordTest 因 Spring Boot 测试上下文缓存导致的偶发失败
diff --git a/server/src/test/java/org/apache/seata/server/logging/AppenderTest.java b/server/src/test/java/org/apache/seata/server/logging/AppenderTest.java
index bf9a1a3fd2..c885617eb2 100644
--- a/server/src/test/java/org/apache/seata/server/logging/AppenderTest.java
+++ b/server/src/test/java/org/apache/seata/server/logging/AppenderTest.java
@@ -16,32 +16,120 @@
*/
package org.apache.seata.server.logging;
+import ch.qos.logback.classic.Level;
import ch.qos.logback.classic.LoggerContext;
import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.classic.spi.LoggingEvent;
import ch.qos.logback.core.Appender;
import com.github.danielwegener.logback.kafka.KafkaAppender;
import net.logstash.logback.appender.LogstashTcpSocketAppender;
import org.apache.seata.server.BaseSpringBootTest;
import org.apache.seata.server.logging.logback.appender.MetricLogbackAppender;
+import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.slf4j.LoggerFactory;
import org.springframework.test.context.TestPropertySource;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
import java.lang.reflect.Field;
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.nio.charset.StandardCharsets;
import java.util.Iterator;
+import java.util.Set;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
@TestPropertySource(
properties = {
"logging.extend.logstash-appender.enabled=true",
+ "logging.extend.logstash-appender.destination=${seata.test.logstash.destination}",
"logging.extend.kafka-appender.enabled=true",
"logging.extend.kafka-appender.topic=test",
"logging.extend.metric-appender.enabled=true"
})
public class AppenderTest extends BaseSpringBootTest {
+ private static final String DESTINATION_PROPERTY = "seata.test.logstash.destination";
+ private static final String ORIGINAL_DESTINATION = System.getProperty(DESTINATION_PROPERTY);
+ private static final BlockingQueue<String> RECEIVED = new LinkedBlockingQueue<>();
+ private static final AtomicReference<Throwable> SINK_FAILURE = new AtomicReference<>();
+ private static final Set<Socket> CONNECTIONS = ConcurrentHashMap.newKeySet();
+ private static final ExecutorService SINK_THREADS = Executors.newCachedThreadPool(task -> {
+ Thread thread = new Thread(task, "logstash-test-sink");
+ thread.setDaemon(true);
+ return thread;
+ });
+ private static final ServerSocket SINK = startSink();
+
+ private static ServerSocket startSink() {
+ try {
+ ServerSocket server = new ServerSocket(0, 50, InetAddress.getByName("127.0.0.1"));
+ System.setProperty(DESTINATION_PROPERTY, "127.0.0.1:" + server.getLocalPort());
+ SINK_THREADS.submit(() -> {
+ try {
+ while (!server.isClosed()) {
+ Socket connection = server.accept();
+ CONNECTIONS.add(connection);
+ SINK_THREADS.submit(() -> {
+ try (Socket socket = connection;
+ BufferedReader reader = new BufferedReader(
+ new InputStreamReader(socket.getInputStream(), StandardCharsets.UTF_8))) {
+ String line;
+ while ((line = reader.readLine()) != null) {
+ RECEIVED.add(line);
+ }
+ } catch (IOException error) {
+ if (!server.isClosed()) {
+ SINK_FAILURE.compareAndSet(null, error);
+ }
+ } finally {
+ CONNECTIONS.remove(connection);
+ }
+ });
+ }
+ } catch (IOException error) {
+ if (!server.isClosed()) {
+ SINK_FAILURE.compareAndSet(null, error);
+ }
+ }
+ });
+ return server;
+ } catch (IOException error) {
+ throw new IllegalStateException("Cannot start Logstash test sink", error);
+ }
+ }
+
+ @AfterAll
+ static void closeSink() throws Exception {
+ LoggerContext context = (LoggerContext) LoggerFactory.getILoggerFactory();
+ Appender<ILoggingEvent> appender = context.getLogger("ROOT").getAppender("LOGSTASH");
+ if (appender != null) {
+ appender.stop();
+ }
+ SINK.close();
+ for (Socket connection : CONNECTIONS) {
+ connection.close();
+ }
+ SINK_THREADS.shutdownNow();
+ Assertions.assertTrue(SINK_THREADS.awaitTermination(5, TimeUnit.SECONDS));
+ if (ORIGINAL_DESTINATION == null) {
+ System.clearProperty(DESTINATION_PROPERTY);
+ } else {
+ System.setProperty(DESTINATION_PROPERTY, ORIGINAL_DESTINATION);
+ }
+ }
@Test
- public void testAppenderEnabled() {
+ public void testAppenderEnabled() throws InterruptedException {
LoggerContext lc = (LoggerContext) LoggerFactory.getILoggerFactory();
Iterator<Appender<ILoggingEvent>> appenderIterator =
lc.getLogger("ROOT").iteratorForAppenders();
@@ -85,6 +173,19 @@ public class AppenderTest extends BaseSpringBootTest {
Assertions.assertTrue(kafkaFound);
Assertions.assertTrue(metricFound);
Assertions.assertTrue(logstashFound);
+ String marker = "logstash-fixture-marker";
+ lc.getLogger("ROOT")
+ .getAppender("LOGSTASH")
+ .doAppend(new LoggingEvent(
+ getClass().getName(), lc.getLogger(getClass()), Level.INFO, marker, null, null));
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+ boolean delivered = false;
+ while (!delivered && System.nanoTime() < deadline) {
+ String line = RECEIVED.poll(100, TimeUnit.MILLISECONDS);
+ delivered = line != null && line.contains(marker);
+ }
+ Assertions.assertNull(SINK_FAILURE.get());
+ Assertions.assertTrue(delivered, "The configured Logstash appender must deliver the event to the local sink");
}
private static Field getDeclaredFieldRecursive(Class<?> clazz, String fieldName) throws NoSuchFieldException {
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org