Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[AMORO-3365]The table is always in committing state #3366

Merged
merged 6 commits into from
Jan 23, 2025
Merged

[AMORO-3365]The table is always in committing state #3366

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
76d7d78
[AMORO-3365]The table is always in committing state
7hong Dec 16, 2024
3fcc6ac
Merge branch 'master' into master
klion26 Dec 20, 2024
8708028
Close the committing process on start
7hong Jan 22, 2025
1da7a8e
Merge branch 'master' of https://github.com/7hong/amoro
7hong Jan 22, 2025
aef1178
Merge branch 'master' into master
7hong Jan 22, 2025
008763a
Fix unit tests
7hong Jan 23, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 25 additions & 4 deletions amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import org.apache.amoro.ServerTableIdentifier;
import org.apache.amoro.api.OptimizingTaskId;
import org.apache.amoro.exception.OptimizingClosedException;
import org.apache.amoro.exception.PersistenceException;
import org.apache.amoro.optimizing.MetricsSummary;
import org.apache.amoro.optimizing.OptimizingType;
import org.apache.amoro.optimizing.RewriteFilesInput;
Expand Down Expand Up @@ -120,9 +121,20 @@ private void initTableRuntime(TableRuntime tableRuntime) {
if (tableRuntime.isOptimizingEnabled()) {
tableRuntime.resetTaskQuotas(
System.currentTimeMillis() - AmoroServiceConstants.QUOTA_LOOK_BACK_TIME);
// Close the committing process to avoid duplicate commit on the table.
if (tableRuntime.getOptimizingStatus() == OptimizingStatus.COMMITTING) {
OptimizingProcess process = tableRuntime.getOptimizingProcess();
if (process != null) {
LOG.warn(
"Close the committing process {} on table {}",
process.getProcessId(),
tableRuntime.getTableIdentifier());
process.close();
}
}
if (!tableRuntime.getOptimizingStatus().isProcessing()) {
scheduler.addTable(tableRuntime);
} else if (tableRuntime.getOptimizingStatus() != OptimizingStatus.COMMITTING) {
} else {
tableQueue.offer(new TableOptimizingProcess(tableRuntime));
}
} else {
Expand Down Expand Up @@ -569,6 +581,10 @@ public void commit() {
try {
if (hasCommitted) {
LOG.warn("{} has already committed, give up", tableRuntime.getTableIdentifier());
try {
persistAndSetCompleted(status == ProcessStatus.SUCCESS);
} catch (Exception ignored) {
}
throw new IllegalStateException("repeat commit, and last error " + failedReason);
}
try {
Expand All @@ -577,10 +593,15 @@ public void commit() {
status = ProcessStatus.SUCCESS;
endTime = System.currentTimeMillis();
persistAndSetCompleted(true);
} catch (Exception e) {
LOG.error("{} Commit optimizing failed ", tableRuntime.getTableIdentifier(), e);
} catch (PersistenceException e) {
LOG.warn(
"{} failed to persist process completed, will retry next commit",
tableRuntime.getTableIdentifier(),
e);
} catch (Throwable t) {
LOG.error("{} Commit optimizing failed ", tableRuntime.getTableIdentifier(), t);
status = ProcessStatus.FAILED;
failedReason = ExceptionUtil.getErrorMessage(e, 4000);
failedReason = ExceptionUtil.getErrorMessage(t, 4000);
endTime = System.currentTimeMillis();
persistAndSetCompleted(false);
}
Expand Down
8 changes: 6 additions & 2 deletions amoro-ams/src/test/java/org/apache/amoro/server/TestDefaultOptimizingService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -325,8 +325,12 @@ public void testReloadCompletedTask() {
optimizingService().completeTask(token, buildOptimizingTaskResult(task.getTaskId()));

reload();
assertTaskCompleted(null);
Assertions.assertNull(optimizingService().pollTask(token, THREAD_ID));
// Committing process will be closed when reloading
Assertions.assertNull(
tableService().getRuntime(serverTableIdentifier().getId()).getOptimizingProcess());
Assertions.assertEquals(
OptimizingStatus.IDLE,
tableService().getRuntime(serverTableIdentifier().getId()).getOptimizingStatus());
}

@Test
Expand Down
Loading