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 8d5da83f0df Preserve Pipe sink exception root causes (#18335)
8d5da83f0df is described below
commit 8d5da83f0dfb33da552606badb2d735d7f21352e
Author: Caideyipi <[email protected]>
AuthorDate: Wed Jul 29 10:06:52 2026 +0800
Preserve Pipe sink exception root causes (#18335)
---
.../agent/task/subtask/sink/PipeSinkSubtask.java | 2 +-
.../task/subtask/sink/PipeSinkSubtaskTest.java | 93 ++++++++++++++++++++++
.../pipe/PipeRuntimeCriticalException.java | 4 +
.../exception/pipe/PipeRuntimeException.java | 4 +
.../pipe/PipeRuntimeSinkCriticalException.java | 4 +
.../task/subtask/PipeAbstractSinkSubtask.java | 10 ++-
6 files changed, 113 insertions(+), 4 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java
index dc1db3386dd..365f211fb51 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java
@@ -575,7 +575,7 @@ public class PipeSinkSubtask extends
PipeAbstractSinkSubtask {
@Override
protected String getRootCause(final Throwable throwable) {
- return ErrorHandlingCommonUtils.getRootCause(throwable).getMessage();
+ return ErrorHandlingCommonUtils.getRootCause(throwable).toString();
}
@Override
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskTest.java
index 8af86f1b128..673fa3e91e4 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskTest.java
@@ -20,10 +20,14 @@
package org.apache.iotdb.db.pipe.agent.task.subtask.sink;
import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.commons.exception.pipe.PipeRuntimeException;
+import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeSinkCriticalException;
import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeSinkNonReportTimeConfigurableException;
import
org.apache.iotdb.commons.pipe.agent.task.connection.UnboundedBlockingPendingQueue;
import org.apache.iotdb.commons.pipe.agent.task.progress.CommitterKey;
+import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import
org.apache.iotdb.commons.pipe.sink.protocol.PipeConnectorWithEventDiscard;
+import org.apache.iotdb.commons.utils.ErrorHandlingCommonUtils;
import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent;
import org.apache.iotdb.pipe.api.PipeConnector;
import
org.apache.iotdb.pipe.api.customizer.configuration.PipeConnectorRuntimeConfiguration;
@@ -269,6 +273,63 @@ public class PipeSinkSubtaskTest {
}
}
+ @Test
+ public void
testTransferExceptionWithNullRootCauseMessageIncludesExceptionType()
+ throws Exception {
+ final PipeConnector connector = mock(PipeConnector.class);
+ final UnboundedBlockingPendingQueue<Event> pendingQueue =
+ mock(UnboundedBlockingPendingQueue.class);
+ final Event event = mock(Event.class);
+ final NullPointerException rootCause = new NullPointerException();
+
+ when(pendingQueue.waitedPoll()).thenReturn(event);
+ doThrow(rootCause).when(connector).transfer(any(Event.class));
+
+ final PipeSinkSubtask subtask =
+ new PipeSinkSubtask(
+ "PipeSinkSubtaskTest",
+ System.currentTimeMillis(),
+ "data_test",
+ "data_test",
+ 0,
+ pendingQueue,
+ connector);
+
+ try {
+ subtask.executeOnce();
+ Assert.fail();
+ } catch (final PipeException e) {
+ Assert.assertTrue(e.getMessage().contains("root cause:
java.lang.NullPointerException"));
+ Assert.assertFalse(e.getMessage().contains("root cause: null"));
+ Assert.assertSame(rootCause, e.getCause());
+ } finally {
+ subtask.close();
+ }
+ }
+
+ @Test
+ public void testOnFailurePreservesOriginalCause() {
+ final PipeConnector connector = mock(PipeConnector.class);
+ final UnboundedBlockingPendingQueue<Event> pendingQueue =
+ mock(UnboundedBlockingPendingQueue.class);
+ final EnrichedEvent event = mock(EnrichedEvent.class);
+ final NullPointerException rootCause = new NullPointerException();
+ final PipeException failure = new PipeException("transfer failed",
rootCause);
+ final CapturingPipeSinkSubtask subtask = new
CapturingPipeSinkSubtask(pendingQueue, connector);
+
+ subtask.prepareFailure(event);
+ try {
+ subtask.onFailure(failure);
+
+ final PipeRuntimeException reportedException =
subtask.getReportedException();
+ Assert.assertTrue(reportedException instanceof
PipeRuntimeSinkCriticalException);
+ Assert.assertSame(failure, reportedException.getCause());
+ Assert.assertSame(rootCause,
ErrorHandlingCommonUtils.getRootCause(reportedException));
+ } finally {
+ subtask.close();
+ }
+ }
+
@Test
public void testHeartbeatPreservesReceiverProbeDelayException() throws
Exception {
final long originalSleepIntervalInitMs =
@@ -310,6 +371,38 @@ public class PipeSinkSubtaskTest {
}
}
+ private static class CapturingPipeSinkSubtask extends PipeSinkSubtask {
+
+ private PipeRuntimeException reportedException;
+
+ private CapturingPipeSinkSubtask(
+ final UnboundedBlockingPendingQueue<Event> pendingQueue, final
PipeConnector connector) {
+ super(
+ "PipeSinkSubtaskTest",
+ System.currentTimeMillis(),
+ "data_test",
+ "data_test",
+ 0,
+ pendingQueue,
+ connector);
+ }
+
+ private void prepareFailure(final EnrichedEvent event) {
+ setLastEvent(event);
+ setLastExceptionEvent(event);
+ retryCount.set(MAX_RETRY_TIMES);
+ }
+
+ private PipeRuntimeException getReportedException() {
+ return reportedException;
+ }
+
+ @Override
+ protected void report(final EnrichedEvent event, final
PipeRuntimeException exception) {
+ reportedException = exception;
+ }
+ }
+
private static class BlockingHandshakeConnector
implements PipeConnector, PipeConnectorWithEventDiscard {
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/pipe/PipeRuntimeCriticalException.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/pipe/PipeRuntimeCriticalException.java
index 061c9d275ea..611ada309ee 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/pipe/PipeRuntimeCriticalException.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/pipe/PipeRuntimeCriticalException.java
@@ -36,6 +36,10 @@ public class PipeRuntimeCriticalException extends
PipeRuntimeException {
super(message);
}
+ public PipeRuntimeCriticalException(final String message, final Throwable
cause) {
+ super(message, cause);
+ }
+
public PipeRuntimeCriticalException(final String message, final long
timeStamp) {
super(message, timeStamp);
}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/pipe/PipeRuntimeException.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/pipe/PipeRuntimeException.java
index 24fa4a1edda..9e597586d5f 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/pipe/PipeRuntimeException.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/pipe/PipeRuntimeException.java
@@ -32,6 +32,10 @@ public abstract class PipeRuntimeException extends
PipeException {
super(message);
}
+ protected PipeRuntimeException(final String message, final Throwable cause) {
+ super(message, cause);
+ }
+
protected PipeRuntimeException(final String message, final long timeStamp) {
super(message, timeStamp);
}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/pipe/PipeRuntimeSinkCriticalException.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/pipe/PipeRuntimeSinkCriticalException.java
index bb2715246f7..530b64aadd0 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/pipe/PipeRuntimeSinkCriticalException.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/pipe/PipeRuntimeSinkCriticalException.java
@@ -36,6 +36,10 @@ public class PipeRuntimeSinkCriticalException extends
PipeRuntimeCriticalExcepti
super(message);
}
+ public PipeRuntimeSinkCriticalException(final String message, final
Throwable cause) {
+ super(message, cause);
+ }
+
public PipeRuntimeSinkCriticalException(final String message, final long
timeStamp) {
super(message, timeStamp);
}
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 ceaade1bb1f..99ec79badab 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
@@ -176,7 +176,10 @@ public abstract class PipeAbstractSinkSubtask extends
PipeReportableSubtask {
// Print stack trace for better debugging
PipeLogger.log(
LOGGER::warn, throwable,
PipeMessages.NON_CRITICAL_EXCEPTION_WILL_THROW_CRITICAL);
- super.onFailure(new
PipeRuntimeSinkCriticalException(throwable.getMessage()));
+ super.onFailure(
+ new PipeRuntimeSinkCriticalException(
+ throwable.getMessage() != null ? throwable.getMessage() :
throwable.toString(),
+ throwable));
}
}
@@ -227,7 +230,8 @@ public abstract class PipeAbstractSinkSubtask extends
PipeReportableSubtask {
new PipeRuntimeSinkCriticalException(
throwable.getMessage()
+ PipeMessages.EXCEPTION_ROOT_CAUSE_A22E94DE
- + getRootCause(throwable)));
+ + getRootCause(throwable),
+ throwable));
LOGGER.warn(
PipeMessages.HANDSHAKE_FAILED_STOPPING,
outputPipeSink.getClass().getName(),
@@ -374,7 +378,7 @@ public abstract class PipeAbstractSinkSubtask extends
PipeReportableSubtask {
event instanceof EnrichedEvent
? ((EnrichedEvent) event).coreReportMessage()
: event,
- ErrorHandlingCommonUtils.getRootCause(e).getMessage()),
+ ErrorHandlingCommonUtils.getRootCause(e).toString()),
e);
} else {
LOGGER.info(
