This is an automated email from the ASF dual-hosted git repository.
nzw921rx pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 969a46d29d [Fix][CDC] Wait for enumerator ack before clearing finished
snapshot splits (#12163)
969a46d29d is described below
commit 969a46d29d60ab0e3aa81a22a4cc9b483081bdb3
Author: Jast <[email protected]>
AuthorDate: Fri Oct 23 00:46:13 2026 +0800
[Fix][CDC] Wait for enumerator ack before clearing finished snapshot splits
(#12163)
Co-authored-by: zhangshenghang <[email protected]>
Co-authored-by: jast <[email protected]>
Co-authored-by: Claude Code <[email protected]>
Co-authored-by: multica-agent <[email protected]>
---
.../source/reader/IncrementalSourceReader.java | 29 ++++++-
.../source/reader/IncrementalSourceReaderTest.java | 97 ++++++++++++++++++++++
.../serializable/TaskDataSerializerHook.java | 5 ++
.../engine/server/task/SourceSeaTunnelTask.java | 5 ++
.../context/SeaTunnelSplitEnumeratorContext.java | 24 +++++-
.../server/task/flow/SourceFlowLifeCycle.java | 14 ++++
.../source/SourceEnumeratorEventOperation.java | 71 ++++++++++++++++
7 files changed, 240 insertions(+), 5 deletions(-)
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/reader/IncrementalSourceReader.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/reader/IncrementalSourceReader.java
index 98b61593f6..4e577ee7cd 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/reader/IncrementalSourceReader.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/reader/IncrementalSourceReader.java
@@ -18,6 +18,7 @@
package org.apache.seatunnel.connectors.cdc.base.source.reader;
import org.apache.seatunnel.api.source.Collector;
+import org.apache.seatunnel.api.source.SourceEvent;
import org.apache.seatunnel.api.source.SourceReader;
import org.apache.seatunnel.api.table.catalog.CatalogTable;
import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
@@ -28,6 +29,7 @@ import
org.apache.seatunnel.connectors.cdc.base.config.JdbcSourceConfig;
import org.apache.seatunnel.connectors.cdc.base.config.SourceConfig;
import org.apache.seatunnel.connectors.cdc.base.dialect.DataSourceDialect;
import
org.apache.seatunnel.connectors.cdc.base.source.event.CompletedSnapshotPhaseEvent;
+import
org.apache.seatunnel.connectors.cdc.base.source.event.CompletedSnapshotSplitsAckEvent;
import
org.apache.seatunnel.connectors.cdc.base.source.event.CompletedSnapshotSplitsReportEvent;
import
org.apache.seatunnel.connectors.cdc.base.source.event.SnapshotSplitWatermark;
import org.apache.seatunnel.connectors.cdc.base.source.offset.Offset;
@@ -51,10 +53,10 @@ import lombok.extern.slf4j.Slf4j;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Supplier;
import java.util.stream.Collectors;
@@ -101,7 +103,9 @@ public class IncrementalSourceReader<T, C extends
SourceConfig>
context);
this.dataSourceDialect = dataSourceDialect;
this.sourceConfig = sourceConfig;
- this.finishedUnackedSplits = new HashMap<>();
+ // Ack events are handled on the engine's event-delivery thread while
splits finish on the
+ // reader thread, so the map must tolerate concurrent access.
+ this.finishedUnackedSplits = new ConcurrentHashMap<>();
this.subtaskId = context.getIndexOfSubtask();
this.debeziumDeserializationSchema = debeziumDeserializationSchema;
}
@@ -224,8 +228,10 @@ public class IncrementalSourceReader<T, C extends
SourceConfig>
new CompletedSnapshotSplitsReportEvent();
reportEvent.setCompletedSnapshotSplitWatermarks(completedSnapshotSplitWatermarks);
context.sendSourceEventToEnumerator(reportEvent);
- // TODO need enumerator return ack
- finishedUnackedSplits.clear();
+ // The splits stay in finishedUnackedSplits until the enumerator
returns a
+ // CompletedSnapshotSplitsAckEvent via handleSourceEvent. Clearing
them eagerly
+ // here would let a snapshot phase advance without the enumerator
ever knowing
+ // which splits the reader finished.
log.debug(
"The subtask {} reports offsets of finished snapshot
splits {}.",
subtaskId,
@@ -233,6 +239,21 @@ public class IncrementalSourceReader<T, C extends
SourceConfig>
}
}
+ @Override
+ public void handleSourceEvent(SourceEvent sourceEvent) {
+ if (sourceEvent instanceof CompletedSnapshotSplitsAckEvent) {
+ CompletedSnapshotSplitsAckEvent ackEvent =
+ (CompletedSnapshotSplitsAckEvent) sourceEvent;
+
ackEvent.getCompletedSplits().forEach(finishedUnackedSplits::remove);
+ log.debug(
+ "The subtask {} receives acknowledgements for finished
snapshot splits {}.",
+ subtaskId,
+ ackEvent.getCompletedSplits());
+ } else {
+ super.handleSourceEvent(sourceEvent);
+ }
+ }
+
@Override
protected SourceSplitStateBase initializedState(SourceSplitBase split) {
if (split.isSnapshotSplit()) {
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/reader/IncrementalSourceReaderTest.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/reader/IncrementalSourceReaderTest.java
index c7832751cb..18bebd9213 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/reader/IncrementalSourceReaderTest.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/reader/IncrementalSourceReaderTest.java
@@ -32,9 +32,13 @@ import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
import org.apache.seatunnel.connectors.cdc.base.config.JdbcSourceConfig;
import org.apache.seatunnel.connectors.cdc.base.config.SourceConfig;
import org.apache.seatunnel.connectors.cdc.base.dialect.DataSourceDialect;
+import
org.apache.seatunnel.connectors.cdc.base.source.event.CompletedSnapshotSplitsAckEvent;
+import
org.apache.seatunnel.connectors.cdc.base.source.event.CompletedSnapshotSplitsReportEvent;
+import
org.apache.seatunnel.connectors.cdc.base.source.event.SnapshotSplitWatermark;
import org.apache.seatunnel.connectors.cdc.base.source.offset.Offset;
import org.apache.seatunnel.connectors.cdc.base.source.offset.OffsetFactory;
import org.apache.seatunnel.connectors.cdc.base.source.split.IncrementalSplit;
+import org.apache.seatunnel.connectors.cdc.base.source.split.SnapshotSplit;
import org.apache.seatunnel.connectors.cdc.base.source.split.SourceRecords;
import org.apache.seatunnel.connectors.cdc.base.source.split.SourceSplitBase;
import
org.apache.seatunnel.connectors.cdc.base.source.split.state.SourceSplitStateBase;
@@ -78,6 +82,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.LockSupport;
+import java.util.stream.Collectors;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -578,6 +583,76 @@ class IncrementalSourceReaderTest {
}
}
+ @Test
+ void testAckRemovesFinishedSplitsFromUnackedState() {
+ SourceConfig sourceConfig = Mockito.mock(SourceConfig.class);
+ DataSourceDialect<SourceConfig> dialect =
+ Mockito.mock(DataSourceDialect.class,
Mockito.CALLS_REAL_METHODS);
+ SourceReader.Context context =
Mockito.mock(SourceReader.Context.class);
+ IncrementalSourceReader<Object, SourceConfig> reader =
+ createReader(dialect, sourceConfig, context);
+
+ try {
+
reader.addSplits(Collections.singletonList(finishedSnapshotSplit("snapshot-split-0")));
+
+ Assertions.assertEquals(
+ Collections.singletonList("snapshot-split-0"),
+ splitIds(reader.snapshotState(1L)));
+
+ reader.handleSourceEvent(
+ new CompletedSnapshotSplitsAckEvent(
+ Collections.singletonList("snapshot-split-0")));
+
+ Assertions.assertTrue(reader.snapshotState(2L).isEmpty());
+ } finally {
+ reader.close();
+ }
+ }
+
+ @Test
+ void testFinishedSplitsAreResentUntilAcknowledged() {
+ SourceConfig sourceConfig = Mockito.mock(SourceConfig.class);
+ DataSourceDialect<SourceConfig> dialect =
+ Mockito.mock(DataSourceDialect.class,
Mockito.CALLS_REAL_METHODS);
+ SourceReader.Context context =
Mockito.mock(SourceReader.Context.class);
+ List<CompletedSnapshotSplitsReportEvent> reports = new ArrayList<>();
+ Mockito.doAnswer(
+ invocation -> {
+ reports.add(invocation.getArgument(0));
+ return null;
+ })
+ .when(context)
+ .sendSourceEventToEnumerator(Mockito.any());
+ IncrementalSourceReader<Object, SourceConfig> reader =
+ createReader(dialect, sourceConfig, context);
+
+ try {
+ reader.addSplits(
+ Arrays.asList(
+ finishedSnapshotSplit("snapshot-split-0"),
+ finishedSnapshotSplit("snapshot-split-1")));
+
reader.addSplits(Collections.singletonList(finishedSnapshotSplit("snapshot-split-2")));
+
+ // no ack received yet, so the latest report still carries the
full unacked set
+ Assertions.assertEquals(
+ Arrays.asList("snapshot-split-0", "snapshot-split-1",
"snapshot-split-2"),
+ reportSplitIds(reports.get(reports.size() - 1)));
+
+ reader.handleSourceEvent(
+ new CompletedSnapshotSplitsAckEvent(
+ Arrays.asList(
+ "snapshot-split-0", "snapshot-split-1",
"snapshot-split-2")));
+
reader.addSplits(Collections.singletonList(finishedSnapshotSplit("snapshot-split-3")));
+
+ // previously acked splits are pruned from the report
+ Assertions.assertEquals(
+ Collections.singletonList("snapshot-split-3"),
+ reportSplitIds(reports.get(reports.size() - 1)));
+ } finally {
+ reader.close();
+ }
+ }
+
private static IncrementalSourceReader<Object, SourceConfig> createReader(
DataSourceDialect<SourceConfig> dialect,
SourceConfig sourceConfig,
@@ -640,6 +715,28 @@ class IncrementalSourceReaderTest {
historyTableChanges);
}
+ private static SnapshotSplit finishedSnapshotSplit(String splitId) {
+ return new SnapshotSplit(
+ splitId,
+ KEPT_TABLE,
+ null,
+ new Object[] {1},
+ new Object[] {2},
+ Mockito.mock(Offset.class),
+ Mockito.mock(Offset.class));
+ }
+
+ private static List<String> splitIds(List<SourceSplitBase> splits) {
+ return
splits.stream().map(SourceSplitBase::splitId).collect(Collectors.toList());
+ }
+
+ private static List<String>
reportSplitIds(CompletedSnapshotSplitsReportEvent event) {
+ return event.getCompletedSnapshotSplitWatermarks().stream()
+ .map(SnapshotSplitWatermark::getSplitId)
+ .sorted()
+ .collect(Collectors.toList());
+ }
+
private static CatalogTable catalogTable(TableId tableId) {
TablePath tablePath = TablePath.of(tableId.catalog(), tableId.table());
return CatalogTable.of(
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/serializable/TaskDataSerializerHook.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/serializable/TaskDataSerializerHook.java
index da2e185108..228d215f70 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/serializable/TaskDataSerializerHook.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/serializable/TaskDataSerializerHook.java
@@ -42,6 +42,7 @@ import
org.apache.seatunnel.engine.server.task.operation.source.CloseIdleReaderO
import
org.apache.seatunnel.engine.server.task.operation.source.LastCheckpointNotifyOperation;
import
org.apache.seatunnel.engine.server.task.operation.source.RequestSplitOperation;
import
org.apache.seatunnel.engine.server.task.operation.source.RestoredSplitOperation;
+import
org.apache.seatunnel.engine.server.task.operation.source.SourceEnumeratorEventOperation;
import
org.apache.seatunnel.engine.server.task.operation.source.SourceNoMoreElementOperation;
import
org.apache.seatunnel.engine.server.task.operation.source.SourceReaderEventOperation;
import
org.apache.seatunnel.engine.server.task.operation.source.SourceRegisterOperation;
@@ -110,6 +111,8 @@ public class TaskDataSerializerHook implements
DataSerializerHook {
public static final int REPORT_METRICS_OPERATION = 28;
+ public static final int SOURCE_ENUMERATOR_EVENT_TYPE = 29;
+
public static final int FACTORY_ID =
FactoryIdHelper.getFactoryId(
SeaTunnelFactoryIdConstant.SEATUNNEL_TASK_DATA_SERIALIZER_FACTORY,
@@ -170,6 +173,8 @@ public class TaskDataSerializerHook implements
DataSerializerHook {
return new CleanTaskGroupContextOperation();
case SOURCE_READER_EVENT_OPERATOR:
return new SourceReaderEventOperation();
+ case SOURCE_ENUMERATOR_EVENT_TYPE:
+ return new SourceEnumeratorEventOperation();
case CHECK_TASKGROUP_IS_EXECUTING:
return new CheckTaskGroupIsExecutingOperation();
case GET_METRICS_OPERATION:
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/SourceSeaTunnelTask.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/SourceSeaTunnelTask.java
index 1c78baf963..8ba6088eec 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/SourceSeaTunnelTask.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/SourceSeaTunnelTask.java
@@ -20,6 +20,7 @@ package org.apache.seatunnel.engine.server.task;
import org.apache.seatunnel.api.common.metrics.MetricsContext;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
import org.apache.seatunnel.api.serialization.Serializer;
+import org.apache.seatunnel.api.source.SourceEvent;
import org.apache.seatunnel.api.source.SourceSplit;
import org.apache.seatunnel.api.table.catalog.CatalogTable;
import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
@@ -162,6 +163,10 @@ public class SourceSeaTunnelTask<T, SplitT extends
SourceSplit> extends SeaTunne
((SourceFlowLifeCycle<T, SplitT>)
startFlowLifeCycle).receivedSplits(splits);
}
+ public void receivedSourceEvent(SourceEvent sourceEvent) {
+ ((SourceFlowLifeCycle<T, SplitT>)
startFlowLifeCycle).receivedSourceEvent(sourceEvent);
+ }
+
@Override
public void triggerBarrier(Barrier barrier) throws Exception {
SourceFlowLifeCycle<T, SplitT> sourceFlow =
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/context/SeaTunnelSplitEnumeratorContext.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/context/SeaTunnelSplitEnumeratorContext.java
index b4e3de60fb..03c2442115 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/context/SeaTunnelSplitEnumeratorContext.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/context/SeaTunnelSplitEnumeratorContext.java
@@ -22,9 +22,12 @@ import org.apache.seatunnel.api.event.EventListener;
import org.apache.seatunnel.api.source.SourceEvent;
import org.apache.seatunnel.api.source.SourceSplit;
import org.apache.seatunnel.api.source.SourceSplitEnumerator;
+import org.apache.seatunnel.engine.server.execution.TaskLocation;
import org.apache.seatunnel.engine.server.task.SourceSplitEnumeratorTask;
import
org.apache.seatunnel.engine.server.task.operation.source.AssignSplitOperation;
+import
org.apache.seatunnel.engine.server.task.operation.source.SourceEnumeratorEventOperation;
+import com.hazelcast.cluster.Address;
import lombok.extern.slf4j.Slf4j;
import java.util.Collections;
@@ -102,7 +105,26 @@ public class SeaTunnelSplitEnumeratorContext<SplitT
extends SourceSplit>
}
@Override
- public void sendEventToSourceReader(int subtaskId, SourceEvent event) {}
+ public void sendEventToSourceReader(int subtaskId, SourceEvent event) {
+ TaskLocation targetLocation =
task.getTaskMemberLocationByIndex(subtaskId);
+ Address targetAddress = task.getTaskMemberAddressByIndex(subtaskId);
+ if (targetLocation == null || targetAddress == null) {
+ log.warn(
+ "No registered reader for subtask {}, skip sending source
event {}",
+ subtaskId,
+ event);
+ return;
+ }
+ // Fire-and-forget: this can be invoked on a Hazelcast operation
thread while dispatching a
+ // reader event (e.g. a split report), so blocking on the send future
may deadlock the
+ // operation system. A lost event is safe to drop: sources that rely
on acknowledgements
+ // re-send their reports.
+ task.getExecutionContext()
+ .sendToMember(
+ new SourceEnumeratorEventOperation(
+ targetLocation, task.getTaskLocation(), event),
+ targetAddress);
+ }
@Override
public MetricsContext getMetricsContext() {
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/flow/SourceFlowLifeCycle.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/flow/SourceFlowLifeCycle.java
index 54c1e9af68..40a2cea117 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/flow/SourceFlowLifeCycle.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/flow/SourceFlowLifeCycle.java
@@ -461,6 +461,20 @@ public class SourceFlowLifeCycle<T, SplitT extends
SourceSplit> extends ActionFl
}
}
+ /**
+ * Delivers a source event sent by the remote split enumerator to this
reader.
+ *
+ * <p>This runs on a Hazelcast operation thread, not the reader's task
thread, so the
+ * implementation must be safe against concurrent access with the {@code
pollNext} loop. The
+ * event is dispatched to {@link
SourceReader#handleSourceEvent(SourceEvent)}, which connectors
+ * such as the CDC sources use to receive enumerator acknowledgements.
+ *
+ * @param sourceEvent the event sent by the enumerator
+ */
+ public void receivedSourceEvent(SourceEvent sourceEvent) {
+ reader.handleSourceEvent(sourceEvent);
+ }
+
/**
* Injects a checkpoint barrier into the record stream.
*
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/operation/source/SourceEnumeratorEventOperation.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/operation/source/SourceEnumeratorEventOperation.java
new file mode 100644
index 0000000000..206fe62cd3
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/operation/source/SourceEnumeratorEventOperation.java
@@ -0,0 +1,71 @@
+/*
+ * 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.seatunnel.engine.server.task.operation.source;
+
+import org.apache.seatunnel.api.source.SourceEvent;
+import org.apache.seatunnel.common.utils.RetryUtils;
+import org.apache.seatunnel.common.utils.SerializationUtils;
+import org.apache.seatunnel.engine.common.Constant;
+import org.apache.seatunnel.engine.server.SeaTunnelServer;
+import
org.apache.seatunnel.engine.server.exception.TaskGroupContextNotFoundException;
+import org.apache.seatunnel.engine.server.execution.TaskLocation;
+import org.apache.seatunnel.engine.server.serializable.TaskDataSerializerHook;
+import org.apache.seatunnel.engine.server.task.SourceSeaTunnelTask;
+
+/**
+ * For {@link org.apache.seatunnel.api.source.SourceSplitEnumerator} send
event to the {@link
+ * org.apache.seatunnel.api.source.SourceReader}
+ */
+public class SourceEnumeratorEventOperation extends SourceEventOperation {
+ public SourceEnumeratorEventOperation() {}
+
+ public SourceEnumeratorEventOperation(
+ TaskLocation targetTaskLocation, TaskLocation currentTaskLocation,
SourceEvent event) {
+ super(targetTaskLocation, currentTaskLocation, event);
+ }
+
+ @Override
+ public int getClassId() {
+ return TaskDataSerializerHook.SOURCE_ENUMERATOR_EVENT_TYPE;
+ }
+
+ @Override
+ public void runInternal() throws Exception {
+ SeaTunnelServer server = getService();
+ RetryUtils.retryWithException(
+ () -> {
+ SourceSeaTunnelTask<?, ?> task =
+
server.getTaskExecutionService().getTask(taskLocation);
+ ClassLoader classLoader =
+ server.getTaskExecutionService()
+
.getExecutionContext(taskLocation.getTaskGroupLocation())
+ .getClassLoader(task.getTaskID());
+ SourceEvent sourceEvent =
+ SerializationUtils.deserialize(this.sourceEvent,
classLoader);
+ task.receivedSourceEvent(sourceEvent);
+ return null;
+ },
+ new RetryUtils.RetryMaterial(
+ Constant.OPERATION_RETRY_TIME,
+ true,
+ exception ->
+ exception instanceof
TaskGroupContextNotFoundException
+ &&
!server.taskIsEnded(taskLocation.getTaskGroupLocation()),
+ Constant.OPERATION_RETRY_SLEEP));
+ }
+}
