This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new 4e68fb8e7e3 [Pipe] Reduce AirGap sink reconnect log noise (#18243)
(#18253)
4e68fb8e7e3 is described below
commit 4e68fb8e7e342246263f1b3fd7d3c224639a744e
Author: Caideyipi <[email protected]>
AuthorDate: Tue Jul 21 09:55:46 2026 +0800
[Pipe] Reduce AirGap sink reconnect log noise (#18243) (#18253)
---
.../task/subtask/PipeAbstractSinkSubtask.java | 14 ++--
.../commons/pipe/resource/log/PipeLogger.java | 4 +-
.../pipe/sink/protocol/IoTDBAirGapSink.java | 15 +++--
.../commons/pipe/resource/PipeLoggerTest.java | 74 ++++++++++++++++++++++
4 files changed, 93 insertions(+), 14 deletions(-)
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 eadd889287c..3593c96b5cd 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,10 +186,11 @@ public abstract class PipeAbstractSinkSubtask extends
PipeReportableSubtask {
* @return {@code true} if the {@link PipeSubtask} should be stopped, {@code
false} otherwise
*/
private boolean onPipeConnectionException(final Throwable throwable) {
- LOGGER.warn(
- "PipeConnectionException occurred, {} retries to handshake with the
target system.",
+ PipeLogger.log(
+ LOGGER::warn,
+ "PipeConnectionException occurred, {} retries to handshake with the
target system. Root cause: {}.",
outputPipeConnector.getClass().getName(),
- throwable);
+ ErrorHandlingCommonUtils.getRootCause(throwable).toString());
int retry = 0;
while (retry < MAX_RETRY_TIMES) {
@@ -203,13 +204,14 @@ public abstract class PipeAbstractSinkSubtask extends
PipeReportableSubtask {
break;
} catch (final Exception e) {
retry++;
- LOGGER.warn(
+ PipeLogger.log(
+ LOGGER::warn,
"{} failed to handshake with the target system for {} times, "
- + "will retry at most {} times.",
+ + "will retry at most {} times. Root cause: {}.",
outputPipeConnector.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 f2c54cd2c94..49caa1fc5ca 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
@@ -22,8 +22,10 @@ package org.apache.iotdb.commons.pipe.sink.protocol;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
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.customizer.configuration.PipeConnectorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.pipe.api.exception.PipeConnectionException;
@@ -182,7 +184,7 @@ public abstract class IoTDBAirGapSink extends IoTDBSink {
socket.connect(new InetSocketAddress(ip, port), handshakeTimeoutMs);
socket.setKeepAlive(true);
sockets.set(i, socket);
- LOGGER.info("Successfully connected to target server ip: {}, port:
{}.", ip, port);
+ LOGGER.debug("Successfully connected to target server ip: {}, port:
{}.", ip, port);
failLogTimes.remove(nodeUrls.get(i));
} catch (final Exception e) {
final TEndPoint endPoint = nodeUrls.get(i);
@@ -203,9 +205,10 @@ public abstract class IoTDBAirGapSink extends IoTDBSink {
sendHandshakeReq(socket);
isSocketAlive.set(i, true);
} catch (Exception e) {
- LOGGER.warn(
- "Handshake error occurs. It may be caused by an error on the
receiving end. Ignore it.",
- e);
+ PipeLogger.log(
+ LOGGER::warn,
+ "Handshake error occurs. It may be caused by an error on the
receiving end. Ignore it. Root cause: {}.",
+ ErrorHandlingCommonUtils.getRootCause(e).toString());
}
}
@@ -226,7 +229,9 @@ public abstract class IoTDBAirGapSink extends IoTDBSink {
if (!send(socket, generateHandShakeV2Payload())) {
supportModsIfIsDataNodeReceiver = false;
if (!send(socket, generateHandShakeV1Payload())) {
- throw new PipeConnectionException("Handshake error with target server,
socket: " + socket);
+ throw new PipeConnectionException(
+ String.format(
+ "Handshake error with target server, endpoint: %s",
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
new file mode 100644
index 00000000000..651bd552081
--- /dev/null
+++
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/resource/PipeLoggerTest.java
@@ -0,0 +1,74 @@
+/*
+ * 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.iotdb.commons.pipe.resource;
+
+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<>();
+
+ PipeLogger.log(message::set, "data_{sink.password=%/broken}");
+
+ Assert.assertEquals("data_{sink.password=%/broken}", message.get());
+ }
+
+ @Test
+ public void testLogMessageWithSlf4jPlaceholder() {
+ final AtomicReference<String> message = new AtomicReference<>();
+
+ PipeLogger.log(message::set, "PipeLoggerCacheMaxSizeInBytes: {}", 1024);
+
+ Assert.assertEquals("PipeLoggerCacheMaxSizeInBytes: 1024", message.get());
+ }
+
+ @Test
+ public void testLogThrowableWithPercentInStackTrace() {
+ final AtomicReference<String> message = new AtomicReference<>();
+
+ 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}"));
+ }
+}
