This is an automated email from the ASF dual-hosted git repository.
hansva pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hop.git
The following commit(s) were added to refs/heads/main by this push:
new 958c4f48e1 Issue #8667 : Cancel Table Input and Database Join queries
when a pipeline stops (#8846)
958c4f48e1 is described below
commit 958c4f48e195f7f1e5df26673e557a1d0dfec63d
Author: Matt Casters <[email protected]>
AuthorDate: Thu Oct 15 16:53:08 2026 +0200
Issue #8667 : Cancel Table Input and Database Join queries when a pipeline
stops (#8846)
* Issue #8667 : Cancel Table Input and Database Join queries when a
pipeline stops
* Issue #8667 : Cancel a running query without deadlocking and without
reporting a failure
Database Join stopRunning() cancelled the statement while holding dbLock,
the
same lock lookupValues holds across the query and putRow. stopAll holds the
transform list while it calls stopRunning, and a stopped putRow calls
stopAll,
so the two threads deadlocked.
Cancel the statement without dbLock. A query that fails because the pipeline
is already stopped is a clean stop in both Database Join and Table Input: no
error is logged and the error count stays at 0.
Co-authored-by: Matt Casters <[email protected]>
---------
Co-authored-by: Cursor Agent <[email protected]>
---
.../transforms/databasejoin/DatabaseJoin.java | 44 +++--
.../transforms/databasejoin/DatabaseJoinTest.java | 179 ++++++++++++++++--
.../pipeline/transforms/tableinput/TableInput.java | 35 +++-
.../transforms/tableinput/TableInputTest.java | 210 ++++++++++++++++++++-
4 files changed, 435 insertions(+), 33 deletions(-)
diff --git
a/plugins/transforms/databasejoin/src/main/java/org/apache/hop/pipeline/transforms/databasejoin/DatabaseJoin.java
b/plugins/transforms/databasejoin/src/main/java/org/apache/hop/pipeline/transforms/databasejoin/DatabaseJoin.java
index 7db91d2ac7..8af6a0e870 100644
---
a/plugins/transforms/databasejoin/src/main/java/org/apache/hop/pipeline/transforms/databasejoin/DatabaseJoin.java
+++
b/plugins/transforms/databasejoin/src/main/java/org/apache/hop/pipeline/transforms/databasejoin/DatabaseJoin.java
@@ -17,6 +17,7 @@
package org.apache.hop.pipeline.transforms.databasejoin;
+import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.ArrayList;
import java.util.List;
@@ -244,6 +245,12 @@ public class DatabaseJoin extends
BaseTransform<DatabaseJoinMeta, DatabaseJoinDa
logBasic(BaseMessages.getString(PKG, "DatabaseJoin.Log.LineNumber") +
getLinesRead());
}
} catch (HopException e) {
+ if (isStopped()) {
+ // The lookup was canceled because the pipeline stopped. The driver
reports that as a
+ // failed query, but this is a clean stop: no error, and no further
rows.
+ logDebug(e.getMessage());
+ return false;
+ }
if (getTransformMeta().isDoingErrorHandling()) {
sendToErrorRow = true;
@@ -269,28 +276,35 @@ public class DatabaseJoin extends
BaseTransform<DatabaseJoinMeta, DatabaseJoinDa
}
/**
- * Stop the running query In the Database Join transform data.isCancelled is
checked before
- * synchronization and set after synchronization is completed.
+ * Cancel the running prepared statement.
+ *
+ * <p>{@code Pipeline.stopTransform} marks this transform stopped before it
calls this method, so
+ * the stopped flag must not skip the cancel. {@code data.isCanceled}
prevents a second cancel.
+ * Cancel is skipped once the transform is disposed, because dispose has
already closed the
+ * statement and dropped the connection.
*
- * <p>To cancel a prepared statement we need a valid database connection
which we do not have if
- * disposed has already been called
+ * <p>Do not hold {@code dbLock} across {@link
Database#cancelStatement(java.sql.Statement)}.
+ * {@code lookupValues} keeps that lock for the whole lookup, including
{@code putRow}. A stopped
+ * {@code putRow} calls {@code stopAll}, and {@code stopAll} holds the
pipeline transform list
+ * while it calls this method. Waiting for {@code dbLock} here deadlocks
those two threads. JDBC
+ * {@code Statement.cancel()} is safe to call from this stopping thread
while the lookup thread is
+ * blocked in the driver.
*/
@Override
public void stopRunning() throws HopException {
- if (this.isStopped() || data.isDisposed()) {
+ if (data.isDisposed()) {
return;
}
- dbLock.lock();
-
- try {
- if (data.db != null && data.db.getConnection() != null &&
!data.isCanceled) {
- data.db.cancelStatement(data.pstmt);
- setStopped(true);
- data.isCanceled = true;
- }
- } finally {
- dbLock.unlock();
+ // Read the connection and statement without dbLock, then cancel outside
any lock the lookup
+ // thread holds. dispose() may clear these fields; a cancel of a statement
that is already
+ // closing is harmless compared with deadlocking the pipeline.
+ Database database = data.db;
+ PreparedStatement statement = data.pstmt;
+ if (database != null && database.getConnection() != null &&
!data.isCanceled) {
+ database.cancelStatement(statement);
+ setStopped(true);
+ data.isCanceled = true;
}
}
diff --git
a/plugins/transforms/databasejoin/src/test/java/org/apache/hop/pipeline/transforms/databasejoin/DatabaseJoinTest.java
b/plugins/transforms/databasejoin/src/test/java/org/apache/hop/pipeline/transforms/databasejoin/DatabaseJoinTest.java
index 3566da85ac..f0c59658d8 100644
---
a/plugins/transforms/databasejoin/src/test/java/org/apache/hop/pipeline/transforms/databasejoin/DatabaseJoinTest.java
+++
b/plugins/transforms/databasejoin/src/test/java/org/apache/hop/pipeline/transforms/databasejoin/DatabaseJoinTest.java
@@ -17,12 +17,17 @@
package org.apache.hop.pipeline.transforms.databasejoin;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
@@ -31,8 +36,16 @@ import static org.mockito.Mockito.withSettings;
import java.sql.Connection;
import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.hop.core.database.Database;
+import org.apache.hop.core.exception.HopDatabaseException;
import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.core.row.RowMeta;
import org.apache.hop.pipeline.Pipeline;
import org.apache.hop.pipeline.PipelineMeta;
import org.apache.hop.pipeline.engines.local.LocalPipelineEngine;
@@ -83,38 +96,41 @@ class DatabaseJoinTest {
mockPipeline));
}
+ /**
+ * Pipeline.stopTransform marks the transform stopped before stopRunning().
That flag must not
+ * skip the cancel.
+ */
@Test
- void testStopRunningWhenTransformIsStopped() throws HopException {
+ void testStopRunningCancelsWhenTransformIsAlreadyStopped() throws
HopException {
doReturn(true).when(mockDatabaseJoin).isStopped();
+ doReturn(false).when(mockTransformDataInterface).isDisposed();
+
when(mockTransformDataInterface.db.getConnection()).thenReturn(mock(Connection.class));
mockDatabaseJoin.stopRunning();
- verify(mockDatabaseJoin, times(1)).isStopped();
- verify(mockTransformDataInterface, times(0)).isDisposed();
+ verify(mockTransformDataInterface.db,
times(1)).cancelStatement(any(PreparedStatement.class));
+ assertTrue(mockTransformDataInterface.isCanceled);
+ verify(mockDatabaseJoin, times(1)).setStopped(true);
}
@Test
void testStopRunningWhenTransformDataInterfaceIsDisposed() throws
HopException {
- doReturn(false).when(mockDatabaseJoin).isStopped();
doReturn(true).when(mockTransformDataInterface).isDisposed();
mockDatabaseJoin.stopRunning();
- verify(mockDatabaseJoin, times(1)).isStopped();
verify(mockTransformDataInterface, times(1)).isDisposed();
+ verify(mockTransformDataInterface.db,
times(0)).cancelStatement(any(PreparedStatement.class));
+ assertFalse(mockTransformDataInterface.isCanceled);
}
@Test
- void
-
testStopRunningWhenTransformIsNotStoppedNorTransformDataInterfaceIsDisposedAndDatabaseConnectionIsValid()
- throws HopException {
- doReturn(false).when(mockDatabaseJoin).isStopped();
+ void testStopRunningCancelsWhenDatabaseConnectionIsValid() throws
HopException {
doReturn(false).when(mockTransformDataInterface).isDisposed();
when(mockTransformDataInterface.db.getConnection()).thenReturn(mock(Connection.class));
mockDatabaseJoin.stopRunning();
- verify(mockDatabaseJoin, times(1)).isStopped();
verify(mockTransformDataInterface, times(1)).isDisposed();
verify(mockTransformDataInterface.db, times(1)).getConnection();
verify(mockTransformDataInterface.db,
times(1)).cancelStatement(any(PreparedStatement.class));
@@ -122,19 +138,152 @@ class DatabaseJoinTest {
}
@Test
- void
-
testStopRunningWhenTransformIsNotStoppedNorTransformDataInterfaceIsDisposedAndDatabaseConnectionIsNotValid()
- throws HopException {
- doReturn(false).when(mockDatabaseJoin).isStopped();
+ void testStopRunningDoesNotCancelTwice() throws HopException {
+ doReturn(false).when(mockTransformDataInterface).isDisposed();
+
when(mockTransformDataInterface.db.getConnection()).thenReturn(mock(Connection.class));
+
+ mockDatabaseJoin.stopRunning();
+ mockDatabaseJoin.stopRunning();
+
+ verify(mockTransformDataInterface.db,
times(1)).cancelStatement(any(PreparedStatement.class));
+ assertTrue(mockTransformDataInterface.isCanceled);
+ }
+
+ @Test
+ void testStopRunningWhenDatabaseConnectionIsNotValid() throws HopException {
doReturn(false).when(mockTransformDataInterface).isDisposed();
when(mockTransformDataInterface.db.getConnection()).thenReturn(null);
mockDatabaseJoin.stopRunning();
- verify(mockDatabaseJoin, times(1)).isStopped();
verify(mockTransformDataInterface, times(1)).isDisposed();
verify(mockTransformDataInterface.db, times(1)).getConnection();
verify(mockTransformDataInterface.db,
times(0)).cancelStatement(any(PreparedStatement.class));
assertFalse(mockTransformDataInterface.isCanceled);
}
+
+ /**
+ * lookupValues holds {@code dbLock} for the whole lookup, including the
query and {@code putRow}.
+ * stopAll holds the pipeline transform list while it calls stopRunning.
Cancel has to run without
+ * that lock, or the stop never returns.
+ */
+ @Test
+ void stopDuringRunningQueryDoesNotDeadlockAndEndsWithNoErrors() throws
Exception {
+ stubRunningLookup();
+ doReturn(false).when(mockTransformDataInterface).isDisposed();
+
when(mockTransformDataInterface.db.getConnection()).thenReturn(mock(Connection.class));
+
+ CountDownLatch queryStarted = new CountDownLatch(1);
+ CountDownLatch cancelCalled = new CountDownLatch(1);
+ when(mockTransformDataInterface.db.openQuery(
+ any(PreparedStatement.class), any(IRowMeta.class),
any(Object[].class)))
+ .thenAnswer(
+ invocation -> {
+ // Still inside lookupValues, so dbLock is held by this thread.
+ queryStarted.countDown();
+ assertTrue(
+ cancelCalled.await(5, TimeUnit.SECONDS),
+ "stopRunning did not cancel while the lookup held dbLock");
+ throw new HopDatabaseException(
+ "The query was canceled", new SQLException("The query was
canceled"));
+ });
+ doAnswer(
+ invocation -> {
+ cancelCalled.countDown();
+ return null;
+ })
+ .when(mockTransformDataInterface.db)
+ .cancelStatement(any(PreparedStatement.class));
+
+ AtomicReference<Throwable> queryFailure = new AtomicReference<>();
+ AtomicBoolean keepGoing = new AtomicBoolean(true);
+ Thread queryThread =
+ new Thread(
+ () -> {
+ try {
+ keepGoing.set(mockDatabaseJoin.processRow());
+ } catch (Throwable t) {
+ queryFailure.set(t);
+ }
+ },
+ "database-join-lookup");
+ queryThread.setDaemon(true);
+ queryThread.start();
+ assertTrue(queryStarted.await(5, TimeUnit.SECONDS), "lookup did not reach
the query");
+
+ AtomicReference<Throwable> stopFailure = new AtomicReference<>();
+ Thread stopThread =
+ new Thread(
+ () -> {
+ try {
+ // Pipeline.stopTransform sets the flag before stopRunning().
+ mockDatabaseJoin.setStopped(true);
+ mockDatabaseJoin.stopRunning();
+ } catch (Throwable t) {
+ stopFailure.set(t);
+ }
+ },
+ "database-join-stop");
+ stopThread.setDaemon(true);
+ stopThread.start();
+ stopThread.join(3_000);
+
+ assertFalse(stopThread.isAlive(), "stopRunning deadlocked while the lookup
held dbLock");
+ assertNull(stopFailure.get());
+ queryThread.join(3_000);
+ assertFalse(queryThread.isAlive(), "lookup did not finish after the query
was cancelled");
+ assertNull(queryFailure.get());
+ assertFalse(keepGoing.get());
+ assertEquals(0, mockDatabaseJoin.getErrors());
+ verify(mockDatabaseJoin, never()).stopAll();
+ verify(mockDatabaseJoin, never()).logError(anyString(),
any(Throwable.class));
+ verify(mockTransformDataInterface.db,
times(1)).cancelStatement(any(PreparedStatement.class));
+ }
+
+ /** A cancel that surfaces from the running lookup is a clean stop, not a
failed SQL query. */
+ @Test
+ void cancelDuringQueryEndsWithNoErrors() throws Exception {
+ stubRunningLookup();
+ when(mockTransformDataInterface.db.openQuery(
+ any(PreparedStatement.class), any(IRowMeta.class),
any(Object[].class)))
+ .thenThrow(
+ new HopDatabaseException(
+ "The query was canceled", new SQLException("The query was
canceled")));
+ mockDatabaseJoin.setStopped(true);
+
+ assertFalse(mockDatabaseJoin.processRow());
+
+ assertEquals(0, mockDatabaseJoin.getErrors());
+ verify(mockDatabaseJoin, never()).stopAll();
+ verify(mockDatabaseJoin, never()).logError(anyString(),
any(Throwable.class));
+ verify(mockDatabaseJoin).logDebug(anyString());
+ }
+
+ /** A query error while the pipeline is still running is still a failure. */
+ @Test
+ void queryFailureWhileRunningSetsAnError() throws Exception {
+ stubRunningLookup();
+ doNothing().when(mockDatabaseJoin).stopAll();
+ doNothing().when(mockDatabaseJoin).setOutputDone();
+ when(mockTransformDataInterface.db.openQuery(
+ any(PreparedStatement.class), any(IRowMeta.class),
any(Object[].class)))
+ .thenThrow(new HopDatabaseException("syntax error", new
SQLException("bad sql")));
+
+ assertFalse(mockDatabaseJoin.processRow());
+
+ assertEquals(1, mockDatabaseJoin.getErrors());
+ verify(mockDatabaseJoin).stopAll();
+ verify(mockDatabaseJoin).logError(anyString(), any(Throwable.class));
+ }
+
+ /** Skip the first-row metadata setup and block in the lookup query itself.
*/
+ private void stubRunningLookup() throws HopException {
+ mockDatabaseJoin.first = false;
+ mockTransformDataInterface.lookupRowMeta = new RowMeta();
+ mockTransformDataInterface.keynrs = new int[0];
+ doReturn(new Object[] {"in"}).when(mockDatabaseJoin).getRow();
+ doReturn(new RowMeta()).when(mockDatabaseJoin).getInputRowMeta();
+ doNothing().when(mockDatabaseJoin).logDebug(anyString());
+ doNothing().when(mockDatabaseJoin).logError(anyString(),
any(Throwable.class));
+ }
}
diff --git
a/plugins/transforms/tableinput/src/main/java/org/apache/hop/pipeline/transforms/tableinput/TableInput.java
b/plugins/transforms/tableinput/src/main/java/org/apache/hop/pipeline/transforms/tableinput/TableInput.java
index 6320edcbf8..cb10b4e819 100644
---
a/plugins/transforms/tableinput/src/main/java/org/apache/hop/pipeline/transforms/tableinput/TableInput.java
+++
b/plugins/transforms/tableinput/src/main/java/org/apache/hop/pipeline/transforms/tableinput/TableInput.java
@@ -282,6 +282,9 @@ public class TableInput extends
BaseTransform<TableInputMeta, TableInputData> {
data.rs = data.db.openQuery(sql, boundMeta, boundData,
ResultSet.FETCH_FORWARD, false);
}
} catch (HopDatabaseException ex) {
+ if (stoppedDuringQuery(ex)) {
+ return false;
+ }
Throwable root = ex.getCause();
if (root instanceof SQLException) {
logError(
@@ -346,6 +349,9 @@ public class TableInput extends
BaseTransform<TableInputMeta, TableInputData> {
}
}
} catch (HopDatabaseException ex) {
+ if (stoppedDuringQuery(ex)) {
+ return false;
+ }
logError("Error reading rows from query result: " + ex.getMessage());
if (isDebug()) {
logDebug(Const.getStackTracker(ex));
@@ -389,10 +395,35 @@ public class TableInput extends
BaseTransform<TableInputMeta, TableInputData> {
super.dispose();
}
- /** Stop the running query */
+ /**
+ * A cancel while the statement is executing surfaces as a database
exception. The pipeline has
+ * already been stopped, so this is not a failed query.
+ *
+ * @return {@code true} when the exception should end the transform without
an error
+ */
+ private boolean stoppedDuringQuery(HopDatabaseException ex) {
+ if (!isStopped()) {
+ return false;
+ }
+ // Same handling as a cancel caught while reading later rows in
processRow().
+ logDebug(ex.getMessage());
+ return true;
+ }
+
+ /**
+ * Cancel the open query.
+ *
+ * <p>{@code Pipeline.stopTransform} marks this transform stopped before it
calls this method, so
+ * the stopped flag must not skip the cancel. {@code data.isCanceled}
prevents a second cancel. A
+ * disposed transform has already closed the connection.
+ *
+ * <p>{@code doQuery} does not hold this transform's monitor, so the cancel
below is not waiting
+ * on a lock the query thread holds. Do not add one: {@code stopAll} keeps
the pipeline transform
+ * list locked while it calls this method, and a stopped {@code putRow}
calls {@code stopAll}.
+ */
@Override
public synchronized void stopRunning() throws HopException {
- if (this.isStopped() || data.isDisposed()) {
+ if (data.isDisposed()) {
return;
}
diff --git
a/plugins/transforms/tableinput/src/test/java/org/apache/hop/pipeline/transforms/tableinput/TableInputTest.java
b/plugins/transforms/tableinput/src/test/java/org/apache/hop/pipeline/transforms/tableinput/TableInputTest.java
index ec9607c991..ad37476514 100644
---
a/plugins/transforms/tableinput/src/test/java/org/apache/hop/pipeline/transforms/tableinput/TableInputTest.java
+++
b/plugins/transforms/tableinput/src/test/java/org/apache/hop/pipeline/transforms/tableinput/TableInputTest.java
@@ -17,6 +17,10 @@
package org.apache.hop.pipeline.transforms.tableinput;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.isNull;
import static org.mockito.ArgumentMatchers.notNull;
@@ -24,8 +28,10 @@ import static org.mockito.ArgumentMatchers.nullable;
import static org.mockito.Mockito.any;
import static org.mockito.Mockito.anyInt;
import static org.mockito.Mockito.anyString;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
@@ -33,12 +39,19 @@ import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
+import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
+import java.sql.SQLException;
import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.hop.core.HopClientEnvironment;
import org.apache.hop.core.database.Database;
import org.apache.hop.core.database.DatabaseMeta;
+import org.apache.hop.core.exception.HopDatabaseException;
import org.apache.hop.core.row.IRowMeta;
import org.apache.hop.core.row.RowMeta;
import org.apache.hop.core.row.value.ValueMetaString;
@@ -46,6 +59,7 @@ import org.apache.hop.core.variables.Variables;
import org.apache.hop.databases.h2.H2DatabaseMeta;
import org.apache.hop.metadata.serializer.memory.MemoryMetadataProvider;
import org.apache.hop.pipeline.PipelineMeta;
+import org.apache.hop.pipeline.engine.EngineComponent.ComponentExecutionStatus;
import org.apache.hop.pipeline.engines.local.LocalPipelineEngine;
import org.apache.hop.pipeline.transform.TransformMeta;
import org.junit.jupiter.api.BeforeAll;
@@ -173,7 +187,7 @@ class TableInputTest {
* stopped.
*/
private static void assertSucceeded(TableInput tableInput) {
- org.junit.jupiter.api.Assertions.assertEquals(0, tableInput.getErrors());
+ assertEquals(0, tableInput.getErrors());
verify(tableInput, never()).stopAll();
}
@@ -258,4 +272,198 @@ class TableInputTest {
assertQueryWithoutBind(db, "SELECT '?' AS Q FROM T WHERE X = 1");
assertSucceeded(tableInput);
}
+
+ /**
+ * Pipeline.stopTransform marks the transform stopped before stopRunning().
The open statement
+ * must still be cancelled, or drivers such as SQL Server keep streaming the
rest of the result
+ * set while the pipeline stays in Halting.
+ */
+ @Test
+ void stopRunningCancelsQueryWhenTransformIsAlreadyStopped() throws Exception
{
+ doReturn(mock(Connection.class)).when(db).getConnection();
+ tableInput.setStopped(true);
+
+ tableInput.stopRunning();
+
+ verify(db, times(1)).cancelQuery();
+ assertTrue(data.isCanceled);
+ assertTrue(tableInput.isStopped());
+ }
+
+ @Test
+ void stopRunningDoesNotCancelTwice() throws Exception {
+ doReturn(mock(Connection.class)).when(db).getConnection();
+
+ tableInput.stopRunning();
+ tableInput.stopRunning();
+
+ verify(db, times(1)).cancelQuery();
+ assertTrue(data.isCanceled);
+ }
+
+ @Test
+ void stopRunningDoesNothingWhenDisposed() throws Exception {
+ doReturn(mock(Connection.class)).when(db).getConnection();
+ data.setStatus(ComponentExecutionStatus.STATUS_DISPOSED);
+
+ tableInput.stopRunning();
+
+ verify(db, never()).cancelQuery();
+ assertFalse(data.isCanceled);
+ }
+
+ @Test
+ void stopRunningDoesNothingWhenConnectionIsMissing() throws Exception {
+ doReturn(null).when(db).getConnection();
+
+ tableInput.stopRunning();
+
+ verify(db, never()).cancelQuery();
+ assertFalse(data.isCanceled);
+ assertTrue(tableInput.isStopped());
+ }
+
+ /**
+ * doQuery does not hold a lock stopRunning waits on. A stop during
openQuery must cancel and
+ * return, then the canceled query ends the transform with no errors.
+ */
+ @Test
+ void stopDuringRunningQueryDoesNotBlockAndEndsWithNoErrors() throws
Exception {
+ stubHeaderRow();
+ doReturn(mock(Connection.class)).when(db).getConnection();
+ CountDownLatch queryStarted = new CountDownLatch(1);
+ CountDownLatch cancelCalled = new CountDownLatch(1);
+ doAnswer(
+ invocation -> {
+ queryStarted.countDown();
+ assertTrue(cancelCalled.await(5, TimeUnit.SECONDS), "query was
not cancelled");
+ throw new HopDatabaseException(
+ "The query was canceled", new SQLException("The query was
canceled"));
+ })
+ .when(db)
+ .openQuery(
+ anyString(),
+ nullable(IRowMeta.class),
+ nullable(Object[].class),
+ anyInt(),
+ anyBoolean());
+ doAnswer(
+ invocation -> {
+ cancelCalled.countDown();
+ return null;
+ })
+ .when(db)
+ .cancelQuery();
+
+ AtomicReference<Throwable> queryFailure = new AtomicReference<>();
+ AtomicBoolean keepGoing = new AtomicBoolean(true);
+ Thread queryThread =
+ new Thread(
+ () -> {
+ try {
+ keepGoing.set(tableInput.processRow());
+ } catch (Throwable t) {
+ queryFailure.set(t);
+ }
+ },
+ "table-input-query");
+ queryThread.setDaemon(true);
+ queryThread.start();
+ assertTrue(queryStarted.await(5, TimeUnit.SECONDS), "query did not start");
+
+ AtomicReference<Throwable> stopFailure = new AtomicReference<>();
+ Thread stopThread =
+ new Thread(
+ () -> {
+ try {
+ tableInput.setStopped(true);
+ tableInput.stopRunning();
+ } catch (Throwable t) {
+ stopFailure.set(t);
+ }
+ },
+ "table-input-stop");
+ stopThread.setDaemon(true);
+ stopThread.start();
+ stopThread.join(3_000);
+
+ assertFalse(stopThread.isAlive(), "stopRunning blocked while the query was
running");
+ assertNull(stopFailure.get());
+ queryThread.join(3_000);
+ assertFalse(queryThread.isAlive(), "query did not finish after it was
cancelled");
+ assertNull(queryFailure.get());
+ assertFalse(keepGoing.get());
+ assertCleanStop();
+ verify(db).cancelQuery();
+ }
+
+ /**
+ * Stop before the first row: openQuery fails with "query was canceled", and
that is not an error.
+ */
+ @Test
+ void cancelDuringQueryEndsWithNoErrors() throws Exception {
+ stubHeaderRow();
+ tableInput.setStopped(true);
+ doThrow(
+ new HopDatabaseException(
+ "The query was canceled", new SQLException("The query was
canceled")))
+ .when(db)
+ .openQuery(
+ anyString(),
+ nullable(IRowMeta.class),
+ nullable(Object[].class),
+ anyInt(),
+ anyBoolean());
+
+ assertFalse(tableInput.processRow());
+
+ assertCleanStop();
+ verify(tableInput).logDebug(any());
+ }
+
+ /**
+ * Some drivers return from execute and block on the first fetch. Cancel
there is a clean stop.
+ */
+ @Test
+ void cancelWhileReadingFirstRowEndsWithNoErrors() throws Exception {
+ stubHeaderRow();
+ tableInput.setStopped(true);
+ doThrow(
+ new HopDatabaseException(
+ "The query was canceled", new SQLException("The query was
canceled")))
+ .when(db)
+ .getRow(any(ResultSet.class));
+
+ assertFalse(tableInput.processRow());
+
+ assertCleanStop();
+ verify(tableInput, never())
+ .logError(org.mockito.ArgumentMatchers.contains("Error reading rows"));
+ }
+
+ /** A real SQL failure while the pipeline is still running is still
reported. */
+ @Test
+ void queryFailureWhileRunningSetsAnError() throws Exception {
+ stubHeaderRow();
+ doThrow(new HopDatabaseException("syntax error", new SQLException("bad
sql")))
+ .when(db)
+ .openQuery(
+ anyString(),
+ nullable(IRowMeta.class),
+ nullable(Object[].class),
+ anyInt(),
+ anyBoolean());
+
+ assertFalse(tableInput.processRow());
+
+ assertEquals(1, tableInput.getErrors());
+ verify(tableInput).stopAll();
+ verify(tableInput).logError(org.mockito.ArgumentMatchers.contains("SQL
query failed"));
+ }
+
+ private void assertCleanStop() {
+ assertEquals(0, tableInput.getErrors());
+ verify(tableInput, never()).stopAll();
+ verify(tableInput, never()).logError(any());
+ }
}
