This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12325-395e207dcad48d86d471ba0d322cbde96dfef8cf in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 5127a0a108118be70a1638ec8bd118322986c7da Author: corgy-w <[email protected]> AuthorDate: Fri Oct 9 09:25:36 2026 +0000 [Fix][Connector-V2] Stop the Doris reader polling forever when the async fetch fails (#12325) --- .../doris/source/reader/DorisValueReader.java | 100 +++++++++++++-------- .../split/DorisSourceSplitEnumeratorTest.java | 87 ++++++++++++++++++ 2 files changed, 152 insertions(+), 35 deletions(-) diff --git a/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/source/reader/DorisValueReader.java b/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/source/reader/DorisValueReader.java index 6ec8d47c5c..ee369d94ad 100644 --- a/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/source/reader/DorisValueReader.java +++ b/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/source/reader/DorisValueReader.java @@ -144,41 +144,52 @@ public class DorisValueReader { return params; } - protected Thread asyncThread = - new Thread( - new Runnable() { - @Override - public void run() { - clientLock.lock(); - try { - TScanNextBatchParams nextBatchParams = new TScanNextBatchParams(); - nextBatchParams.setContextId(contextId); - while (!eos.get()) { - nextBatchParams.setOffset(offset); - TScanBatchResult nextResult = client.getNext(nextBatchParams); - eos.set(nextResult.isEos()); - if (!eos.get()) { - ArrowToSeatunnelRowReader rowBatch = - new ArrowToSeatunnelRowReader( - nextResult.getRows(), - seaTunnelRowType) - .readArrow(); - offset += rowBatch.getReadRowCount(); - rowBatch.close(); - try { - rowBatchBlockingQueue.put(rowBatch); - } catch (InterruptedException e) { - throw new DorisConnectorException( - DorisConnectorErrorCode.ROW_BATCH_GET_FAILED, - e); - } - } - } - } finally { - clientLock.unlock(); - } - } - }); + /** + * Failure of {@link #asyncThread}, if any. The fetch thread cannot throw to the reader, so it + * records the failure here and {@link #hasNext()} reports it instead of returning an end of + * stream. + */ + protected volatile Throwable asyncFailure; + + protected Thread asyncThread = new Thread(this::asyncFetchBatches); + + /** + * Fetches row batches from the backend until the scan is exhausted. Runs on {@link + * #asyncThread}. + * + * <p>{@code eos} is set even when the fetch fails: the consumer waits for either a batch or + * {@code eos}, so a fetch thread that died while {@code eos} was still false would leave {@link + * #hasNext()} polling forever instead of failing the job. + */ + private void asyncFetchBatches() { + clientLock.lock(); + try { + TScanNextBatchParams nextBatchParams = new TScanNextBatchParams(); + nextBatchParams.setContextId(contextId); + while (!eos.get()) { + nextBatchParams.setOffset(offset); + TScanBatchResult nextResult = client.getNext(nextBatchParams); + eos.set(nextResult.isEos()); + if (!eos.get()) { + ArrowToSeatunnelRowReader rowBatch = + new ArrowToSeatunnelRowReader(nextResult.getRows(), seaTunnelRowType) + .readArrow(); + offset += rowBatch.getReadRowCount(); + rowBatch.close(); + rowBatchBlockingQueue.put(rowBatch); + } + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + asyncFailure = + new DorisConnectorException(DorisConnectorErrorCode.ROW_BATCH_GET_FAILED, e); + } catch (Throwable t) { + asyncFailure = t; + } finally { + eos.set(true); + clientLock.unlock(); + } + } protected boolean asyncThreadStarted() { boolean started = false; @@ -204,6 +215,7 @@ public class DorisValueReader { try { rowBatch = rowBatchBlockingQueue.take(); } catch (InterruptedException e) { + Thread.currentThread().interrupt(); throw new DorisConnectorException( DorisConnectorErrorCode.ROW_BATCH_GET_FAILED, e); } @@ -214,9 +226,20 @@ public class DorisValueReader { try { Thread.sleep(5); } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new DorisConnectorException( + DorisConnectorErrorCode.ROW_BATCH_GET_FAILED, e); } } } + // The queue is drained and the fetch thread has finished. If it finished with a + // failure, report it instead of reporting that the scan reached its end. + if (asyncFailure != null) { + throw asyncFailure instanceof DorisConnectorException + ? (DorisConnectorException) asyncFailure + : new DorisConnectorException( + DorisConnectorErrorCode.ROW_BATCH_GET_FAILED, asyncFailure); + } } else { hasNext = true; } @@ -266,6 +289,13 @@ public class DorisValueReader { } public void close() { + // Stop the asynchronous fetch before taking the client lock: the fetch thread holds that + // lock for the whole scan, so a reader that is still waiting for its first batch would + // otherwise block close() until the scan finishes on its own. + eos.set(true); + if (asyncThreadStarted) { + asyncThread.interrupt(); + } clientLock.lock(); try { TScanCloseParams closeParams = new TScanCloseParams(); diff --git a/seatunnel-connectors-v2/connector-doris/src/test/java/org/apache/seatunnel/connectors/doris/split/DorisSourceSplitEnumeratorTest.java b/seatunnel-connectors-v2/connector-doris/src/test/java/org/apache/seatunnel/connectors/doris/split/DorisSourceSplitEnumeratorTest.java index 9a2709d465..39478dd009 100644 --- a/seatunnel-connectors-v2/connector-doris/src/test/java/org/apache/seatunnel/connectors/doris/split/DorisSourceSplitEnumeratorTest.java +++ b/seatunnel-connectors-v2/connector-doris/src/test/java/org/apache/seatunnel/connectors/doris/split/DorisSourceSplitEnumeratorTest.java @@ -21,10 +21,18 @@ import org.apache.seatunnel.shade.com.google.common.collect.Maps; import org.apache.seatunnel.api.source.SourceSplitEnumerator; import org.apache.seatunnel.api.table.catalog.TablePath; +import org.apache.seatunnel.api.table.type.BasicType; +import org.apache.seatunnel.api.table.type.SeaTunnelDataType; +import org.apache.seatunnel.api.table.type.SeaTunnelRowType; +import org.apache.seatunnel.common.utils.ReflectionUtils; +import org.apache.seatunnel.connectors.doris.backend.BackendClient; import org.apache.seatunnel.connectors.doris.config.DorisSourceConfig; +import org.apache.seatunnel.connectors.doris.exception.DorisConnectorErrorCode; +import org.apache.seatunnel.connectors.doris.exception.DorisConnectorException; import org.apache.seatunnel.connectors.doris.rest.PartitionDefinition; import org.apache.seatunnel.connectors.doris.rest.RestService; import org.apache.seatunnel.connectors.doris.source.DorisSourceTable; +import org.apache.seatunnel.connectors.doris.source.reader.DorisValueReader; import org.apache.seatunnel.connectors.doris.source.split.DorisSourceSplit; import org.apache.seatunnel.connectors.doris.source.split.DorisSourceSplitEnumerator; @@ -36,10 +44,14 @@ import org.mockito.Mockito; import lombok.extern.slf4j.Slf4j; +import java.time.Duration; import java.util.ArrayList; import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.locks.ReentrantLock; import java.util.stream.Collectors; import java.util.stream.IntStream; @@ -141,4 +153,79 @@ public class DorisSourceSplitEnumeratorTest { return filesPerIteration; } } + + /** + * The asynchronous fetch thread cannot throw to the reader, so it records the failure and + * {@code hasNext()} must report it instead of returning an end of stream. Before the fix the + * reader polled forever in this case. + */ + @Test + public void dorisValueReaderReportsAsyncFetchFailure() { + BackendClient client = Mockito.mock(BackendClient.class); + Mockito.when(client.getNext(Mockito.any())) + .thenThrow( + new DorisConnectorException( + DorisConnectorErrorCode.BACKEND_CLIENT_FAILED, + "simulated backend failure")); + DorisValueReader reader = newAsyncValueReader(client); + + // The asynchronous fetch fails while reading the first batch. + ReflectionUtils.invoke(reader, "asyncFetchBatches"); + + DorisConnectorException exception = + Assertions.assertThrows( + DorisConnectorException.class, + () -> + Assertions.assertTimeoutPreemptively( + Duration.ofSeconds(10), reader::hasNext)); + Assertions.assertTrue( + String.valueOf(exception.getMessage()).contains("simulated backend failure"), + "the backend failure should be reported: " + exception); + } + + /** + * {@code close()} must stop the asynchronous fetch, otherwise a reader waiting for its first + * batch keeps polling and {@code hasNext()} never returns. + */ + @Test + public void dorisValueReaderCloseStopsAsyncFetch() { + BackendClient client = Mockito.mock(BackendClient.class); + DorisValueReader reader = newAsyncValueReader(client); + Thread asyncThread = Mockito.mock(Thread.class); + ReflectionUtils.setField(reader, DorisValueReader.class, "asyncThread", asyncThread); + + reader.close(); + + Mockito.verify(asyncThread).interrupt(); + Mockito.verify(client).closeScanner(Mockito.any()); + Assertions.assertFalse( + Assertions.assertTimeoutPreemptively(Duration.ofSeconds(10), reader::hasNext)); + } + + /** + * Builds a reader with the asynchronous read path enabled and a mocked backend, without running + * the constructor, which opens a real scan on a Doris backend. + */ + private DorisValueReader newAsyncValueReader(BackendClient client) { + DorisValueReader reader = Mockito.mock(DorisValueReader.class, Mockito.CALLS_REAL_METHODS); + ReflectionUtils.setField(reader, DorisValueReader.class, "client", client); + ReflectionUtils.setField(reader, DorisValueReader.class, "clientLock", new ReentrantLock()); + ReflectionUtils.setField(reader, DorisValueReader.class, "eos", new AtomicBoolean(false)); + ReflectionUtils.setField( + reader, + DorisValueReader.class, + "rowBatchBlockingQueue", + new ArrayBlockingQueue<>(2)); + ReflectionUtils.setField( + reader, + DorisValueReader.class, + "seaTunnelRowType", + new SeaTunnelRowType( + new String[] {"id"}, new SeaTunnelDataType<?>[] {BasicType.LONG_TYPE})); + ReflectionUtils.setField( + reader, DorisValueReader.class, "deserializeArrowToRowBatchAsync", true); + ReflectionUtils.setField(reader, DorisValueReader.class, "asyncThreadStarted", true); + ReflectionUtils.setField(reader, DorisValueReader.class, "contextId", "test-context"); + return reader; + } }
