This is an automated email from the ASF dual-hosted git repository.
baiyangtx pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/amoro.git
The following commit(s) were added to refs/heads/master by this push:
new dd930a1ca Use SimpleFuture to refactor TaskRuntime (#3399)
dd930a1ca is described below
commit dd930a1ca61272fd7d98be29cdc5591708a84858
Author: majin1102 <[email protected]>
AuthorDate: Fri Jan 10 10:15:17 2025 +0800
Use SimpleFuture to refactor TaskRuntime (#3399)
* [AMORO-3258] Use SimpleFuture to refactor TaskRuntime and eliminate task
lock
* [AMORO-3265] fix ut test
* Optimize logics
* Spotless apply
* Fix ut issues
---------
Co-authored-by: majin.nathan <[email protected]>
---
.../amoro/server/DefaultOptimizingService.java | 2 +-
.../amoro/server/optimizing/OptimizingQueue.java | 81 ++++++++++++----------
.../amoro/server/optimizing/TaskRuntime.java | 51 ++++++--------
.../server/persistence/StatedPersistentBase.java | 22 +-----
.../amoro/server/table/DefaultTableService.java | 4 +-
.../apache/amoro/server/table/TableRuntime.java | 57 +++++++++------
.../amoro/server/TestDefaultOptimizingService.java | 8 +--
.../server/optimizing/TestOptimizingQueue.java | 6 +-
.../org/apache/amoro/process/OptimizingStages.java | 2 +-
.../org/apache/amoro/process/OptimizingState.java | 1 +
.../org/apache/amoro/process/ProcessStage.java | 6 ++
.../org/apache/amoro/process/ProcessStatus.java | 8 ++-
.../org/apache/amoro/process/SimpleFuture.java | 9 ++-
.../apache/amoro/process/TableProcessState.java | 4 ++
.../apache/amoro/optimizing/RewriteStageTask.java | 2 +-
15 files changed, 134 insertions(+), 129 deletions(-)
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/DefaultOptimizingService.java
b/amoro-ams/src/main/java/org/apache/amoro/server/DefaultOptimizingService.java
index 5a9d0c052..5d13c6dfb 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/DefaultOptimizingService.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/DefaultOptimizingService.java
@@ -199,7 +199,7 @@ public class DefaultOptimizingService extends
StatedPersistentBase
}
private OptimizingTask extractOptimizingTask(
- TaskRuntime task, String authToken, int threadId, OptimizingQueue queue)
{
+ TaskRuntime<?> task, String authToken, int threadId, OptimizingQueue
queue) {
try {
OptimizerThread optimizerThread =
getAuthenticatedOptimizer(authToken).getThread(threadId);
task.schedule(optimizerThread);
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java
b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java
index 64267b8c5..2ef3b550f 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java
@@ -79,7 +79,7 @@ public class OptimizingQueue extends PersistentBase {
private final QuotaProvider quotaProvider;
private final Queue<TableOptimizingProcess> tableQueue = new
LinkedTransferQueue<>();
- private final Queue<TaskRuntime<RewriteStageTask>> retryTaskQueue = new
LinkedTransferQueue<>();
+ private final Queue<TaskRuntime<?>> retryTaskQueue = new
LinkedTransferQueue<>();
private final SchedulingPolicy scheduler;
private final TableManager tableManager;
private final Executor planExecutor;
@@ -151,6 +151,14 @@ public class OptimizingQueue extends PersistentBase {
public void releaseTable(TableRuntime tableRuntime) {
scheduler.removeTable(tableRuntime);
+ List<OptimizingProcess> processList =
+ tableQueue.stream()
+ .filter(process -> process.tableRuntime == tableRuntime)
+ .collect(Collectors.toList());
+ for (OptimizingProcess process : processList) {
+ process.close();
+ clearProcess(process);
+ }
LOG.info(
"Release queue {} with table {}",
optimizerGroup.getName(),
@@ -161,15 +169,15 @@ public class OptimizingQueue extends PersistentBase {
return scheduler.getTableRuntime(identifier) != null;
}
- private void clearProcess(TableOptimizingProcess optimizingProcess) {
+ private void clearProcess(OptimizingProcess optimizingProcess) {
tableQueue.removeIf(process -> process.getProcessId() ==
optimizingProcess.getProcessId());
retryTaskQueue.removeIf(
taskRuntime -> taskRuntime.getTaskId().getProcessId() ==
optimizingProcess.getProcessId());
}
- public TaskRuntime pollTask(long maxWaitTime) {
+ public TaskRuntime<?> pollTask(long maxWaitTime) {
long deadline = calculateDeadline(maxWaitTime);
- TaskRuntime task = fetchTask();
+ TaskRuntime<?> task = fetchTask();
while (task == null && waitTask(deadline)) {
task = fetchTask();
}
@@ -196,11 +204,12 @@ public class OptimizingQueue extends PersistentBase {
}
}
- private TaskRuntime fetchTask() {
- return
Optional.ofNullable(retryTaskQueue.poll()).orElseGet(this::fetchScheduledTask);
+ private TaskRuntime<?> fetchTask() {
+ TaskRuntime<?> task = retryTaskQueue.poll();
+ return task != null ? task : fetchScheduledTask();
}
- private TaskRuntime fetchScheduledTask() {
+ private TaskRuntime<?> fetchScheduledTask() {
return tableQueue.stream()
.map(TableOptimizingProcess::poll)
.filter(Objects::nonNull)
@@ -307,7 +316,7 @@ public class OptimizingQueue extends PersistentBase {
.collect(Collectors.toList());
}
- public void retryTask(TaskRuntime taskRuntime) {
+ public void retryTask(TaskRuntime<?> taskRuntime) {
taskRuntime.reset();
retryTaskQueue.offer(taskRuntime);
}
@@ -349,7 +358,9 @@ public class OptimizingQueue extends PersistentBase {
return scheduler;
}
- private class TableOptimizingProcess implements OptimizingProcess,
TaskRuntime.TaskOwner {
+ private class TableOptimizingProcess implements OptimizingProcess {
+
+ private final Lock lock = new ReentrantLock();
private final long processId;
private final OptimizingType optimizingType;
private final TableRuntime tableRuntime;
@@ -358,7 +369,6 @@ public class OptimizingQueue extends PersistentBase {
private final long targetChangeSnapshotId;
private final Map<OptimizingTaskId, TaskRuntime<RewriteStageTask>> taskMap
= Maps.newHashMap();
private final Queue<TaskRuntime<RewriteStageTask>> taskQueue = new
LinkedList<>();
- private final Lock lock = new ReentrantLock();
private volatile ProcessStatus status = ProcessStatus.RUNNING;
private volatile String failedReason;
private long endTime = AmoroServiceConstants.INVALID_TIME;
@@ -366,10 +376,12 @@ public class OptimizingQueue extends PersistentBase {
private Map<String, Long> toSequence = Maps.newHashMap();
private boolean hasCommitted = false;
- public TaskRuntime poll() {
+ public TaskRuntime<?> poll() {
lock.lock();
try {
- return taskQueue.poll();
+ return status != ProcessStatus.CLOSED && status != ProcessStatus.FAILED
+ ? taskQueue.poll()
+ : null;
} finally {
lock.unlock();
}
@@ -431,16 +443,13 @@ public class OptimizingQueue extends PersistentBase {
}
this.status = ProcessStatus.CLOSED;
this.endTime = System.currentTimeMillis();
- persistProcessCompleted(false);
- clearProcess(this);
+ persistAndSetCompleted(false);
} finally {
lock.unlock();
}
- releaseResourcesIfNecessary();
}
- @Override
- public void acceptResult(TaskRuntime taskRuntime) {
+ private void acceptResult(TaskRuntime<?> taskRuntime) {
lock.lock();
try {
try {
@@ -452,6 +461,10 @@ public class OptimizingQueue extends PersistentBase {
taskRuntime.getTaskId(),
throwable);
}
+ // task cancel means persistAndSetCompleted has been called
+ if (taskRuntime.getStatus() == TaskRuntime.Status.CANCELED) {
+ return;
+ }
if (isClosed()) {
throw new OptimizingClosedException(processId);
}
@@ -461,7 +474,6 @@ public class OptimizingQueue extends PersistentBase {
&& tableRuntime.getOptimizingStatus().isProcessing()
&& tableRuntime.getOptimizingStatus() !=
OptimizingStatus.COMMITTING) {
tableRuntime.beginCommitting();
- clearProcess(this);
}
} else if (taskRuntime.getStatus() == TaskRuntime.Status.FAILED) {
if (taskRuntime.getRetry() < tableRuntime.getMaxExecuteRetryCount())
{
@@ -471,11 +483,10 @@ public class OptimizingQueue extends PersistentBase {
taskRuntime.getFailReason());
retryTask(taskRuntime);
} else {
- clearProcess(this);
this.failedReason = taskRuntime.getFailReason();
this.status = ProcessStatus.FAILED;
this.endTime = taskRuntime.getEndTime();
- persistProcessCompleted(false);
+ persistAndSetCompleted(false);
}
}
} finally {
@@ -483,14 +494,6 @@ public class OptimizingQueue extends PersistentBase {
}
}
- // the cleanup of task should be done after unlock to avoid deadlock
- @Override
- public void releaseResourcesIfNecessary() {
- if (this.status == ProcessStatus.FAILED || this.status ==
ProcessStatus.CLOSED) {
- cancelTasks();
- }
- }
-
@Override
public boolean isClosed() {
return status == ProcessStatus.CLOSED;
@@ -572,15 +575,13 @@ public class OptimizingQueue extends PersistentBase {
buildCommit().commit();
status = ProcessStatus.SUCCESS;
endTime = System.currentTimeMillis();
- persistProcessCompleted(true);
+ persistAndSetCompleted(true);
} catch (Exception e) {
LOG.error("{} Commit optimizing failed ",
tableRuntime.getTableIdentifier(), e);
status = ProcessStatus.FAILED;
failedReason = ExceptionUtil.getErrorMessage(e, 4000);
endTime = System.currentTimeMillis();
- persistProcessCompleted(false);
- } finally {
- clearProcess(this);
+ persistAndSetCompleted(false);
}
} finally {
lock.unlock();
@@ -654,8 +655,13 @@ public class OptimizingQueue extends PersistentBase {
() -> tableRuntime.beginProcess(this));
}
- private void persistProcessCompleted(boolean success) {
+ private void persistAndSetCompleted(boolean success) {
doAsTransaction(
+ () -> {
+ if (!success) {
+ cancelTasks();
+ }
+ },
() ->
doAs(
OptimizingMapper.class,
@@ -667,10 +673,10 @@ public class OptimizingQueue extends PersistentBase {
endTime,
getSummary(),
getFailedReason())),
- () -> tableRuntime.completeProcess(success));
+ () -> tableRuntime.completeProcess(success),
+ () -> clearProcess(this));
}
- /** The cancellation should be invoked outside the process lock to avoid
deadlock. */
private void cancelTasks() {
taskMap.values().forEach(TaskRuntime::tryCanceling);
}
@@ -686,7 +692,7 @@ public class OptimizingQueue extends PersistentBase {
Map<Integer, RewriteFilesInput> inputs =
TaskFilesPersistence.loadTaskInputs(processId);
taskRuntimes.forEach(
taskRuntime -> {
- taskRuntime.claimOwnership(this);
+ taskRuntime.getCompletedFuture().whenCompleted(() ->
acceptResult(taskRuntime));
taskRuntime
.getTaskDescriptor()
.setInput(inputs.get(taskRuntime.getTaskId().getTaskId()));
@@ -716,7 +722,8 @@ public class OptimizingQueue extends PersistentBase {
tableRuntime.getTableIdentifier(),
taskRuntime.getTaskId(),
taskRuntime.getSummary());
- taskMap.put(taskRuntime.getTaskId(), taskRuntime.claimOwnership(this));
+ taskRuntime.getCompletedFuture().whenCompleted(() ->
acceptResult(taskRuntime));
+ taskMap.put(taskRuntime.getTaskId(), taskRuntime);
taskQueue.offer(taskRuntime);
}
}
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/TaskRuntime.java
b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/TaskRuntime.java
index 85adbdcdd..59339dece 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/TaskRuntime.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/TaskRuntime.java
@@ -23,8 +23,8 @@ import org.apache.amoro.api.OptimizingTask;
import org.apache.amoro.api.OptimizingTaskId;
import org.apache.amoro.api.OptimizingTaskResult;
import org.apache.amoro.exception.IllegalTaskStateException;
-import org.apache.amoro.exception.OptimizingClosedException;
import org.apache.amoro.exception.TaskRuntimeException;
+import org.apache.amoro.process.SimpleFuture;
import org.apache.amoro.process.StagedTaskDescriptor;
import org.apache.amoro.server.AmoroServiceConstants;
import org.apache.amoro.server.persistence.StatedPersistentBase;
@@ -39,10 +39,11 @@ import java.util.Set;
public class TaskRuntime<T extends StagedTaskDescriptor<?, ?, ?>> extends
StatedPersistentBase {
+ private final SimpleFuture future = new SimpleFuture();
+ private final TaskStatusMachine statusMachine = new TaskStatusMachine();
private OptimizingTaskId taskId;
private T taskDescriptor;
@StateField private Status status = Status.PLANNED;
- private final TaskStatusMachine statusMachine = new TaskStatusMachine();
@StateField private int runTimes = 0;
@StateField private long startTime = AmoroServiceConstants.INVALID_TIME;
@StateField private long endTime = AmoroServiceConstants.INVALID_TIME;
@@ -50,7 +51,6 @@ public class TaskRuntime<T extends StagedTaskDescriptor<?, ?,
?>> extends Stated
@StateField private String token;
@StateField private int threadId = -1;
@StateField private String failReason;
- private TaskOwner owner;
private TaskRuntime() {}
@@ -63,6 +63,10 @@ public class TaskRuntime<T extends StagedTaskDescriptor<?,
?, ?>> extends Stated
return taskDescriptor;
}
+ public SimpleFuture getCompletedFuture() {
+ return future;
+ }
+
public void complete(OptimizerThread thread, OptimizingTaskResult result) {
invokeConsistency(
() -> {
@@ -78,16 +82,18 @@ public class TaskRuntime<T extends StagedTaskDescriptor<?,
?, ?>> extends Stated
costTime += endTime - startTime;
runTimes += 1;
persistTaskRuntime();
- owner.acceptResult(this);
+ future.complete();
token = null;
threadId = -1;
});
- owner.releaseResourcesIfNecessary();
}
void reset() {
invokeConsistency(
() -> {
+ if (status == Status.PLANNED) {
+ return;
+ }
statusMachine.accept(Status.PLANNED);
startTime = AmoroServiceConstants.INVALID_TIME;
endTime = AmoroServiceConstants.INVALID_TIME;
@@ -95,6 +101,7 @@ public class TaskRuntime<T extends StagedTaskDescriptor<?,
?, ?>> extends Stated
threadId = -1;
failReason = null;
taskDescriptor.reset();
+ future.reset();
// The cost time should not be reset since it is the total cost time
of all runs.
persistTaskRuntime();
});
@@ -129,15 +136,11 @@ public class TaskRuntime<T extends
StagedTaskDescriptor<?, ?, ?>> extends Stated
costTime += endTime - startTime;
}
persistTaskRuntime();
+ future.complete();
}
});
}
- public TaskRuntime<T> claimOwnership(TaskOwner owner) {
- this.owner = owner;
- return this;
- }
-
public boolean finished() {
return this.status == Status.SUCCESS
|| this.status == Status.FAILED
@@ -262,25 +265,19 @@ public class TaskRuntime<T extends
StagedTaskDescriptor<?, ?, ?>> extends Stated
private static final Map<Status, Set<Status>> nextStatusMap =
ImmutableMap.<Status, Set<Status>>builder()
- .put(Status.PLANNED, ImmutableSet.of(Status.PLANNED,
Status.SCHEDULED, Status.CANCELED))
- .put(
- Status.SCHEDULED,
- ImmutableSet.of(Status.PLANNED, Status.SCHEDULED, Status.ACKED,
Status.CANCELED))
+ .put(Status.PLANNED, ImmutableSet.of(Status.SCHEDULED,
Status.CANCELED))
+ .put(Status.SCHEDULED, ImmutableSet.of(Status.PLANNED, Status.ACKED,
Status.CANCELED))
.put(
Status.ACKED,
- ImmutableSet.of(
- Status.PLANNED, Status.ACKED, Status.SUCCESS, Status.FAILED,
Status.CANCELED))
- .put(Status.FAILED, ImmutableSet.of(Status.PLANNED, Status.FAILED))
- .put(Status.SUCCESS, ImmutableSet.of(Status.SUCCESS))
- .put(Status.CANCELED, ImmutableSet.of(Status.CANCELED))
+ ImmutableSet.of(Status.PLANNED, Status.SUCCESS, Status.FAILED,
Status.CANCELED))
+ .put(Status.FAILED, ImmutableSet.of(Status.PLANNED))
+ .put(Status.CANCELED, ImmutableSet.of())
+ .put(Status.SUCCESS, ImmutableSet.of())
.build();
private class TaskStatusMachine {
public void accept(Status targetStatus) {
- if (owner.isClosed()) {
- throw new OptimizingClosedException(taskId.getProcessId());
- }
if (!getNext().contains(targetStatus)) {
throw new IllegalTaskStateException(taskId, status.name(),
targetStatus.name());
}
@@ -321,7 +318,7 @@ public class TaskRuntime<T extends StagedTaskDescriptor<?,
?, ?>> extends Stated
public TaskQuota() {}
- public TaskQuota(TaskRuntime task) {
+ public TaskQuota(TaskRuntime<?> task) {
this.startTime = task.getStartTime();
this.endTime = task.getEndTime();
this.processId = task.getTaskId().getProcessId();
@@ -367,12 +364,4 @@ public class TaskRuntime<T extends StagedTaskDescriptor<?,
?, ?>> extends Stated
return endTime <= validTime;
}
}
-
- public interface TaskOwner {
- void acceptResult(TaskRuntime taskRuntime);
-
- void releaseResourcesIfNecessary();
-
- boolean isClosed();
- }
}
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/persistence/StatedPersistentBase.java
b/amoro-ams/src/main/java/org/apache/amoro/server/persistence/StatedPersistentBase.java
index fad12dbe4..fc551d2c6 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/persistence/StatedPersistentBase.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/persistence/StatedPersistentBase.java
@@ -27,8 +27,6 @@ import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Optional;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Supplier;
import java.util.stream.Collectors;
@@ -37,45 +35,29 @@ public abstract class StatedPersistentBase extends
PersistentBase {
private static final Map<Class<? extends PersistentBase>, Field[]> metaCache
=
Maps.newConcurrentMap();
private static final Object NULL_VALUE = new Object();
- private final Lock stateLock = new ReentrantLock();
private final Field[] consistentFields;
protected StatedPersistentBase() {
consistentFields = getOrCreateConsistentFields();
}
- protected final void invokeConsistency(Runnable runnable) {
- stateLock.lock();
+ protected void invokeConsistency(Runnable runnable) {
Map<Field, Object> states = retainStates();
try {
doAsTransaction(runnable);
} catch (Throwable throwable) {
restoreStates(states);
throw throwable;
- } finally {
- stateLock.unlock();
}
}
- protected final <T> T invokeConsistency(Supplier<T> supplier) {
- stateLock.lock();
+ protected <T> T invokeConsistency(Supplier<T> supplier) {
Map<Field, Object> states = retainStates();
try {
return supplier.get();
} catch (Throwable throwable) {
restoreStates(states);
throw throwable;
- } finally {
- stateLock.unlock();
- }
- }
-
- protected final void invokeInStateLock(Runnable runnable) {
- stateLock.lock();
- try {
- runnable.run();
- } finally {
- stateLock.unlock();
}
}
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableService.java
b/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableService.java
index 565cce4e0..a43e91724 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableService.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableService.java
@@ -44,7 +44,7 @@ import org.apache.amoro.server.catalog.InternalCatalog;
import org.apache.amoro.server.catalog.ServerCatalog;
import org.apache.amoro.server.manager.MetricManager;
import org.apache.amoro.server.optimizing.OptimizingStatus;
-import org.apache.amoro.server.persistence.StatedPersistentBase;
+import org.apache.amoro.server.persistence.PersistentBase;
import org.apache.amoro.server.persistence.TableRuntimeMeta;
import org.apache.amoro.server.persistence.mapper.CatalogMetaMapper;
import org.apache.amoro.server.persistence.mapper.TableBlockerMapper;
@@ -84,7 +84,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
-public class DefaultTableService extends StatedPersistentBase implements
TableService {
+public class DefaultTableService extends PersistentBase implements
TableService {
public static final Logger LOG =
LoggerFactory.getLogger(DefaultTableService.class);
private static final int TABLE_BLOCKER_RETRY = 3;
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/table/TableRuntime.java
b/amoro-ams/src/main/java/org/apache/amoro/server/table/TableRuntime.java
index a21967d76..f6733af83 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/table/TableRuntime.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/table/TableRuntime.java
@@ -57,11 +57,13 @@ import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
public class TableRuntime extends StatedPersistentBase {
private static final Logger LOG =
LoggerFactory.getLogger(TableRuntime.class);
-
+ private final Lock tableLock = new ReentrantLock();
private final TableRuntimeHandler tableHandler;
private final ServerTableIdentifier tableIdentifier;
private final List<TaskRuntime.TaskQuota> taskQuotas = new
CopyOnWriteArrayList<>();
@@ -90,14 +92,10 @@ public class TableRuntime extends StatedPersistentBase {
private final TableOptimizingMetrics optimizingMetrics;
private final TableOrphanFilesCleaningMetrics orphanFilesCleaningMetrics;
private final TableSummaryMetrics tableSummaryMetrics;
-
private long targetSnapshotId;
-
private long targetChangeSnapshotId;
-
private Map<String, Long> fromSequence;
private Map<String, Long> toSequence;
-
private OptimizingType optimizingType;
public TableRuntime(
@@ -176,15 +174,17 @@ public class TableRuntime extends StatedPersistentBase {
}
public void dispose() {
- invokeInStateLock(
- () -> {
- doAsTransaction(
- () ->
Optional.ofNullable(optimizingProcess).ifPresent(OptimizingProcess::close),
- () ->
- doAs(
- TableMetaMapper.class,
- mapper ->
mapper.deleteOptimizingRuntime(tableIdentifier.getId())));
- });
+ tableLock.lock();
+ try {
+ doAsTransaction(
+ () ->
Optional.ofNullable(optimizingProcess).ifPresent(OptimizingProcess::close),
+ () ->
+ doAs(
+ TableMetaMapper.class,
+ mapper ->
mapper.deleteOptimizingRuntime(tableIdentifier.getId())));
+ } finally {
+ tableLock.unlock();
+ }
optimizingMetrics.unregister();
orphanFilesCleaningMetrics.unregister();
tableSummaryMetrics.unregister();
@@ -318,15 +318,16 @@ public class TableRuntime extends StatedPersistentBase {
* @param startTimeMills
*/
public void resetTaskQuotas(long startTimeMills) {
- invokeInStateLock(
- () -> {
- taskQuotas.clear();
- taskQuotas.addAll(
- getAs(
- OptimizingMapper.class,
- mapper ->
- mapper.selectTaskQuotasByTime(tableIdentifier.getId(),
startTimeMills)));
- });
+ tableLock.lock();
+ try {
+ taskQuotas.clear();
+ taskQuotas.addAll(
+ getAs(
+ OptimizingMapper.class,
+ mapper -> mapper.selectTaskQuotasByTime(tableIdentifier.getId(),
startTimeMills)));
+ } finally {
+ tableLock.unlock();
+ }
}
public void completeProcess(boolean success) {
@@ -646,4 +647,14 @@ public class TableRuntime extends StatedPersistentBase {
System.currentTimeMillis()));
return TableBlocker.conflict(operation, tableBlockers);
}
+
+ @Override
+ protected void invokeConsistency(Runnable runnable) {
+ tableLock.lock();
+ try {
+ super.invokeConsistency(runnable);
+ } finally {
+ tableLock.unlock();
+ }
+ }
}
diff --git
a/amoro-ams/src/test/java/org/apache/amoro/server/TestDefaultOptimizingService.java
b/amoro-ams/src/test/java/org/apache/amoro/server/TestDefaultOptimizingService.java
index b780f6107..a5c406766 100644
---
a/amoro-ams/src/test/java/org/apache/amoro/server/TestDefaultOptimizingService.java
+++
b/amoro-ams/src/test/java/org/apache/amoro/server/TestDefaultOptimizingService.java
@@ -186,7 +186,7 @@ public class TestDefaultOptimizingService extends
AMSTableTestBase {
optimizingService().ackTask(token, THREAD_ID, task.getTaskId());
assertTaskStatus(TaskRuntime.Status.ACKED);
- TaskRuntime taskRuntime =
+ TaskRuntime<?> taskRuntime =
optimizingService().listTasks(defaultResourceGroup().getName()).get(0);
optimizingService().completeTask(token,
buildOptimizingTaskResult(task.getTaskId()));
assertTaskCompleted(taskRuntime);
@@ -310,7 +310,7 @@ public class TestDefaultOptimizingService extends
AMSTableTestBase {
reload();
assertTaskStatus(TaskRuntime.Status.ACKED);
- TaskRuntime taskRuntime =
+ TaskRuntime<?> taskRuntime =
optimizingService().listTasks(defaultResourceGroup().getName()).get(0);
optimizingService().completeTask(token,
buildOptimizingTaskResult(task.getTaskId()));
assertTaskCompleted(taskRuntime);
@@ -658,12 +658,10 @@ public class TestDefaultOptimizingService extends
AMSTableTestBase {
optimizingService().listTasks(defaultResourceGroup().getName()).get(0).getStatus());
}
- private void assertTaskCompleted(TaskRuntime taskRuntime) {
+ private void assertTaskCompleted(TaskRuntime<?> taskRuntime) {
if (taskRuntime != null) {
Assertions.assertEquals(TaskRuntime.Status.SUCCESS,
taskRuntime.getStatus());
}
- Assertions.assertEquals(
- 0,
optimizingService().listTasks(defaultResourceGroup().getName()).size());
Assertions.assertEquals(
ProcessStatus.RUNNING,
tableService()
diff --git
a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestOptimizingQueue.java
b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestOptimizingQueue.java
index bf668fbcf..eeeba78c3 100644
---
a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestOptimizingQueue.java
+++
b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestOptimizingQueue.java
@@ -174,12 +174,12 @@ public class TestOptimizingQueue extends AMSTableTestBase
{
OptimizingQueue queue = buildOptimizingGroupService(tableRuntimeMeta);
// 1.poll task
- TaskRuntime task = queue.pollTask(MAX_POLLING_TIME);
+ TaskRuntime<?> task = queue.pollTask(MAX_POLLING_TIME);
Assert.assertNotNull(task);
for (int i = 0; i <
TableProperties.SELF_OPTIMIZING_EXECUTE_RETRY_NUMBER_DEFAULT; i++) {
queue.retryTask(task);
- TaskRuntime retryTask = queue.pollTask(MAX_POLLING_TIME);
+ TaskRuntime<?> retryTask = queue.pollTask(MAX_POLLING_TIME);
Assert.assertEquals(retryTask.getTaskId(), task.getTaskId());
retryTask.schedule(optimizerThread);
retryTask.ack(optimizerThread);
@@ -190,7 +190,7 @@ public class TestOptimizingQueue extends AMSTableTestBase {
}
queue.retryTask(task);
- TaskRuntime retryTask = queue.pollTask(MAX_POLLING_TIME);
+ TaskRuntime<?> retryTask = queue.pollTask(MAX_POLLING_TIME);
Assert.assertEquals(retryTask.getTaskId(), task.getTaskId());
retryTask.schedule(optimizerThread);
retryTask.ack(optimizerThread);
diff --git
a/amoro-common/src/main/java/org/apache/amoro/process/OptimizingStages.java
b/amoro-common/src/main/java/org/apache/amoro/process/OptimizingStages.java
index 6049621cd..f4bec4a2a 100644
--- a/amoro-common/src/main/java/org/apache/amoro/process/OptimizingStages.java
+++ b/amoro-common/src/main/java/org/apache/amoro/process/OptimizingStages.java
@@ -48,7 +48,7 @@ public class OptimizingStages {
private static final Map<String, ProcessStage> STAGES =
ImmutableMap.<String, ProcessStage>builder()
.put(MINOR.getDesc(), MINOR)
- .put(MINOR.getDesc(), MAJOR)
+ .put(MAJOR.getDesc(), MAJOR)
.put(FULL.getDesc(), FULL)
.put(COMMITTING.getDesc(), COMMITTING)
.put(PLANNING.getDesc(), PLANNING)
diff --git
a/amoro-common/src/main/java/org/apache/amoro/process/OptimizingState.java
b/amoro-common/src/main/java/org/apache/amoro/process/OptimizingState.java
index 99691cf84..c4a900dca 100644
--- a/amoro-common/src/main/java/org/apache/amoro/process/OptimizingState.java
+++ b/amoro-common/src/main/java/org/apache/amoro/process/OptimizingState.java
@@ -60,6 +60,7 @@ public abstract class OptimizingState extends
TableProcessState {
return watermark;
}
+ @Override
public ProcessStage getStage() {
return stage;
}
diff --git
a/amoro-common/src/main/java/org/apache/amoro/process/ProcessStage.java
b/amoro-common/src/main/java/org/apache/amoro/process/ProcessStage.java
index 47be213ce..ee74627e2 100644
--- a/amoro-common/src/main/java/org/apache/amoro/process/ProcessStage.java
+++ b/amoro-common/src/main/java/org/apache/amoro/process/ProcessStage.java
@@ -20,6 +20,12 @@ package org.apache.amoro.process;
public class ProcessStage {
+ /**
+ * An arbitrary stage belongs to processes which only contain one stage For
those one stage
+ * processes, should transform ${@link ProcessStage} to ProcessStage for
displaying
+ */
+ public static final ProcessStage ARBITRARY = new ProcessStage("", 0);
+
/**
* Action Stage description value, normally this value should be identical
within certain actions
*/
diff --git
a/amoro-common/src/main/java/org/apache/amoro/process/ProcessStatus.java
b/amoro-common/src/main/java/org/apache/amoro/process/ProcessStatus.java
index 77d2ae023..43f9471b5 100644
--- a/amoro-common/src/main/java/org/apache/amoro/process/ProcessStatus.java
+++ b/amoro-common/src/main/java/org/apache/amoro/process/ProcessStatus.java
@@ -22,9 +22,13 @@ package org.apache.amoro.process;
public enum ProcessStatus {
UNKNOWN,
PENDING,
- RUNNING,
SUBMITTED,
+ RUNNING,
SUCCESS,
CLOSED,
- FAILED
+ FAILED;
+
+ public ProcessStage toStage() {
+ return new ProcessStage(name(), ordinal());
+ }
}
diff --git
a/amoro-common/src/main/java/org/apache/amoro/process/SimpleFuture.java
b/amoro-common/src/main/java/org/apache/amoro/process/SimpleFuture.java
index ce1332bfd..f69e5e9a2 100644
--- a/amoro-common/src/main/java/org/apache/amoro/process/SimpleFuture.java
+++ b/amoro-common/src/main/java/org/apache/amoro/process/SimpleFuture.java
@@ -71,18 +71,21 @@ public class SimpleFuture {
public void reset() {
triggerFuture = new CompletableFuture<>();
callbackFuture = triggerFuture;
- whenCompleted(() -> {});
callbackMap.keySet().forEach(this::whenCompleted);
}
/**
* This method will trigger all callback functions in the same thread and
wait until all callback
- * functions are completed. If throws exception, completedFuture is not done.
+ * functions are completed. If throws exception, completedFuture is not
done. Pay attention that
+ * reset could be called during whenCompleted runnable, should ignore waiting
*/
public void complete() {
try {
+ CompletableFuture<?> originalFuture = triggerFuture;
if (triggerFuture.complete(null)) {
- callbackFuture.get();
+ if (triggerFuture == originalFuture) {
+ callbackFuture.get();
+ }
}
} catch (Throwable throwable) {
throw normalize(throwable);
diff --git
a/amoro-common/src/main/java/org/apache/amoro/process/TableProcessState.java
b/amoro-common/src/main/java/org/apache/amoro/process/TableProcessState.java
index c58a6f4d0..335ab82b5 100644
--- a/amoro-common/src/main/java/org/apache/amoro/process/TableProcessState.java
+++ b/amoro-common/src/main/java/org/apache/amoro/process/TableProcessState.java
@@ -126,6 +126,10 @@ public class TableProcessState implements ProcessState {
this.endTime = System.currentTimeMillis();
}
+ public ProcessStage getStage() {
+ return status.toStage();
+ }
+
protected void setId(long processId) {
this.id = processId;
}
diff --git
a/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/RewriteStageTask.java
b/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/RewriteStageTask.java
index 363b1042e..60c5fba77 100644
---
a/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/RewriteStageTask.java
+++
b/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/RewriteStageTask.java
@@ -95,7 +95,7 @@ public class RewriteStageTask
}
private static long getRecordCnt(ContentFile<?>[] contentFiles) {
- int recordCnt = 0;
+ long recordCnt = 0;
if (contentFiles != null) {
for (ContentFile<?> contentFile : contentFiles) {
recordCnt += contentFile.recordCount();
