diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java index a8cbcab6ac6826..7c9fb6bfbc5ecc 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java +++ b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java @@ -404,7 +404,7 @@ private String validateComputeGroupProperty(Map props) throws An private SourceOffsetProvider createOffsetProvider(Map jdbcSourceProps) { SourceOffsetProvider provider; if (tvfType != null) { - provider = SourceOffsetProviderFactory.createSourceOffsetProvider(tvfType); + provider = SourceOffsetProviderFactory.createSourceOffsetProvider(tvfType, jobProperties); } else { provider = new JdbcSourceOffsetProvider(getJobId(), dataSourceType, jdbcSourceProps); } @@ -555,6 +555,9 @@ public void alterJob(AlterJobCommand alterJobCommand) throws AnalysisException, public void updateJobStatus(JobStatus status) throws JobException { lock.writeLock().lock(); try { + if (isFinalStatus() && !getJobStatus().equals(status)) { + throw new JobException("Can't update final job status " + getJobStatus() + " to " + status); + } super.updateJobStatus(status); if (JobStatus.PAUSED.equals(getJobStatus())) { clearRunningStreamTask(status); @@ -592,6 +595,41 @@ public boolean updateJobStatusIfCurrent(JobStatus expectedStatus, JobStatus newS } } + public boolean tryFinishJob() throws JobException { + lock.writeLock().lock(); + try { + if (!isActive()) { + return false; + } + if (runningStreamTask != null && TaskStatus.PENDING.equals(runningStreamTask.getStatus())) { + // Cancel the waiting task when a metadata scan detects the end of the source. + cancelAllTasks(false); + } + resetFailureInfo(null); + updateJobStatus(JobStatus.FINISHED); + logUpdateOperation(); + return true; + } finally { + lock.writeLock().unlock(); + } + } + + private boolean tryPauseJob(FailureReason reason) throws JobException { + lock.writeLock().lock(); + try { + if (!isActive() + || (getFailureReason() != null + && InternalErrorCode.MANUAL_PAUSE_ERR.equals(getFailureReason().getCode()))) { + return false; + } + updateJobStatus(JobStatus.PAUSED); + setFailureReason(reason); + return true; + } finally { + lock.writeLock().unlock(); + } + } + public void resetFailureInfo(FailureReason reason) { this.setFailureReason(reason); // Currently, only delayMsg is present here, which needs to be cleared when the status changes. @@ -766,17 +804,8 @@ protected void fetchMeta() throws JobException { offsetProvider.fetchRemoteMeta(props); } catch (Exception ex) { log.warn("fetch remote meta failed, job id: {}", getJobId(), ex); - if (this.getFailureReason() == null - || !InternalErrorCode.MANUAL_PAUSE_ERR.equals(this.getFailureReason().getCode())) { - // When a job is manually paused, it does not need to be set again, - // otherwise, it may be woken up by auto resume. - // Pause before setting the reason: updateJobStatus's writeLock orders this after any - // task-success callback that clears failureReason, so a success can't wipe the reason. - this.updateJobStatus(JobStatus.PAUSED); - this.setFailureReason( - new FailureReason(InternalErrorCode.GET_REMOTE_DATA_ERROR, - "Failed to fetch meta, " + ex.getMessage())); - + if (tryPauseJob(new FailureReason(InternalErrorCode.GET_REMOTE_DATA_ERROR, + "Failed to fetch meta, " + ex.getMessage()))) { if (MetricRepo.isInit) { MetricRepo.COUNTER_STREAMING_JOB_GET_META_FAIL_COUNT.increase(1L); } @@ -814,26 +843,24 @@ public void advanceSplitsIfNeed() throws JobException { } } catch (Exception ex) { log.warn("advance splits failed, job id: {}", getJobId(), ex); - if (this.getFailureReason() == null - || !InternalErrorCode.MANUAL_PAUSE_ERR.equals(this.getFailureReason().getCode())) { - this.setFailureReason(new FailureReason( - InternalErrorCode.GET_REMOTE_DATA_ERROR, - "Failed to advance splits, " + ex.getMessage())); - this.updateJobStatus(JobStatus.PAUSED); - } + tryPauseJob(new FailureReason(InternalErrorCode.GET_REMOTE_DATA_ERROR, + "Failed to advance splits, " + ex.getMessage())); } } public boolean needScheduleTask() { readLock(); try { - return (getJobStatus().equals(JobStatus.RUNNING) - || getJobStatus().equals(JobStatus.PENDING)); + return isActive(); } finally { readUnlock(); } } + private boolean isActive() { + return JobStatus.PENDING.equals(getJobStatus()) || JobStatus.RUNNING.equals(getJobStatus()); + } + public void clearRunningStreamTask(JobStatus newJobStatus) { if (runningStreamTask != null) { log.info("clear running streaming insert task for job {}, task {}, status {} ", @@ -972,6 +999,7 @@ private void updateCloudJobStatisticAndOffset(StreamingTaskTxnCommitAttachment a this.jobStatistic.setFileSize(attachment.getFileBytes()); this.jobStatistic.setFilteredRows(attachment.getFilteredRows()); offsetProvider.updateOffset(offsetProvider.deserializeOffset(attachment.getOffset())); + this.offsetProviderPersist = offsetProvider.getPersistInfo(); //update metric if (MetricRepo.isInit && !isReplay) { @@ -1482,6 +1510,9 @@ public void replayOnVisible(TransactionState txnState) { @Override public void gsonPostProcess() throws IOException { + if (jobProperties == null && properties != null) { + jobProperties = new StreamingJobProperties(properties); + } if (offsetProvider == null) { offsetProvider = createOffsetProvider(sourceProperties); if (tvfType != null) { @@ -1489,9 +1520,6 @@ public void gsonPostProcess() throws IOException { } } - if (jobProperties == null && properties != null) { - jobProperties = new StreamingJobProperties(properties); - } recomputeDerivedFields(); if (null == getSucceedTaskCount()) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobProperties.java b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobProperties.java index 3db3aecaf8b2ac..498b21fb78e93f 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobProperties.java +++ b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobProperties.java @@ -33,6 +33,7 @@ import java.util.HashMap; import java.util.HashSet; import java.util.List; +import java.util.Locale; import java.util.Map; import java.util.Set; @@ -41,12 +42,16 @@ public class StreamingJobProperties implements JobProperties { public static final String MAX_INTERVAL_SECOND_PROPERTY = "max_interval"; public static final String S3_MAX_BATCH_FILES_PROPERTY = "s3.max_batch_files"; public static final String S3_MAX_BATCH_BYTES_PROPERTY = "s3.max_batch_bytes"; + public static final String S3_INGESTION_MODE_PROPERTY = "s3.ingestion_mode"; + public static final String S3_INGESTION_MODE_LEXICAL = "LEXICAL"; + public static final String S3_INGESTION_MODE_ONCE = "ONCE"; public static final String SESSION_VAR_PREFIX = "session."; public static final String INTERNAL_KEY_PREFIX = "__"; public static final String OFFSET_PROPERTY = "offset"; public static final String COMPUTE_GROUP_PROPERTY = "compute_group"; public static final List SUPPORT_STREAM_JOB_PROPS = Arrays.asList(MAX_INTERVAL_SECOND_PROPERTY, - S3_MAX_BATCH_FILES_PROPERTY, S3_MAX_BATCH_BYTES_PROPERTY, OFFSET_PROPERTY, COMPUTE_GROUP_PROPERTY); + S3_MAX_BATCH_FILES_PROPERTY, S3_MAX_BATCH_BYTES_PROPERTY, S3_INGESTION_MODE_PROPERTY, + OFFSET_PROPERTY, COMPUTE_GROUP_PROPERTY); public static final long DEFAULT_MAX_INTERVAL_SECOND = 10; public static final long DEFAULT_MAX_S3_BATCH_FILES = 256; @@ -119,6 +124,15 @@ public void validate() throws AnalysisException { && v <= (long) (1024 * 1024 * 1024) * 10, StreamingJobProperties.S3_MAX_BATCH_BYTES_PROPERTY + " should between 100MB and 10GB"); + String ingestionMode = getS3IngestionMode(); + if (!S3_INGESTION_MODE_LEXICAL.equals(ingestionMode) + && !S3_INGESTION_MODE_ONCE.equals(ingestionMode)) { + throw new AnalysisException("Unsupported s3.ingestion_mode: " + ingestionMode); + } + if (S3_INGESTION_MODE_ONCE.equals(ingestionMode) && properties.containsKey(OFFSET_PROPERTY)) { + throw new AnalysisException("offset is not supported when s3.ingestion_mode is ONCE"); + } + // validate session variables try { Map sessionVarMap = parseSessionVarMap(); @@ -200,4 +214,13 @@ public String getOffsetProperty() { public String getComputeGroup() { return properties.get(COMPUTE_GROUP_PROPERTY); } + + public String getS3IngestionMode() { + return properties.getOrDefault(S3_INGESTION_MODE_PROPERTY, S3_INGESTION_MODE_LEXICAL) + .trim().toUpperCase(Locale.ROOT); + } + + public boolean isS3OnceMode() { + return S3_INGESTION_MODE_ONCE.equals(getS3IngestionMode()); + } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobSchedulerTask.java b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobSchedulerTask.java index c5dfacaadfa2d1..69619eb91103b8 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobSchedulerTask.java +++ b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobSchedulerTask.java @@ -74,9 +74,7 @@ private void handlePendingState() throws JobException { } if (streamingInsertJob.hasReachedEnd()) { // Source already fully consumed (e.g. snapshot-only mode recovered after FE restart). - // Transition directly to FINISHED without creating a new task. - streamingInsertJob.updateJobStatus(JobStatus.FINISHED); - streamingInsertJob.logUpdateOperation(); + streamingInsertJob.tryFinishJob(); return; } streamingInsertJob.createStreamingTask(); diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/offset/SourceOffsetProviderFactory.java b/fe/fe-core/src/main/java/org/apache/doris/job/offset/SourceOffsetProviderFactory.java index 30f9d0edd579eb..631162d27fec31 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/job/offset/SourceOffsetProviderFactory.java +++ b/fe/fe-core/src/main/java/org/apache/doris/job/offset/SourceOffsetProviderFactory.java @@ -18,6 +18,7 @@ package org.apache.doris.job.offset; import org.apache.doris.job.exception.JobException; +import org.apache.doris.job.extensions.insert.streaming.StreamingJobProperties; import org.apache.doris.job.offset.jdbc.JdbcTvfSourceOffsetProvider; import org.apache.doris.job.offset.s3.S3SourceOffsetProvider; @@ -35,8 +36,12 @@ public class SourceOffsetProviderFactory { map.put("cdc_stream", JdbcTvfSourceOffsetProvider.class); } - public static SourceOffsetProvider createSourceOffsetProvider(String sourceType) { + public static SourceOffsetProvider createSourceOffsetProvider( + String sourceType, StreamingJobProperties jobProperties) { try { + if ("s3".equalsIgnoreCase(sourceType) && jobProperties.isS3OnceMode()) { + return new S3SourceOffsetProvider(jobProperties); + } Class cla = map.get(sourceType.toLowerCase()); if (cla == null) { throw new JobException("Unsupported source type: " + sourceType); diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3Offset.java b/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3Offset.java index ebd3380349029e..44df1756e0292b 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3Offset.java +++ b/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3Offset.java @@ -34,6 +34,8 @@ public class S3Offset implements Offset { // s3://bucket/path/{1.csv,2.csv} String fileLists; int fileNum; + @SerializedName("lastBatch") + boolean lastBatch; @Override public String toSerializedJson() { diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3SourceOffsetProvider.java b/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3SourceOffsetProvider.java index 2777013471cda7..f58287ee9a57dd 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3SourceOffsetProvider.java +++ b/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3SourceOffsetProvider.java @@ -49,8 +49,18 @@ @Log4j2 public class S3SourceOffsetProvider implements SourceOffsetProvider { - S3Offset currentOffset; - String maxEndFile; + private final boolean onceMode; + private volatile S3Offset noMoreFilesAfterOffset; + volatile S3Offset currentOffset; + volatile String maxEndFile; + + public S3SourceOffsetProvider() { + this.onceMode = false; + } + + public S3SourceOffsetProvider(StreamingJobProperties jobProperties) { + this.onceMode = jobProperties.isS3OnceMode(); + } @Override public String getSourceType() { @@ -95,6 +105,7 @@ public S3Offset getNextOffset(StreamingJobProperties jobProps, Map properties) throws Exception { Map copiedProps = Maps.newTreeMap(String.CASE_INSENSITIVE_ORDER); copiedProps.putAll(properties); StorageAdapter storageAdapter = StorageAdapter.of(copiedProps); - String startFile = currentOffset == null ? null : currentOffset.endFile; + S3Offset offsetAtScan = currentOffset; + String startFile = offsetAtScan == null ? null : offsetAtScan.endFile; try (FileSystem fileSystem = FileSystemFactory.getFileSystem(storageAdapter)) { String uri = storageAdapter.validateAndGetUri(copiedProps); String filePath = storageAdapter.validateAndNormalizeUri(uri); @@ -168,6 +180,9 @@ public void fetchRemoteMeta(Map properties) throws Exception { throw new java.io.IOException("debug point: simulated S3 auth error"); } GlobListing globListing = fileSystem.globListWithLimit(Location.of(filePath), startFile, 1, 1); + if (onceMode) { + noMoreFilesAfterOffset = globListing.getFiles().isEmpty() ? offsetAtScan : null; + } if (!globListing.getFiles().isEmpty() && StringUtils.isNotEmpty(globListing.getMaxFile())) { maxEndFile = globListing.getMaxFile(); } @@ -176,6 +191,9 @@ public void fetchRemoteMeta(Map properties) throws Exception { @Override public boolean hasMoreDataToConsume() { + if (hasReachedEnd()) { + return false; + } if (currentOffset == null || currentOffset.endFile == null) { return true; } @@ -186,6 +204,12 @@ public boolean hasMoreDataToConsume() { return false; } + @Override + public boolean hasReachedEnd() { + S3Offset offset = currentOffset; + return onceMode && offset != null && (offset.isLastBatch() || noMoreFilesAfterOffset == offset); + } + @Override public String getPersistInfo() { if (currentOffset == null) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/scheduler/StreamingTaskScheduler.java b/fe/fe-core/src/main/java/org/apache/doris/job/scheduler/StreamingTaskScheduler.java index 91d5fb1a658737..62d324061b0bdb 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/job/scheduler/StreamingTaskScheduler.java +++ b/fe/fe-core/src/main/java/org/apache/doris/job/scheduler/StreamingTaskScheduler.java @@ -104,7 +104,7 @@ private void scheduleTasks(List tasks) { } } - private void scheduleOneTask(AbstractStreamingTask task) { + private void scheduleOneTask(AbstractStreamingTask task) throws JobException { if (DebugPointUtil.isEnable("StreamingJob.scheduleTask.exception")) { throw new RuntimeException("debug point StreamingJob.scheduleTask.exception"); } @@ -122,6 +122,10 @@ private void scheduleOneTask(AbstractStreamingTask task) { } // reject task if no more data to consume if (!job.hasMoreDataToConsume()) { + if (job.hasReachedEnd()) { + job.tryFinishJob(); + return; + } String delayMsg = "No data available for consumption at the moment, will retry after " + (System.currentTimeMillis() + DELAY_SCHEDULER_MS); job.setJobRuntimeMsg(delayMsg); diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterJobCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterJobCommand.java index d75be51c457670..cbd2ed430f387d 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterJobCommand.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterJobCommand.java @@ -41,6 +41,7 @@ import com.google.common.base.Preconditions; import org.apache.commons.lang3.StringUtils; +import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Objects; @@ -277,11 +278,19 @@ private void checkUnmodifiableSourceProperties(Map originSourceP } private void validateProps(StreamingInsertJob streamingJob) throws AnalysisException { - StreamingJobProperties jobProperties = new StreamingJobProperties(properties); - jobProperties.validate(); - if (jobProperties.getOffsetProperty() != null) { - streamingJob.validateAlterOffset(jobProperties.getOffsetProperty()); - streamingJob.validateOffset(jobProperties.getOffsetProperty()); + StreamingJobProperties originJobProperties = + new StreamingJobProperties(streamingJob.getProperties()); + Map mergedProperties = new HashMap<>(streamingJob.getProperties()); + mergedProperties.putAll(properties); + StreamingJobProperties updatedJobProperties = new StreamingJobProperties(mergedProperties); + updatedJobProperties.validate(); + if (!originJobProperties.getS3IngestionMode().equals(updatedJobProperties.getS3IngestionMode())) { + throw new AnalysisException("s3.ingestion_mode cannot be altered"); + } + String offset = properties.get(StreamingJobProperties.OFFSET_PROPERTY); + if (offset != null) { + streamingJob.validateAlterOffset(offset); + streamingJob.validateOffset(offset); } } diff --git a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobOffsetPersistenceTest.java b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobOffsetPersistenceTest.java index e3246f1b53074e..1a724b2330a684 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobOffsetPersistenceTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobOffsetPersistenceTest.java @@ -27,7 +27,11 @@ import org.apache.doris.job.exception.JobException; import org.apache.doris.job.manager.JobManager; import org.apache.doris.job.manager.StreamingTaskManager; +import org.apache.doris.job.offset.SourceOffsetProvider; import org.apache.doris.job.offset.jdbc.JdbcSourceOffsetProvider; +import org.apache.doris.job.offset.s3.S3Offset; +import org.apache.doris.job.offset.s3.S3SourceOffsetProvider; +import org.apache.doris.persist.gson.GsonUtils; import org.apache.doris.transaction.GlobalTransactionMgrIface; import org.apache.doris.transaction.TxnStateCallbackFactory; @@ -38,6 +42,7 @@ import java.util.Collections; import java.util.HashMap; +import java.util.Map; import java.util.concurrent.locks.ReentrantReadWriteLock; public class StreamingInsertJobOffsetPersistenceTest { @@ -163,6 +168,103 @@ public void testNaturalFinishPersistsFinalState() throws Exception { } } + @Test + public void testS3OnceLastBatchFinishesOnlyAfterSuccess() throws Exception { + StreamingJobProperties properties = new StreamingJobProperties(Map.of("s3.ingestion_mode", "ONCE")); + TestStreamingInsertJob failedJob = newJob(new S3SourceOffsetProvider(properties), 1017L); + NoopStreamingMultiTblTask failedTask = + (NoopStreamingMultiTblTask) Deencapsulation.getField(failedJob, "runningStreamTask"); + S3Offset failedOffset = new S3Offset(); + failedOffset.setLastBatch(true); + Deencapsulation.setField(failedTask, "runningOffset", failedOffset); + failedTask.setErrMsg("failed"); + + TestStreamingInsertJob succeededJob = newJob(new S3SourceOffsetProvider(properties), 1018L); + NoopStreamingMultiTblTask succeededTask = + (NoopStreamingMultiTblTask) Deencapsulation.getField(succeededJob, "runningStreamTask"); + S3Offset succeededOffset = new S3Offset(); + succeededOffset.setEndFile("data/b.csv"); + succeededOffset.setLastBatch(true); + Deencapsulation.setField(succeededTask, "runningOffset", succeededOffset); + + try (MockedStatic envMockedStatic = Mockito.mockStatic(Env.class)) { + Env env = Mockito.mock(Env.class); + JobManager jobManager = Mockito.mock(JobManager.class); + StreamingTaskManager streamingTaskManager = Mockito.mock(StreamingTaskManager.class); + GlobalTransactionMgrIface transactionMgr = Mockito.mock(GlobalTransactionMgrIface.class); + TxnStateCallbackFactory callbackFactory = Mockito.mock(TxnStateCallbackFactory.class); + envMockedStatic.when(Env::getCurrentEnv).thenReturn(env); + envMockedStatic.when(Env::getCurrentGlobalTransactionMgr).thenReturn(transactionMgr); + Mockito.when(env.getJobManager()).thenReturn(jobManager); + Mockito.when(jobManager.getStreamingTaskManager()).thenReturn(streamingTaskManager); + Mockito.when(transactionMgr.getCallbackFactory()).thenReturn(callbackFactory); + + Assertions.assertEquals(JobStatus.RUNNING, failedJob.getJobStatus()); + Assertions.assertFalse(failedJob.hasReachedEnd()); + failedJob.onStreamTaskFail(failedTask); + Assertions.assertEquals(JobStatus.PAUSED, failedJob.getJobStatus()); + Assertions.assertEquals(0, failedJob.journalCount); + + Assertions.assertEquals(JobStatus.RUNNING, succeededJob.getJobStatus()); + Assertions.assertFalse(succeededJob.hasReachedEnd()); + Deencapsulation.invoke(succeededJob, "updateJobStatisticAndOffset", + new StreamingTaskTxnCommitAttachment(9001L, 1018L, 0, 0, 0, 0, 0, + succeededOffset.toSerializedJson()), false); + succeededJob.onStreamTaskSuccess(succeededTask); + Assertions.assertEquals(JobStatus.FINISHED, succeededJob.getJobStatus()); + Assertions.assertEquals(1, succeededJob.journalCount); + } + } + + @Test + public void testRecoveredSourceFinishesBeforeCreatingTask() throws Exception { + TestStreamingInsertJob job = newJob(new EndJdbcSourceOffsetProvider(), 1019L); + job.setJobStatus(JobStatus.PENDING); + Deencapsulation.setField(job, "runningStreamTask", null); + job.setJobRuntimeMsg("will retry"); + + try (MockedStatic envMockedStatic = Mockito.mockStatic(Env.class)) { + GlobalTransactionMgrIface transactionMgr = Mockito.mock(GlobalTransactionMgrIface.class); + TxnStateCallbackFactory callbackFactory = Mockito.mock(TxnStateCallbackFactory.class); + envMockedStatic.when(Env::getCurrentGlobalTransactionMgr).thenReturn(transactionMgr); + Mockito.when(transactionMgr.getCallbackFactory()).thenReturn(callbackFactory); + + Assertions.assertTrue(job.tryFinishJob()); + } + + Assertions.assertEquals(JobStatus.FINISHED, job.getJobStatus()); + Assertions.assertNull(job.getRunningStreamTask()); + Assertions.assertEquals("", job.getJobRuntimeMsg()); + Assertions.assertEquals(1, job.journalCount); + } + + @Test + public void testCloudReplayRefreshesPersistedOffset() { + Map properties = Map.of("s3.ingestion_mode", "ONCE"); + S3SourceOffsetProvider provider = new S3SourceOffsetProvider(new StreamingJobProperties(properties)); + StreamingInsertJob job = new StreamingInsertJob(); + job.offsetProvider = provider; + job.setJobStatus(JobStatus.PENDING); + Deencapsulation.setField(job, "properties", properties); + Deencapsulation.setField(job, "tvfType", "s3"); + job.setOffsetProviderPersist("{\"endFile\":\"data/a.csv\"}"); + S3Offset offset = new S3Offset(); + offset.setEndFile("data/b.csv"); + offset.setLastBatch(true); + StreamingTaskTxnCommitAttachment attachment = new StreamingTaskTxnCommitAttachment( + 9001L, 1020L, 0, 0, 0, 0, 0, offset.toSerializedJson()); + + Deencapsulation.invoke(job, "updateCloudJobStatisticAndOffset", attachment, true); + + Assertions.assertEquals(provider.getPersistInfo(), job.getOffsetProviderPersist()); + Assertions.assertTrue(job.getOffsetProviderPersist().contains("data/b.csv")); + StreamingInsertJob recovered = GsonUtils.GSON.fromJson( + GsonUtils.GSON.toJson(job), StreamingInsertJob.class); + Assertions.assertEquals(JobStatus.PENDING, recovered.getJobStatus()); + Assertions.assertTrue(recovered.hasReachedEnd()); + Assertions.assertFalse(recovered.hasMoreDataToConsume()); + } + @Test public void testReplayUpdatedRestoresFinalStateAndRemovesCallback() { TestStreamingInsertJob job = newJob(new JdbcSourceOffsetProvider(), 1013L); @@ -195,7 +297,7 @@ public void testReplayUpdatedRestoresStartTime() { Assertions.assertEquals(1234L, job.getStartTimeMs()); } - private static TestStreamingInsertJob newJob(JdbcSourceOffsetProvider provider, long taskId) { + private static TestStreamingInsertJob newJob(SourceOffsetProvider provider, long taskId) { TestStreamingInsertJob job = new TestStreamingInsertJob(); Deencapsulation.setField(job, "lock", new ReentrantReadWriteLock(true)); Deencapsulation.setField(job, "jobId", 9001L); diff --git a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobStatusTransitionTest.java b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobStatusTransitionTest.java index 3e5c97c61d78ca..23b35983d84f2a 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobStatusTransitionTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobStatusTransitionTest.java @@ -19,6 +19,7 @@ import org.apache.doris.common.jmockit.Deencapsulation; import org.apache.doris.job.common.JobStatus; +import org.apache.doris.job.exception.JobException; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; @@ -63,4 +64,12 @@ public void testStoppedSurvivesTheRunningWrite() throws Exception { Assertions.assertFalse(job.updateJobStatusIfCurrent(JobStatus.PENDING, JobStatus.RUNNING)); Assertions.assertEquals(JobStatus.STOPPED, job.getJobStatus()); } + + @Test + public void testFinishedStatusCannotBeOverwritten() { + StreamingInsertJob job = newJob(JobStatus.FINISHED); + + Assertions.assertThrows(JobException.class, () -> job.updateJobStatus(JobStatus.PAUSED)); + Assertions.assertEquals(JobStatus.FINISHED, job.getJobStatus()); + } } diff --git a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobPropertiesTest.java b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobPropertiesTest.java index 10ae94e1cc06c2..5ecddff884cb1b 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobPropertiesTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobPropertiesTest.java @@ -18,8 +18,11 @@ package org.apache.doris.job.extensions.insert.streaming; import org.apache.doris.common.AnalysisException; +import org.apache.doris.common.jmockit.Deencapsulation; +import org.apache.doris.job.common.JobStatus; import org.apache.doris.job.exception.JobException; import org.apache.doris.job.extensions.insert.InsertTask; +import org.apache.doris.nereids.trees.plans.commands.AlterJobCommand; import org.apache.doris.qe.ConnectContext; import org.apache.doris.qe.SessionVariable; @@ -27,9 +30,49 @@ import org.junit.jupiter.api.Test; import java.util.HashMap; +import java.util.Map; public class StreamingJobPropertiesTest { + @Test + public void testS3OnceModeValidationAndAlter() throws Exception { + StreamingJobProperties properties = new StreamingJobProperties(Map.of("s3.ingestion_mode", " once ")); + properties.validate(); + Assertions.assertTrue(properties.isS3OnceMode()); + Assertions.assertThrows(AnalysisException.class, + () -> new StreamingJobProperties(Map.of("s3.ingestion_mode", "invalid")).validate()); + Assertions.assertThrows(AnalysisException.class, + () -> new StreamingJobProperties( + Map.of("s3.ingestion_mode", "ONCE", "offset", "{\"fileName\":\"a.csv\"}")) + .validate()); + + StreamingInsertJob job = new StreamingInsertJob(); + Deencapsulation.setField(job, "properties", properties.getProperties()); + AlterJobCommand alterBatch = new AlterJobCommand("job", Map.of("s3.max_batch_files", "1"), + null, null, null, Map.of(), Map.of()); + Deencapsulation.invoke(alterBatch, "validateProps", job); + AlterJobCommand alterMode = new AlterJobCommand("job", Map.of("s3.ingestion_mode", "LEXICAL"), + null, null, null, Map.of(), Map.of()); + Assertions.assertThrows(AnalysisException.class, + () -> Deencapsulation.invoke(alterMode, "validateProps", job)); + AlterJobCommand alterOffset = new AlterJobCommand("job", + Map.of("offset", "{\"fileName\":\"a.csv\"}"), null, null, null, Map.of(), Map.of()); + Assertions.assertThrows(AnalysisException.class, + () -> Deencapsulation.invoke(alterOffset, "validateProps", job)); + } + + @Test + public void testS3OnceModeRestoredWithLegacyOffset() throws Exception { + StreamingInsertJob job = new StreamingInsertJob(); + Deencapsulation.setField(job, "properties", Map.of("s3.ingestion_mode", "ONCE")); + Deencapsulation.setField(job, "tvfType", "s3"); + job.setOffsetProviderPersist("{\"endFile\":\"data/a.csv\"}"); + job.gsonPostProcess(); + job.setJobStatus(JobStatus.RUNNING); + Assertions.assertFalse(job.hasMoreDataToConsume()); + Assertions.assertFalse(job.hasReachedEnd()); + } + /** * Simulate FE restart: constructor is called without validate(). * Before the fix, maxIntervalSecond would be 0 when properties is non-empty, diff --git a/fe/fe-core/src/test/java/org/apache/doris/job/offset/s3/S3SourceOffsetProviderTest.java b/fe/fe-core/src/test/java/org/apache/doris/job/offset/s3/S3SourceOffsetProviderTest.java new file mode 100644 index 00000000000000..4f72cfd909eb7b --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/job/offset/s3/S3SourceOffsetProviderTest.java @@ -0,0 +1,147 @@ +// 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.doris.job.offset.s3; + +import org.apache.doris.datasource.storage.StorageAdapter; +import org.apache.doris.filesystem.FileEntry; +import org.apache.doris.filesystem.FileSystem; +import org.apache.doris.filesystem.GlobListing; +import org.apache.doris.filesystem.Location; +import org.apache.doris.fs.FileSystemFactory; +import org.apache.doris.job.extensions.insert.streaming.StreamingJobProperties; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.mockito.MockedStatic; +import org.mockito.Mockito; + +import java.io.IOException; +import java.util.Collections; +import java.util.List; +import java.util.Map; + +public class S3SourceOffsetProviderTest { + private static final Map TVF_PROPS = Map.of( + "uri", "s3://bucket/data/*.csv", "s3.endpoint", "s3.us-east-1.amazonaws.com", + "s3.region", "us-east-1", "s3.access_key", "ak", "s3.secret_key", "sk"); + private static final StreamingJobProperties ONCE_PROPS = new StreamingJobProperties( + Map.of("s3.ingestion_mode", "ONCE", "s3.max_batch_files", "1")); + + @Test + public void testOnceMarksLastBatchAndRecovers() throws Exception { + FileSystem fs = Mockito.mock(FileSystem.class); + try (MockedStatic factory = Mockito.mockStatic(FileSystemFactory.class)) { + factory.when(() -> FileSystemFactory.getFileSystem(Mockito.any(StorageAdapter.class))).thenReturn(fs); + Mockito.when(fs.globListWithLimit(Mockito.any(Location.class), Mockito.isNull(), + Mockito.eq(1L), Mockito.eq(1L))).thenReturn(page("data/a.csv", "data/b.csv")); + Mockito.when(fs.globListWithLimit(Mockito.any(Location.class), Mockito.isNull(), + Mockito.eq(ONCE_PROPS.getS3BatchBytes()), Mockito.eq(1L))) + .thenReturn(page("data/a.csv", "data/b.csv")); + Mockito.when(fs.globListWithLimit(Mockito.any(Location.class), Mockito.eq("data/a.csv"), + Mockito.eq(1L), Mockito.eq(1L))) + .thenReturn(page("data/b.csv", "data/b.csv")); + Mockito.when(fs.globListWithLimit(Mockito.any(Location.class), Mockito.eq("data/a.csv"), + Mockito.eq(ONCE_PROPS.getS3BatchBytes()), Mockito.eq(1L))) + .thenReturn(page("data/b.csv", "data/b.csv")); + Mockito.when(fs.globListWithLimit(Mockito.any(Location.class), Mockito.eq("data/b.csv"), + Mockito.eq(1L), Mockito.eq(1L))) + .thenReturn(new GlobListing(Collections.emptyList(), "bucket", "data/", "")); + + S3SourceOffsetProvider provider = new S3SourceOffsetProvider(ONCE_PROPS); + Assertions.assertTrue(provider.hasMoreDataToConsume()); + provider.fetchRemoteMeta(TVF_PROPS); + Assertions.assertTrue(provider.hasMoreDataToConsume()); + S3Offset first = provider.getNextOffset(ONCE_PROPS, TVF_PROPS); + Assertions.assertEquals("data/a.csv", first.getEndFile()); + Assertions.assertFalse(first.isLastBatch()); + Assertions.assertEquals("data/a.csv", provider.getNextOffset(ONCE_PROPS, TVF_PROPS).getEndFile()); + provider.updateOffset(provider.deserializeOffset(first.toSerializedJson())); + Assertions.assertTrue(provider.hasMoreDataToConsume()); + S3Offset second = provider.getNextOffset(ONCE_PROPS, TVF_PROPS); + Assertions.assertEquals("data/b.csv", second.getEndFile()); + Assertions.assertTrue(second.isLastBatch()); + Assertions.assertFalse(provider.hasReachedEnd()); + // A concurrent metadata probe cannot change this task's completion decision. + provider.fetchRemoteMeta(TVF_PROPS); + Assertions.assertTrue(second.isLastBatch()); + Assertions.assertFalse(provider.hasReachedEnd()); + Assertions.assertEquals("data/b.csv", provider.getNextOffset(ONCE_PROPS, TVF_PROPS).getEndFile()); + provider.updateOffset(provider.deserializeOffset(second.toSerializedJson())); + Assertions.assertTrue(provider.hasReachedEnd()); + Assertions.assertFalse(provider.hasMoreDataToConsume()); + + S3SourceOffsetProvider recovered = new S3SourceOffsetProvider(ONCE_PROPS); + recovered.restoreFromPersistInfo(provider.getPersistInfo()); + Assertions.assertTrue(recovered.hasReachedEnd()); + Assertions.assertFalse(recovered.hasMoreDataToConsume()); + + recovered.fetchRemoteMeta(TVF_PROPS); + Assertions.assertTrue(recovered.hasReachedEnd()); + + Mockito.when(fs.globListWithLimit(Mockito.any(Location.class), Mockito.eq("data/b.csv"), + Mockito.eq(1L), Mockito.eq(1L))).thenReturn(page("data/c.csv", "data/c.csv")); + S3SourceOffsetProvider recoveredWithNewFile = new S3SourceOffsetProvider(ONCE_PROPS); + recoveredWithNewFile.restoreFromPersistInfo(provider.getPersistInfo()); + recoveredWithNewFile.fetchRemoteMeta(TVF_PROPS); + Assertions.assertTrue(recoveredWithNewFile.hasReachedEnd()); + Assertions.assertFalse(recoveredWithNewFile.hasMoreDataToConsume()); + + S3SourceOffsetProvider legacyWithNewFile = new S3SourceOffsetProvider(); + legacyWithNewFile.restoreFromPersistInfo("{\"endFile\":\"data/b.csv\"}"); + legacyWithNewFile.fetchRemoteMeta(TVF_PROPS); + Assertions.assertFalse(legacyWithNewFile.hasReachedEnd()); + Assertions.assertTrue(legacyWithNewFile.hasMoreDataToConsume()); + + S3SourceOffsetProvider lexical = new S3SourceOffsetProvider(); + lexical.restoreFromPersistInfo(provider.getPersistInfo()); + lexical.fetchRemoteMeta(TVF_PROPS); + Assertions.assertFalse(lexical.hasReachedEnd()); + Assertions.assertTrue(lexical.hasMoreDataToConsume()); + } + } + + @Test + public void testEmptyAndListingErrorRemainDistinct() throws Exception { + FileSystem fs = Mockito.mock(FileSystem.class); + try (MockedStatic factory = Mockito.mockStatic(FileSystemFactory.class)) { + factory.when(() -> FileSystemFactory.getFileSystem(Mockito.any(StorageAdapter.class))).thenReturn(fs); + Mockito.when(fs.globListWithLimit(Mockito.any(Location.class), Mockito.isNull(), + Mockito.anyLong(), Mockito.anyLong())) + .thenThrow(new IOException("listing failed")) + .thenReturn(new GlobListing(Collections.emptyList(), "bucket", "data/", "")); + + S3SourceOffsetProvider once = new S3SourceOffsetProvider(ONCE_PROPS); + Assertions.assertThrows(IOException.class, () -> once.fetchRemoteMeta(TVF_PROPS)); + Assertions.assertFalse(once.hasReachedEnd()); + once.fetchRemoteMeta(TVF_PROPS); + Assertions.assertFalse(once.hasReachedEnd()); + Assertions.assertTrue(once.hasMoreDataToConsume()); + RuntimeException emptyError = Assertions.assertThrows(RuntimeException.class, + () -> once.getNextOffset(ONCE_PROPS, TVF_PROPS)); + Assertions.assertTrue(emptyError.getMessage().contains("No new files found in path:")); + + S3SourceOffsetProvider lexical = new S3SourceOffsetProvider(); + Assertions.assertThrows(RuntimeException.class, () -> lexical.getNextOffset(ONCE_PROPS, TVF_PROPS)); + } + } + + private static GlobListing page(String key, String maxFile) { + return new GlobListing(List.of(new FileEntry(Location.of("s3://bucket/" + key), + 10, false, 0, null)), "bucket", "data/", maxFile); + } +} diff --git a/fe/fe-core/src/test/java/org/apache/doris/job/scheduler/StreamingTaskSchedulerTest.java b/fe/fe-core/src/test/java/org/apache/doris/job/scheduler/StreamingTaskSchedulerTest.java new file mode 100644 index 00000000000000..d2a5676102c3e2 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/job/scheduler/StreamingTaskSchedulerTest.java @@ -0,0 +1,70 @@ +// 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.doris.job.scheduler; + +import org.apache.doris.catalog.Env; +import org.apache.doris.common.jmockit.Deencapsulation; +import org.apache.doris.job.common.JobStatus; +import org.apache.doris.job.extensions.insert.streaming.AbstractStreamingTask; +import org.apache.doris.job.extensions.insert.streaming.StreamingInsertJob; +import org.apache.doris.job.manager.JobManager; +import org.apache.doris.job.manager.StreamingTaskManager; + +import org.junit.jupiter.api.Test; +import org.mockito.MockedStatic; +import org.mockito.Mockito; + +public class StreamingTaskSchedulerTest { + + @Test + public void testRecoveredSourceFinishesOrContinuesAfterProbe() throws Exception { + Env env = Mockito.mock(Env.class); + JobManager jobManager = Mockito.mock(JobManager.class); + StreamingTaskManager taskManager = Mockito.mock(StreamingTaskManager.class); + StreamingInsertJob exhaustedJob = Mockito.mock(StreamingInsertJob.class); + AbstractStreamingTask exhaustedTask = Mockito.mock(AbstractStreamingTask.class); + StreamingInsertJob continuedJob = Mockito.mock(StreamingInsertJob.class); + AbstractStreamingTask continuedTask = Mockito.mock(AbstractStreamingTask.class); + + Mockito.when(env.getJobManager()).thenReturn(jobManager); + Mockito.when(jobManager.getStreamingTaskManager()).thenReturn(taskManager); + Mockito.when(exhaustedTask.getJobId()).thenReturn(1L); + Mockito.doReturn(exhaustedJob).when(jobManager).getJob(1L); + Mockito.when(exhaustedJob.needScheduleTask()).thenReturn(true); + Mockito.when(exhaustedJob.hasMoreDataToConsume()).thenReturn(false); + Mockito.when(exhaustedJob.hasReachedEnd()).thenReturn(true); + Mockito.when(continuedTask.getJobId()).thenReturn(2L); + Mockito.doReturn(continuedJob).when(jobManager).getJob(2L); + Mockito.when(continuedJob.needScheduleTask()).thenReturn(true); + Mockito.when(continuedJob.hasMoreDataToConsume()).thenReturn(true); + + try (MockedStatic envMockedStatic = Mockito.mockStatic(Env.class)) { + envMockedStatic.when(Env::getCurrentEnv).thenReturn(env); + StreamingTaskScheduler scheduler = new StreamingTaskScheduler(); + + Deencapsulation.invoke(scheduler, "scheduleOneTask", exhaustedTask); + Mockito.verify(exhaustedJob).tryFinishJob(); + Mockito.verify(exhaustedTask, Mockito.never()).execute(); + + Deencapsulation.invoke(scheduler, "scheduleOneTask", continuedTask); + Mockito.verify(taskManager).addRunningTask(continuedTask); + Mockito.verify(continuedTask).execute(); + Mockito.verify(continuedJob, Mockito.never()).updateJobStatus(JobStatus.FINISHED); + } + } +} diff --git a/regression-test/suites/job_p0/streaming_job/test_streaming_insert_job_s3_once.groovy b/regression-test/suites/job_p0/streaming_job/test_streaming_insert_job_s3_once.groovy new file mode 100644 index 00000000000000..799312f3de9dda --- /dev/null +++ b/regression-test/suites/job_p0/streaming_job/test_streaming_insert_job_s3_once.groovy @@ -0,0 +1,107 @@ +// 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. + +import org.awaitility.Awaitility + +import static java.util.concurrent.TimeUnit.SECONDS + +suite("test_streaming_insert_job_s3_once") { + sql """DROP JOB IF EXISTS WHERE jobname = 'test_streaming_insert_job_s3_once'""" + sql """DROP TABLE IF EXISTS test_streaming_insert_job_s3_once_tbl FORCE""" + + sql """ + CREATE TABLE test_streaming_insert_job_s3_once_tbl ( + `c1` INT NULL, + `c2` STRING NULL, + `c3` INT NULL + ) ENGINE=OLAP + DUPLICATE KEY(`c1`) + DISTRIBUTED BY HASH(`c1`) BUCKETS 3 + PROPERTIES ("replication_allocation" = "tag.location.default: 1") + """ + + test { + sql """ + CREATE JOB test_streaming_insert_job_s3_once + PROPERTIES ( + "s3.ingestion_mode" = "ONCE", + "offset" = '{"fileName":"regression/load/data/example_0.csv"}' + ) + ON STREAMING DO INSERT INTO test_streaming_insert_job_s3_once_tbl + SELECT * FROM S3 ( + "uri" = "s3://${s3BucketName}/regression/load/data/example_[0-1].csv", + "format" = "csv", + "provider" = "${getS3Provider()}", + "column_separator" = ",", + "s3.endpoint" = "${getS3Endpoint()}", + "s3.region" = "${getS3Region()}", + "s3.access_key" = "${getS3AK()}", + "s3.secret_key" = "${getS3SK()}" + ) + """ + exception "offset is not supported when s3.ingestion_mode is ONCE" + } + + sql """ + CREATE JOB test_streaming_insert_job_s3_once + PROPERTIES ( + "s3.ingestion_mode" = "ONCE", + "s3.max_batch_files" = "1" + ) + ON STREAMING DO INSERT INTO test_streaming_insert_job_s3_once_tbl + SELECT * FROM S3 ( + "uri" = "s3://${s3BucketName}/regression/load/data/example_[0-1].csv", + "format" = "csv", + "provider" = "${getS3Provider()}", + "column_separator" = ",", + "s3.endpoint" = "${getS3Endpoint()}", + "s3.region" = "${getS3Region()}", + "s3.access_key" = "${getS3AK()}", + "s3.secret_key" = "${getS3SK()}" + ) + """ + + try { + Awaitility.await().atMost(300, SECONDS) + .pollInterval(1, SECONDS).until { + def job = sql """ + SELECT Status, SucceedTaskCount + FROM jobs("type"="insert") + WHERE Name = 'test_streaming_insert_job_s3_once' + AND ExecuteType = 'STREAMING' + """ + def rows = sql """SELECT COUNT(*) FROM test_streaming_insert_job_s3_once_tbl""" + log.info("S3 ONCE job: ${job}, row count: ${rows}") + job.size() == 1 + && job.get(0).get(0) == "FINISHED" + && job.get(0).get(1).toString() == "2" + && rows.get(0).get(0).toString() == "20" + } + } catch (Exception ex) { + def showJob = sql """ + SELECT * FROM jobs("type"="insert") + WHERE Name = 'test_streaming_insert_job_s3_once' + """ + def showTask = sql """ + SELECT * FROM tasks("type"="insert") + WHERE JobName = 'test_streaming_insert_job_s3_once' + """ + log.info("show job: " + showJob) + log.info("show task: " + showTask) + throw ex + } +}