This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new cef210d25 [core] Add commit max retries if commit failed (#4165)
cef210d25 is described below
commit cef210d25cf15f58005a7fec5f774f3856a9a67c
Author: askwang <[email protected]>
AuthorDate: Wed Sep 11 11:19:56 2024 +0800
[core] Add commit max retries if commit failed (#4165)
---
.../shortcodes/generated/core_configuration.html | 6 +++++
.../main/java/org/apache/paimon/CoreOptions.java | 10 ++++++++
.../java/org/apache/paimon/AbstractFileStore.java | 3 ++-
.../paimon/operation/FileStoreCommitImpl.java | 28 +++++++++++++++++++---
4 files changed, 43 insertions(+), 4 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/core_configuration.html
b/docs/layouts/shortcodes/generated/core_configuration.html
index 272c31dc7..6dfd125d8 100644
--- a/docs/layouts/shortcodes/generated/core_configuration.html
+++ b/docs/layouts/shortcodes/generated/core_configuration.html
@@ -116,6 +116,12 @@ under the License.
<td>Boolean</td>
<td>Whether to force create snapshot on commit.</td>
</tr>
+ <tr>
+ <td><h5>commit.max-retries</h5></td>
+ <td style="word-wrap: break-word;">10</td>
+ <td>Integer</td>
+ <td>Maximum number of retries when commit failed.</td>
+ </tr>
<tr>
<td><h5>commit.user-prefix</h5></td>
<td style="word-wrap: break-word;">(none)</td>
diff --git a/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
b/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
index 7b8a35b58..85d732787 100644
--- a/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
@@ -493,6 +493,12 @@ public class CoreOptions implements Serializable {
.defaultValue(false)
.withDescription("Whether to force a compaction before
commit.");
+ public static final ConfigOption<Integer> COMMIT_MAX_RETRIES =
+ key("commit.max-retries")
+ .intType()
+ .defaultValue(10)
+ .withDescription("Maximum number of retries when commit
failed.");
+
public static final ConfigOption<Integer>
COMPACTION_MAX_SIZE_AMPLIFICATION_PERCENT =
key("compaction.max-size-amplification-percent")
.intType()
@@ -1736,6 +1742,10 @@ public class CoreOptions implements Serializable {
return options.get(COMMIT_FORCE_COMPACT);
}
+ public int commitMaxRetries() {
+ return options.get(COMMIT_MAX_RETRIES);
+ }
+
public int maxSizeAmplificationPercent() {
return options.get(COMPACTION_MAX_SIZE_AMPLIFICATION_PERCENT);
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
index 7c10a1aa7..c12d95e5c 100644
--- a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
+++ b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
@@ -218,7 +218,8 @@ abstract class AbstractFileStore<T> implements FileStore<T>
{
newStatsFileHandler(),
bucketMode(),
options.scanManifestParallelism(),
- callbacks);
+ callbacks,
+ options.commitMaxRetries());
}
@Override
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
index d2034d71c..0b19cae17 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
@@ -131,6 +131,7 @@ public class FileStoreCommitImpl implements FileStoreCommit
{
private final List<CommitCallback> commitCallbacks;
private final StatsFileHandler statsFileHandler;
private final BucketMode bucketMode;
+ private final int commitMaxRetries;
@Nullable private Lock lock;
private boolean ignoreEmptyCommit;
@@ -159,7 +160,8 @@ public class FileStoreCommitImpl implements FileStoreCommit
{
StatsFileHandler statsFileHandler,
BucketMode bucketMode,
@Nullable Integer manifestReadParallelism,
- List<CommitCallback> commitCallbacks) {
+ List<CommitCallback> commitCallbacks,
+ int commitMaxRetries) {
this.fileIO = fileIO;
this.schemaManager = schemaManager;
this.commitUser = commitUser;
@@ -180,6 +182,7 @@ public class FileStoreCommitImpl implements FileStoreCommit
{
this.branchName = branchName;
this.manifestReadParallelism = manifestReadParallelism;
this.commitCallbacks = commitCallbacks;
+ this.commitMaxRetries = commitMaxRetries;
this.lock = null;
this.ignoreEmptyCommit = true;
@@ -712,6 +715,12 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
while (true) {
Snapshot latestSnapshot = snapshotManager.latestSnapshot();
cnt++;
+ if (cnt >= commitMaxRetries) {
+ throw new RuntimeException(
+ String.format(
+ "Commit failed after %s attempts, there maybe
exist commit conflicts between multiple jobs.",
+ commitMaxRetries));
+ }
if (tryCommitOnce(
tableFiles,
changelogFiles,
@@ -742,6 +751,12 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
Snapshot latestSnapshot = snapshotManager.latestSnapshot();
cnt++;
+ if (cnt >= commitMaxRetries) {
+ throw new RuntimeException(
+ String.format(
+ "Commit failed after %s attempts, there maybe
exist commit conflicts between multiple jobs.",
+ commitMaxRetries));
+ }
List<ManifestEntry> changesWithOverwrite = new ArrayList<>();
List<IndexManifestEntry> indexChangesWithOverwrite = new
ArrayList<>();
if (latestSnapshot != null) {
@@ -805,6 +820,7 @@ public class FileStoreCommitImpl implements FileStoreCommit
{
ConflictCheck conflictCheck,
String branchName,
@Nullable String newStatsFileName) {
+ long startMillis = System.currentTimeMillis();
long newSnapshotId =
latestSnapshot == null ? Snapshot.FIRST_SNAPSHOT_ID :
latestSnapshot.id() + 1;
Path newSnapshotPath =
@@ -1012,12 +1028,18 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
}
// atomic rename fails, clean up and try again
+ long commitTime = (System.currentTimeMillis() - startMillis) / 1000;
LOG.warn(
String.format(
"Atomic commit failed for snapshot #%d (path %s) by
user %s "
- + "with identifier %s and kind %s. "
+ + "with identifier %s and kind %s after %s
seconds. "
+ "Clean up and try again.",
- newSnapshotId, newSnapshotPath, commitUser,
identifier, commitKind.name()));
+ newSnapshotId,
+ newSnapshotPath,
+ commitUser,
+ identifier,
+ commitKind.name(),
+ commitTime));
cleanUpTmpManifests(
previousChangesListName,
newChangesListName,
