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 ed87f2d0e83 [To dev/1.3] [Pipe] Preserve sink exception root causes
(#18335) (#18342)
ed87f2d0e83 is described below
commit ed87f2d0e83c90b8a364d367be05f45000de86b2
Author: Caideyipi <[email protected]>
AuthorDate: Thu Jul 30 10:44:12 2026 +0800
[To dev/1.3] [Pipe] Preserve sink exception root causes (#18335) (#18342)
* Preserve Pipe sink exception root causes (#18335)
* Fix PipeSinkSubtaskTest compilation
---
.../agent/task/subtask/sink/PipeSinkSubtask.java | 2 +-
.../task/subtask/sink/PipeSinkSubtaskTest.java | 92 ++++++++++++++++++++++
.../pipe/PipeRuntimeCriticalException.java | 4 +
.../exception/pipe/PipeRuntimeException.java | 4 +
.../pipe/PipeRuntimeSinkCriticalException.java | 4 +
.../task/subtask/PipeAbstractSinkSubtask.java | 9 ++-
6 files changed, 111 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 eed4b729ed2..91a588ce699 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
@@ -504,7 +504,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 6f14de41471..8960e7fea96 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;
@@ -32,6 +36,7 @@ import
org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.pipe.api.event.Event;
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
import org.apache.iotdb.pipe.api.exception.PipeConnectionException;
+import org.apache.iotdb.pipe.api.exception.PipeException;
import org.junit.Assert;
import org.junit.Test;
@@ -227,6 +232,62 @@ 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",
+ 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 =
@@ -267,6 +328,37 @@ 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",
+ 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 c6db38e23fa..a52cfc4a18b 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
@@ -35,6 +35,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 ccc91f7fef1..2211458324b 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
@@ -35,6 +35,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 3593c96b5cd..f3c1d72605c 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
@@ -178,7 +178,10 @@ public abstract class PipeAbstractSinkSubtask extends
PipeReportableSubtask {
LOGGER::warn,
throwable,
"A non PipeRuntimeSinkCriticalException occurred, will throw a
PipeRuntimeSinkCriticalException.");
- super.onFailure(new
PipeRuntimeSinkCriticalException(throwable.getMessage()));
+ super.onFailure(
+ new PipeRuntimeSinkCriticalException(
+ throwable.getMessage() != null ? throwable.getMessage() :
throwable.toString(),
+ throwable));
}
}
@@ -231,7 +234,7 @@ public abstract class PipeAbstractSinkSubtask extends
PipeReportableSubtask {
report(
(EnrichedEvent) lastEvent,
new PipeRuntimeSinkCriticalException(
- throwable.getMessage() + ", root cause: " +
getRootCause(throwable)));
+ throwable.getMessage() + ", root cause: " +
getRootCause(throwable), throwable));
LOGGER.warn(
"{} failed to handshake with the target system after {} times, "
+ "stopping current subtask {} (creation time: {}, simple class:
{}). "
@@ -346,7 +349,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(
