This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 1b44799fa97 [Pipe] Reduce AirGap sink reconnect log noise (#18243)
1b44799fa97 is described below
commit 1b44799fa9733d1a57d26ab2d2d972337372fa34
Author: Caideyipi <[email protected]>
AuthorDate: Mon Jul 20 15:24:25 2026 +0800
[Pipe] Reduce AirGap sink reconnect log noise (#18243)
---
.../apache/iotdb/commons/i18n/PipeMessages.java | 8 ++--
.../apache/iotdb/commons/i18n/PipeMessages.java | 8 ++--
.../task/subtask/PipeAbstractSinkSubtask.java | 9 +++--
.../commons/pipe/resource/log/PipeLogger.java | 4 +-
.../pipe/sink/protocol/IoTDBAirGapSink.java | 12 ++++--
.../commons/pipe/resource/PipeLoggerTest.java | 47 +++++++++-------------
6 files changed, 42 insertions(+), 46 deletions(-)
diff --git
a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java
b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java
index aa04fcafbd9..e69dc15c1dd 100644
---
a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java
+++
b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java
@@ -432,11 +432,11 @@ public final class PipeMessages {
public static final String NON_CRITICAL_EXCEPTION_WILL_THROW_CRITICAL =
"A non PipeRuntimeSinkCriticalException occurred, will throw a
PipeRuntimeSinkCriticalException.";
public static final String PIPE_CONNECTION_EXCEPTION_RETRYING =
- "PipeConnectionException occurred, %s retries to handshake with the
target system.";
+ "PipeConnectionException occurred, %s retries to handshake with the
target system. Root cause: %s.";
public static final String HANDSHAKE_SUCCESS = "{} handshakes with the
target system successfully.";
public static final String HANDSHAKE_FAILED_RETRYING =
"{} failed to handshake with the target system for {} times, "
- + "will retry at most {} times.";
+ + "will retry at most {} times. Root cause: {}.";
public static final String INTERRUPTED_WHILE_SLEEPING_RETRY_HANDSHAKE =
"Interrupted while sleeping, will retry to handshake with the target
system.";
public static final String HANDSHAKE_FAILED_STOPPING =
@@ -511,9 +511,9 @@ public final class PipeMessages {
public static final String FAILED_TO_CONNECT_TO_TARGET =
"Failed to connect to target server ip: {}, port: {}, because: {}.
Ignore it.";
public static final String HANDSHAKE_ERROR_RECEIVING_END =
- "Handshake error occurs. It may be caused by an error on the receiving
end. Ignore it.";
+ "Handshake error occurs. It may be caused by an error on the receiving
end. Ignore it. Root cause: {}.";
public static final String HANDSHAKE_ERROR_WITH_TARGET =
- "Handshake error with target server, socket: %s";
+ "Handshake error with target server, endpoint: %s";
public static final String HANDSHAKE_SUCCESS_SOCKET = "Handshake success.
Socket: {}";
public static final String FAILED_TO_CLOSE_CLIENT = "Failed to close client
{}.";
public static final String UNKNOWN_LOAD_BALANCE_STRATEGY =
diff --git
a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java
b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java
index 9c808eecbf8..ca3243cdc01 100644
---
a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java
+++
b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java
@@ -415,10 +415,10 @@ public final class PipeMessages {
public static final String NON_CRITICAL_EXCEPTION_WILL_THROW_CRITICAL =
"发生非 PipeRuntimeSinkCriticalException，将抛出
PipeRuntimeSinkCriticalException。";
public static final String PIPE_CONNECTION_EXCEPTION_RETRYING =
- "发生 PipeConnectionException，%s 正在重试与目标系统握手。";
+ "发生 PipeConnectionException，%s 正在重试与目标系统握手。根因：%s。";
public static final String HANDSHAKE_SUCCESS = "{} 与目标系统握手成功。";
public static final String HANDSHAKE_FAILED_RETRYING =
- "{} 与目标系统握手失败，已重试 {} 次，最多重试 {} 次。";
+ "{} 与目标系统握手失败，已重试 {} 次，最多重试 {} 次。根因：{}。";
public static final String INTERRUPTED_WHILE_SLEEPING_RETRY_HANDSHAKE =
"休眠时被中断，将重试与目标系统握手。";
public static final String HANDSHAKE_FAILED_STOPPING =
@@ -491,9 +491,9 @@ public final class PipeMessages {
public static final String FAILED_TO_CONNECT_TO_TARGET =
"连接目标服务器失败（ip：{}，端口：{}），原因：{}。已忽略。";
public static final String HANDSHAKE_ERROR_RECEIVING_END =
- "握手错误，可能由接收端错误引起。已忽略。";
+ "握手错误，可能由接收端错误引起。已忽略。根因：{}。";
public static final String HANDSHAKE_ERROR_WITH_TARGET =
- "与目标服务器握手失败，socket：%s";
+ "与目标服务器握手失败，endpoint：%s";
public static final String HANDSHAKE_SUCCESS_SOCKET = "握手成功。Socket：{}";
public static final String FAILED_TO_CLOSE_CLIENT = "关闭客户端 {} 失败。";
public static final String UNKNOWN_LOAD_BALANCE_STRATEGY =
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java
index 1b51e460956..ceaade1bb1f 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java
@@ -186,9 +186,9 @@ public abstract class PipeAbstractSinkSubtask extends
PipeReportableSubtask {
private boolean onPipeConnectionException(final Throwable throwable) {
PipeLogger.log(
LOGGER::warn,
- throwable,
PipeMessages.PIPE_CONNECTION_EXCEPTION_RETRYING,
- outputPipeSink.getClass().getName());
+ outputPipeSink.getClass().getName(),
+ ErrorHandlingCommonUtils.getRootCause(throwable).toString());
int retry = 0;
while (retry < MAX_RETRY_TIMES) {
@@ -200,12 +200,13 @@ public abstract class PipeAbstractSinkSubtask extends
PipeReportableSubtask {
break;
} catch (final Exception e) {
retry++;
- LOGGER.warn(
+ PipeLogger.log(
+ LOGGER::warn,
PipeMessages.HANDSHAKE_FAILED_RETRYING,
outputPipeSink.getClass().getName(),
retry,
MAX_RETRY_TIMES,
- e);
+ ErrorHandlingCommonUtils.getRootCause(e).toString());
try {
sleepIfNoHighPriorityTask(getHandshakeRetrySleepInterval(e, retry));
} catch (final InterruptedException interruptedException) {
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/resource/log/PipeLogger.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/resource/log/PipeLogger.java
index 720999aae97..b906fc432cd 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/resource/log/PipeLogger.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/resource/log/PipeLogger.java
@@ -26,9 +26,7 @@ import java.io.PrintStream;
import java.util.function.Consumer;
public class PipeLogger {
- private static PipePeriodicalLogger logger =
- (loggerFunction, rawMessage, formatter) ->
-
loggerFunction.accept(LoggerPeriodicalLogReducer.formatMessage(rawMessage,
formatter));
+ private static PipePeriodicalLogger logger = LoggerPeriodicalLogReducer::log;
public static void log(
final Consumer<String> loggerFunction, final String rawMessage, final
Object... formatter) {
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBAirGapSink.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBAirGapSink.java
index 29a679e8037..8bcca0315f4 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBAirGapSink.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBAirGapSink.java
@@ -23,8 +23,10 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.i18n.PipeMessages;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
+import org.apache.iotdb.commons.pipe.resource.log.PipeLogger;
import
org.apache.iotdb.commons.pipe.sink.payload.airgap.AirGapELanguageConstant;
import org.apache.iotdb.commons.pipe.sink.payload.airgap.AirGapOneByteResponse;
+import org.apache.iotdb.commons.utils.ErrorHandlingCommonUtils;
import org.apache.iotdb.pipe.api.annotation.TableModel;
import org.apache.iotdb.pipe.api.annotation.TreeModel;
import
org.apache.iotdb.pipe.api.customizer.configuration.PipeConnectorRuntimeConfiguration;
@@ -181,7 +183,7 @@ public abstract class IoTDBAirGapSink extends IoTDBSink {
socket.connect(new InetSocketAddress(ip, port), handshakeTimeoutMs);
socket.setKeepAlive(true);
sockets.set(i, socket);
- LOGGER.info(PipeMessages.CONNECTED_TO_TARGET_SERVER, ip, port);
+ LOGGER.debug(PipeMessages.CONNECTED_TO_TARGET_SERVER, ip, port);
failLogTimes.remove(nodeUrls.get(i));
} catch (final Exception e) {
final TEndPoint endPoint = nodeUrls.get(i);
@@ -198,7 +200,10 @@ public abstract class IoTDBAirGapSink extends IoTDBSink {
sendHandshakeReq(socket);
isSocketAlive.set(i, true);
} catch (Exception e) {
- LOGGER.warn(PipeMessages.HANDSHAKE_ERROR_RECEIVING_END, e);
+ PipeLogger.log(
+ LOGGER::warn,
+ PipeMessages.HANDSHAKE_ERROR_RECEIVING_END,
+ ErrorHandlingCommonUtils.getRootCause(e).toString());
}
}
@@ -219,7 +224,8 @@ public abstract class IoTDBAirGapSink extends IoTDBSink {
if (!send(socket, generateHandShakeV2Payload())) {
supportModsIfIsDataNodeReceiver = false;
if (!send(socket, generateHandShakeV1Payload())) {
- throw new
PipeConnectionException(PipeMessages.HANDSHAKE_ERROR_WITH_TARGET + socket);
+ throw new PipeConnectionException(
+ String.format(PipeMessages.HANDSHAKE_ERROR_WITH_TARGET,
socket.getEndPoint()));
}
} else {
supportModsIfIsDataNodeReceiver = true;
diff --git
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/resource/PipeLoggerTest.java
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/resource/PipeLoggerTest.java
index d37c109c9f5..651bd552081 100644
---
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/resource/PipeLoggerTest.java
+++
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/resource/PipeLoggerTest.java
@@ -24,20 +24,27 @@ import
org.apache.iotdb.commons.pipe.resource.log.PipeLogger;
import org.junit.Assert;
import org.junit.Test;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
public class PipeLoggerTest {
+ @Test
+ public void testDefaultLoggerReducesDuplicateMessages() {
+ final AtomicInteger logCount = new AtomicInteger(0);
+ final String message = "PipeLoggerTest-" + System.nanoTime();
+
+ PipeLogger.log(ignored -> logCount.incrementAndGet(), message);
+ PipeLogger.log(ignored -> logCount.incrementAndGet(), message);
+
+ Assert.assertEquals(1, logCount.get());
+ }
+
@Test
public void testLogMessageWithPercent() {
final AtomicReference<String> message = new AtomicReference<>();
- setStringFormatLogger();
- try {
- PipeLogger.log(message::set, "data_{sink.password=%/broken}");
- } finally {
- setStringFormatLogger();
- }
+ PipeLogger.log(message::set, "data_{sink.password=%/broken}");
Assert.assertEquals("data_{sink.password=%/broken}", message.get());
}
@@ -46,12 +53,7 @@ public class PipeLoggerTest {
public void testLogMessageWithSlf4jPlaceholder() {
final AtomicReference<String> message = new AtomicReference<>();
- setStringFormatLogger();
- try {
- PipeLogger.log(message::set, "PipeLoggerCacheMaxSizeInBytes: {}", 1024);
- } finally {
- setStringFormatLogger();
- }
+ PipeLogger.log(message::set, "PipeLoggerCacheMaxSizeInBytes: {}", 1024);
Assert.assertEquals("PipeLoggerCacheMaxSizeInBytes: 1024", message.get());
}
@@ -60,24 +62,13 @@ public class PipeLoggerTest {
public void testLogThrowableWithPercentInStackTrace() {
final AtomicReference<String> message = new AtomicReference<>();
- setStringFormatLogger();
- try {
- PipeLogger.log(
- message::set,
- new RuntimeException("data_{sink.password=%/broken}"),
- "Failed to transfer event %s",
- "root.sg.d1");
- } finally {
- setStringFormatLogger();
- }
+ PipeLogger.log(
+ message::set,
+ new RuntimeException("data_{sink.password=%/broken}"),
+ "Failed to transfer event %s",
+ "root.sg.d1");
Assert.assertTrue(message.get().contains("Failed to transfer event
root.sg.d1"));
Assert.assertTrue(message.get().contains("data_{sink.password=%/broken}"));
}
-
- private void setStringFormatLogger() {
- PipeLogger.setLogger(
- (loggerFunction, rawMessage, formatter) ->
- loggerFunction.accept(String.format(rawMessage, formatter)));
- }
}
