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-12942-a89de9b70419116b47be4ecb4e018f12a942bfc6 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 4ae9604aaf876c64890cce88cc202a755bc36a64 Author: Goutam Adwant <[email protected]> AuthorDate: Mon Oct 26 14:45:05 2026 +0000 [Fix][Connector-V2] Restore bounded file splits after reader registration (#12942) Signed-off-by: Goutam Adwant <[email protected]> --- docs/en/connectors/source/LocalFile.md | 4 + docs/zh/connectors/source/LocalFile.md | 4 + .../MultipleTableFileSourceSplitEnumerator.java | 45 +++++-- ...MultipleTableFileSourceSplitEnumeratorTest.java | 147 ++++++++++++++++++++ .../e2e/connector/file/local/LocalFileIT.java | 148 +++++++++++++++++++++ .../resources/text/local_file_bounded_restore.conf | 44 ++++++ 6 files changed, 383 insertions(+), 9 deletions(-) diff --git a/docs/en/connectors/source/LocalFile.md b/docs/en/connectors/source/LocalFile.md index c169b133ec..7f98eb8618 100644 --- a/docs/en/connectors/source/LocalFile.md +++ b/docs/en/connectors/source/LocalFile.md @@ -49,6 +49,10 @@ If you use SeaTunnel Engine, It automatically integrated the hadoop jar when you ::: +## Checkpoint recovery + +For bounded Zeta jobs, restore resumes pending file splits after readers register and skips files already completed in the checkpoint. Keep input files unchanged and accessible at the same paths while restoring. Sink delivery guarantees still depend on the selected sink and its checkpoint support. + ## Options | name | type | required | default value | diff --git a/docs/zh/connectors/source/LocalFile.md b/docs/zh/connectors/source/LocalFile.md index c82cf03359..bc69c073cb 100644 --- a/docs/zh/connectors/source/LocalFile.md +++ b/docs/zh/connectors/source/LocalFile.md @@ -49,6 +49,10 @@ import ChangeLog from '../changelog/connector-file-local.md'; ::: +## 检查点恢复 + +对于 Zeta 批处理作业，恢复时会等待读取器注册后再分配未完成的文件分片，并跳过检查点中已经完成的文件。恢复期间，输入文件应保持不变，且仍可通过相同路径访问。最终写入语义仍取决于所选 Sink 及其检查点支持。 + ## 选项 | 名称 | 类型 | 是否必须 | 默认值 | diff --git a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/split/MultipleTableFileSourceSplitEnumerator.java b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/split/MultipleTableFileSourceSplitEnumerator.java index 188714cf26..33b6e5f6f0 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/split/MultipleTableFileSourceSplitEnumerator.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/split/MultipleTableFileSourceSplitEnumerator.java @@ -53,6 +53,8 @@ public class MultipleTableFileSourceSplitEnumerator private final List<BaseFileSourceConfig> fileSourceConfigs; private final AtomicInteger assignCount = new AtomicInteger(0); private final Object lock = new Object(); + // Guarded by lock; set only in run() after reader registration and kept true for late returns. + private boolean running; private final FileSplitStrategy fileSplitStrategy; private final Set<String> documentRoutingTableIds; @@ -150,25 +152,47 @@ public class MultipleTableFileSourceSplitEnumerator if (CollectionUtils.isEmpty(splits)) { return; } - allSplit.addAll(splits); - assignSplit(subtaskId); + synchronized (lock) { + allSplit.addAll(splits); + assignedSplit.removeAll(splits); + // Native readers restore their pending splits before registering. run() starts + // after registration; only a running enumerator can dispatch a late return. + if (running) { + assignPendingSplitsToRegisteredReaders(); + } + } } @Override public int currentUnassignedSplitSize() { - return allSplit.size() - assignedSplit.size(); + synchronized (lock) { + return (int) allSplit.stream().filter(split -> !assignedSplit.contains(split)).count(); + } } @Override public void handleSplitRequest(int subtaskId) {} @Override - public void registerReader(int subtaskId) {} + public void registerReader(int subtaskId) { + synchronized (lock) { + if (running) { + assignPendingSplitsToRegisteredReaders(); + } + } + } + + // Caller must hold lock while reading registered readers and updating split assignments. + private void assignPendingSplitsToRegisteredReaders() { + for (int reader : context.registeredReaders()) { + assignSplit(reader); + } + } @Override public FileSourceState snapshotState(long checkpointId) { synchronized (lock) { - return new FileSourceState(assignedSplit); + return new FileSourceState(new HashSet<>(assignedSplit)); } } @@ -197,7 +221,9 @@ public class MultipleTableFileSourceSplitEnumerator } } } - // assign splits + // Keep the full candidate order for round-robin ownership, then exclude files + // already assigned in the checkpoint. Reader-returned pending splits are unassigned. + currentTaskSplits.removeIf(assignedSplit::contains); context.assignSplit(taskId, currentTaskSplits); // save the state of assigned splits assignedSplit.addAll(currentTaskSplits); @@ -249,9 +275,10 @@ public class MultipleTableFileSourceSplitEnumerator @Override public void run() throws Exception { - for (int i = 0; i < context.currentParallelism(); i++) { - log.info("Assigned splits to reader [{}]", i); - synchronized (lock) { + synchronized (lock) { + running = true; + for (int i = 0; i < context.currentParallelism(); i++) { + log.info("Assigned splits to reader [{}]", i); assignSplit(i); } } diff --git a/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/split/MultipleTableFileSourceSplitEnumeratorTest.java b/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/split/MultipleTableFileSourceSplitEnumeratorTest.java index 83729daf01..af0b9e6052 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/split/MultipleTableFileSourceSplitEnumeratorTest.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/split/MultipleTableFileSourceSplitEnumeratorTest.java @@ -31,6 +31,7 @@ import org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseSourceOptio import org.apache.seatunnel.connectors.seatunnel.file.config.HadoopConf; import org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorException; import org.apache.seatunnel.connectors.seatunnel.file.source.FileSourceDocumentRouting; +import org.apache.seatunnel.connectors.seatunnel.file.source.state.FileSourceState; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FSDataInputStream; @@ -52,8 +53,10 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.stream.IntStream; class MultipleTableFileSourceSplitEnumeratorTest { @@ -263,6 +266,150 @@ class MultipleTableFileSourceSplitEnumeratorTest { Assertions.assertEquals(2, enumerator.currentUnassignedSplitSize()); } + @Test + void restoredSplitsWaitForReaderRegistration() throws Exception { + SourceSplitEnumerator.Context<FileSourceSplit> context = + Mockito.mock(SourceSplitEnumerator.Context.class); + Mockito.when(context.currentParallelism()).thenReturn(2); + Set<Integer> readers = new HashSet<>(); + Mockito.when(context.registeredReaders()).thenAnswer(invocation -> new HashSet<>(readers)); + List<FileSourceSplit> files = recoveryFiles(); + MultipleTableFileSourceSplitEnumerator enumerator = + recoveryEnumerator(context, files, new HashSet<>(files)); + enumerator.open(); + + // Native readers return checkpointed splits before registering with the enumerator. + enumerator.addSplitsBack(Collections.singletonList(files.get(1)), 0); + readers.add(0); + enumerator.registerReader(0); + enumerator.addSplitsBack(Collections.singletonList(files.get(2)), 1); + Mockito.verify(context, Mockito.never()).assignSplit(Mockito.anyInt(), Mockito.anyList()); + Mockito.verify(context, Mockito.never()).signalNoMoreSplits(Mockito.anyInt()); + + readers.add(1); + enumerator.registerReader(1); + Assertions.assertEquals(2, enumerator.currentUnassignedSplitSize()); + enumerator.run(); + ArgumentCaptor<List> assignments = ArgumentCaptor.forClass(List.class); + Mockito.verify(context, Mockito.times(2)) + .assignSplit(Mockito.anyInt(), assignments.capture()); + List<FileSourceSplit> delivered = new ArrayList<>(); + assignments.getAllValues().forEach(delivered::addAll); + Assertions.assertEquals(2, delivered.size()); + Assertions.assertEquals(new HashSet<>(files.subList(1, 3)), new HashSet<>(delivered)); + Assertions.assertEquals(0, enumerator.currentUnassignedSplitSize()); + } + + @Test + void restoreDoesNotReplayCheckpointCompletedFiles() throws Exception { + SourceSplitEnumerator.Context<FileSourceSplit> context = + Mockito.mock(SourceSplitEnumerator.Context.class); + Mockito.when(context.currentParallelism()).thenReturn(2); + List<FileSourceSplit> files = recoveryFiles(); + MultipleTableFileSourceSplitEnumerator enumerator = + recoveryEnumerator(context, files, new HashSet<>(files)); + enumerator.open(); + Assertions.assertEquals(0, enumerator.currentUnassignedSplitSize()); + enumerator.run(); + Mockito.verify(context, Mockito.times(2)) + .assignSplit(Mockito.anyInt(), Mockito.eq(Collections.emptyList())); + Mockito.verify(context).signalNoMoreSplits(0); + Mockito.verify(context).signalNoMoreSplits(1); + } + + @Test + void checkpointStateDoesNotChangeAfterLaterAssignment() throws Exception { + SourceSplitEnumerator.Context<FileSourceSplit> context = + Mockito.mock(SourceSplitEnumerator.Context.class); + Mockito.when(context.currentParallelism()).thenReturn(1); + MultipleTableFileSourceSplitEnumerator enumerator = + recoveryEnumerator(context, recoveryFiles(), Collections.emptySet()); + enumerator.open(); + FileSourceState checkpoint = enumerator.snapshotState(1L); + enumerator.run(); + Assertions.assertTrue(checkpoint.getAssignedSplit().isEmpty()); + Assertions.assertEquals(3, enumerator.snapshotState(2L).getAssignedSplit().size()); + } + + @Test + void lateReturnedSplitsAreAssignedWithoutReplayingOtherFiles() throws Exception { + SourceSplitEnumerator.Context<FileSourceSplit> context = + Mockito.mock(SourceSplitEnumerator.Context.class); + Mockito.when(context.currentParallelism()).thenReturn(2); + Mockito.when(context.registeredReaders()).thenReturn(new HashSet<>(Arrays.asList(0, 1))); + List<FileSourceSplit> files = recoveryFiles(); + MultipleTableFileSourceSplitEnumerator enumerator = + recoveryEnumerator(context, files, Collections.emptySet()); + enumerator.open(); + enumerator.run(); + Mockito.clearInvocations(context); + + // The reporting subtask need not be the split's route after redistribution. + enumerator.addSplitsBack(Collections.singletonList(files.get(1)), 0); + ArgumentCaptor<List> assignments = ArgumentCaptor.forClass(List.class); + Mockito.verify(context, Mockito.atLeastOnce()) + .assignSplit(Mockito.anyInt(), assignments.capture()); + List<FileSourceSplit> delivered = new ArrayList<>(); + assignments.getAllValues().forEach(delivered::addAll); + Assertions.assertEquals(Collections.singletonList(files.get(1)), delivered); + Assertions.assertEquals(0, enumerator.currentUnassignedSplitSize()); + } + + @Test + void completedFilesMissingFromListingDoNotMakePendingCountNegative() throws Exception { + SourceSplitEnumerator.Context<FileSourceSplit> context = + Mockito.mock(SourceSplitEnumerator.Context.class); + Mockito.when(context.currentParallelism()).thenReturn(1); + List<FileSourceSplit> files = recoveryFiles(); + MultipleTableFileSourceSplitEnumerator enumerator = + recoveryEnumerator( + context, Collections.singletonList(files.get(0)), new HashSet<>(files)); + enumerator.open(); + Assertions.assertEquals(0, enumerator.currentUnassignedSplitSize()); + enumerator.run(); + Mockito.verify(context).assignSplit(0, Collections.emptyList()); + } + + @Test + void repeatedRunDoesNotAssignFilesAgain() throws Exception { + SourceSplitEnumerator.Context<FileSourceSplit> context = + Mockito.mock(SourceSplitEnumerator.Context.class); + Mockito.when(context.currentParallelism()).thenReturn(2); + MultipleTableFileSourceSplitEnumerator enumerator = + recoveryEnumerator(context, recoveryFiles(), Collections.emptySet()); + enumerator.open(); + enumerator.run(); + Mockito.clearInvocations(context); + enumerator.run(); + Mockito.verify(context, Mockito.times(2)) + .assignSplit(Mockito.anyInt(), Mockito.eq(Collections.emptyList())); + } + + private static List<FileSourceSplit> recoveryFiles() { + return Arrays.asList( + new FileSourceSplit("catalog.test.files", "0.txt"), + new FileSourceSplit("catalog.test.files", "1.txt"), + new FileSourceSplit("catalog.test.files", "2.txt")); + } + + private static MultipleTableFileSourceSplitEnumerator recoveryEnumerator( + SourceSplitEnumerator.Context<FileSourceSplit> context, + List<FileSourceSplit> files, + Set<FileSourceSplit> checkpointAssigned) { + BaseFileSourceConfig file = Mockito.mock(BaseFileSourceConfig.class); + CatalogTable table = CatalogTableUtil.buildSimpleTextTable(); + Mockito.when(file.getCatalogTable()).thenReturn(table); + Mockito.when(file.getFilePathsForSplitEnumerator()) + .thenReturn(Collections.singletonList("input")); + BaseMultipleTableFileSourceConfig config = + Mockito.mock(BaseMultipleTableFileSourceConfig.class); + Mockito.when(config.getFileSourceConfigs()).thenReturn(Collections.singletonList(file)); + FileSplitStrategy strategy = Mockito.mock(FileSplitStrategy.class); + Mockito.when(strategy.split(Mockito.anyString(), Mockito.eq("input"))).thenReturn(files); + return new MultipleTableFileSourceSplitEnumerator( + context, config, strategy, new FileSourceState(checkpointAssigned)); + } + private int allocateFiles(int id, int parallelism, int fileSize) { int filesPerIteration = fileSize / parallelism; int remainder = fileSize % parallelism; diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-local-e2e/src/test/java/org/apache/seatunnel/e2e/connector/file/local/LocalFileIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-local-e2e/src/test/java/org/apache/seatunnel/e2e/connector/file/local/LocalFileIT.java index eb135349f2..34ead22e85 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-local-e2e/src/test/java/org/apache/seatunnel/e2e/connector/file/local/LocalFileIT.java +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-local-e2e/src/test/java/org/apache/seatunnel/e2e/connector/file/local/LocalFileIT.java @@ -909,6 +909,154 @@ public class LocalFileIT extends TestSuiteBase { Assertions.assertFalse(localFileCatalog.tableExists(tablePath)); } + @TestTemplate + @DisabledOnContainer( + value = {}, + type = {EngineType.SPARK, EngineType.FLINK}, + disabledReason = "Savepoint restore uses the native Zeta reader lifecycle") + public void testBoundedFileSavepointRestore(TestContainer container) throws Exception { + String jobId = String.valueOf(JobIdGenerator.newJobId()); + String config = "/text/local_file_bounded_restore.conf"; + String root = "/tmp/seatunnel_mnt/bounded-restore"; + Container.ExecResult setup = + baseContainer.execInContainer( + "sh", + "-c", + "rm -rf " + + root + + " && mkdir -p " + + root + + "/source " + + root + + "/output " + + root + + "/pending && i=1; while [ $i -le 120 ]; do echo $i > " + + root + + "/source/$i.txt; i=$((i+1)); done"); + Assertions.assertEquals(0, setup.getExitCode(), setup.getStderr()); + CompletableFuture<Container.ExecResult> firstRun = + CompletableFuture.supplyAsync( + () -> { + try { + return container.executeJob(config, jobId); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + List<CompletableFuture<Container.ExecResult>> runs = new ArrayList<>(); + runs.add(firstRun); + Throwable failure = null; + try { + Awaitility.await() + .atMost(120, TimeUnit.SECONDS) + .failFast( + "Bounded job ended before checkpointed partial output", + firstRun::isDone) + .untilAsserted( + () -> { + Assertions.assertTrue( + container.getCompletedCheckpointCount(jobId) > 0); + int committed = readBoundedRestoreIds().size(); + Assertions.assertTrue( + committed > 0 && committed < 120, + "Checkpoint must commit a proper subset before restore"); + }); + Container.ExecResult savepoint = container.savepointJob(jobId); + Assertions.assertEquals(0, savepoint.getExitCode(), savepoint.getStderr()); + Container.ExecResult stopped = firstRun.get(120, TimeUnit.SECONDS); + Assertions.assertEquals(0, stopped.getExitCode(), stopped.getStderr()); + Assertions.assertTrue( + readBoundedRestoreIds().size() < 120, + "The saved job must still have unfinished file splits"); + CompletableFuture<Container.ExecResult> restored = + CompletableFuture.supplyAsync( + () -> { + try { + return container.restoreJob(config, jobId); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + runs.add(restored); + Container.ExecResult completed = restored.get(180, TimeUnit.SECONDS); + Assertions.assertEquals(0, completed.getExitCode(), completed.getStderr()); + List<Integer> expected = new ArrayList<>(); + for (int id = 1; id <= 120; id++) { + expected.add(id); + } + List<Integer> actual = readBoundedRestoreIds(); + actual.sort(Integer::compareTo); + Assertions.assertEquals( + expected, + actual, + "Restore must neither replay completed files nor lose pending splits"); + } catch (Exception | AssertionError e) { + failure = e; + throw e; + } finally { + try { + List<String> terminal = + Lists.newArrayList("FINISHED", "FAILED", "CANCELED", "SAVEPOINT_DONE"); + String status = container.getJobStatus(jobId); + if (status == null || "UNKNOWABLE".equals(status)) { + throw new IllegalStateException( + "Cannot confirm native termination; preserve fixture files"); + } + if (!terminal.contains(status)) { + CompletableFuture<Container.ExecResult> cancellation = + CompletableFuture.supplyAsync( + () -> { + try { + return container.cancelJob(jobId); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + Container.ExecResult cancel = cancellation.get(30, TimeUnit.SECONDS); + Assertions.assertEquals(0, cancel.getExitCode(), cancel.getStderr()); + } + Awaitility.await() + .atMost(120, TimeUnit.SECONDS) + .untilAsserted( + () -> + Assertions.assertTrue( + terminal.contains(container.getJobStatus(jobId)), + "Keep fixture files until the native job is confirmed terminal")); + for (CompletableFuture<Container.ExecResult> run : runs) { + if (!run.isDone()) { + run.get(120, TimeUnit.SECONDS); + } + } + Container.ExecResult cleanup = + baseContainer.execInContainer("sh", "-c", "rm -rf " + root); + Assertions.assertEquals(0, cleanup.getExitCode(), cleanup.getStderr()); + } catch (Exception | AssertionError cleanupFailure) { + if (failure != null) { + failure.addSuppressed(cleanupFailure); + } else { + throw cleanupFailure; + } + } + } + } + + private List<Integer> readBoundedRestoreIds() throws IOException, InterruptedException { + Container.ExecResult result = + baseContainer.execInContainer( + "sh", + "-c", + "find /tmp/seatunnel_mnt/bounded-restore/output -type f " + + "! -name '.*' -exec awk '1' {} +"); + Assertions.assertEquals(0, result.getExitCode(), result.getStderr()); + List<Integer> ids = new ArrayList<>(); + for (String line : result.getStdout().split("\\R")) { + if (!line.trim().isEmpty()) { + ids.add(Integer.parseInt(line.trim())); + } + } + return ids; + } + private void resetUpdateTestPath() throws IOException, InterruptedException { Container.ExecResult result = baseContainer.execInContainer( diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-local-e2e/src/test/resources/text/local_file_bounded_restore.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-local-e2e/src/test/resources/text/local_file_bounded_restore.conf new file mode 100644 index 0000000000..df19ce2d34 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-local-e2e/src/test/resources/text/local_file_bounded_restore.conf @@ -0,0 +1,44 @@ +# +# 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. +# + +env { + parallelism = 2 + job.mode = "BATCH" + checkpoint.interval = 1000 + read_limit.rows_per_second = 2 +} +source { + LocalFile { + path = "/tmp/seatunnel_mnt/bounded-restore/source" + file_format_type = "text" + schema { + fields { + id = int + } + } + } +} +sink { + LocalFile { + path = "/tmp/seatunnel_mnt/bounded-restore/output" + tmp_path = "/tmp/seatunnel_mnt/bounded-restore/pending" + file_format_type = "text" + is_enable_transaction = true + file_name_expression = "${transactionId}" + row_delimiter = "\n" + } +}
