diff --git a/docs/docs/concepts/concurrency-control.md b/docs/docs/concepts/concurrency-control.md index 25f5fbe62feb..b3d641a4c4ac 100644 --- a/docs/docs/concepts/concurrency-control.md +++ b/docs/docs/concepts/concurrency-control.md @@ -101,6 +101,58 @@ which is stopped when the scope ends, so blocked renewals do not occupy other sc threads. Filesystem locks still depend on the storage's atomic publication behavior; they do not provide a storage fencing token. +## REST Catalog Commit Leases + +Catalog-managed Paimon tables in the REST catalog expose table-bound commit leases through +`Catalog.createLock` and the unified +`Lock.acquire` / `Lock.runWithLock` contract. Set the persisted table property +`rest.commit.lock-enabled = true` to enable server-side lease enforcement. Independently, set +`rest.commit.acquire-lock = true` in a writer's dynamic options to acquire a lease for each +publication attempt. Both options default to `false`. Enabling the table capability alone does +not make clients acquire leases. Requesting acquisition without the table capability fails. +External tables keep filesystem publication. +Callers supply their exact `commitUser` and table UUID. +Leases require the target branch's catalog-resolved UUID. Reload a view switched to another branch +from the catalog before using leases. +Lock contention waits until `commit.timeout` when configured, independently of publication retries. +Scopes share a timer and a cached worker pool. The timer dispatches RPCs to workers so a blocked +renewal cannot block other scopes' scheduling. Each scope permits only one in-flight renewal and +cancels its own scheduled and in-flight renewal tasks when it ends, without stopping the shared +executors. Timer and worker threads exit after one minute of inactivity and restart when needed. +The timer stays alive while renewal tasks remain scheduled. The client validates the lease's +monotonic deadline before publishing. A failed operation stops renewal and its lease expires +automatically. The server keeps the lease duration fixed for each scope; duration changes take effect +on the next acquisition. + +The server must validate the current table incarnation, branch, authenticated caller and exact +owner, and atomically publish the snapshot and clear the lease. All snapshot writers must honor +active leases, including writers that do not request one. Each successful snapshot publication +ends its lease; a later publication must acquire another scope. Snapshot comparison and file +conflict validation remain necessary. + +For a table with lease enforcement disabled, the server can skip lease storage access entirely. +For a table with enforcement enabled, even writers with `rest.commit.acquire-lock = false` must +respect another writer's active lease. The server uses its own persisted table property for this +decision, rather than client options. Capability changes must be serialized with publication and +lease acquisition; disabling enforcement must not bypass an active lease. + +Automatic conflict rollback always supplies the runtime branch and writer's `commitUser`, together +with the expected head's `fromSnapshot` and `fromSnapshotUuid`. Snapshot IDs can be reused after a +rollback, so the server must check both fields against the current head inside the mutation +transaction, before changing partitions or deleting files. Automatic rollback skips legacy snapshots +without a UUID. A known table or branch UUID is supplied independently of writer identity. + +`commitUser` identifies the writer; it does not require a lease to exist. If a live lease exists, the +writer and authenticated caller must match its owner. Without a live lease, conditional optimistic +rollback is allowed. The server must validate admission and the expected head, roll back and clear +the lease atomically. Manual rollback can omit writer identity, but cannot change a leased head. +The table UUID is omitted only when the target table or branch identity is unknown. + +This is a commit admission lease, not a generation or fencing token. Requests with the same +commit user and authenticated caller share ownership. Use a unique commit user per logical writer +and preserve it during recovery. A server without the lease protocol rejects acquisition rather +than silently proceeding without protection. + ## Files conflict A writer validates file-level changes as well as the snapshot ID. For example, if two compactors diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html index a1c67478ac1d..4c30f37c5a87 100644 --- a/docs/generated/core_configuration.html +++ b/docs/generated/core_configuration.html @@ -1470,6 +1470,18 @@ String Time field for record level expire. It supports the following types: `timestamps in seconds with INT`,`timestamps in seconds with BIGINT`, `timestamps in milliseconds with BIGINT` or `timestamp`. + +
rest.commit.acquire-lock
+ false + Boolean + Whether this writer acquires a REST commit lease for each publication attempt. Can be set dynamically for a writer and requires rest.commit.lock-enabled on the server-side table. + + +
rest.commit.lock-enabled
+ false + Boolean + Enable the REST catalog's server-side table commit lease capability. Writers acquire a lease only when rest.commit.acquire-lock is enabled. +
row-tracking.enabled
false diff --git a/docs/static/rest-catalog-open-api.yaml b/docs/static/rest-catalog-open-api.yaml index 559a43302411..276da1ce8261 100644 --- a/docs/static/rest-catalog-open-api.yaml +++ b/docs/static/rest-catalog-open-api.yaml @@ -595,6 +595,12 @@ paths: required: true schema: type: string + description: >- + Validate the table incarnation, branch and expected head atomically with publication. + Every writer must honor an active commit lease, including writers that do not request + one. The authenticated caller and snapshot.commitUser must match the active owner. + Clear the lease only when a new snapshot is successfully published; rejected requests + and idempotent replays of an existing snapshot must not clear a newer lease. requestBody: content: application/json: @@ -615,6 +621,112 @@ paths: $ref: '#/components/responses/TableNotExistErrorResponse' "500": $ref: '#/components/responses/ServerErrorResponse' + /v1/{prefix}/databases/{database}/tables/{table}/commit-lock: + post: + tags: + - table + summary: Acquire a table commit lease + operationId: acquireCommitLock + parameters: + - name: prefix + in: path + required: true + schema: + type: string + - name: database + in: path + required: true + schema: + type: string + - name: table + in: path + required: true + schema: + type: string + description: >- + Requires the server-side table property rest.commit.lock-enabled=true and the same + permission as committing snapshots. Clients opt into acquisition separately using + rest.commit.acquire-lock=true; that client option does not enable the table capability. + Ownership is the exact commitUser together with the authenticated caller. Contention + returns HTTP 200 with acquired=false. A grant includes the authoritative snapshot head. + Renewal never revives an expired lease. Publication retains snapshot CAS validation + and atomically releases the lease on a new successful commit. A rejected publication + does not release the lease. A replay of an already-published snapshot must not release + a newer lease. The lease duration is fixed for each scope. There is no unlock RPC. + requestBody: + required: true + content: + application/json: + schema: + $ref: '#/components/schemas/CommitLockRequest' + responses: + "200": + description: OK + content: + application/json: + schema: + $ref: '#/components/schemas/CommitLockResponse' + "400": + $ref: '#/components/responses/BadRequestErrorResponse' + "401": + $ref: '#/components/responses/UnauthorizedErrorResponse' + "403": + $ref: '#/components/responses/ForbiddenErrorResponse' + "404": + $ref: '#/components/responses/TableNotExistErrorResponse' + "500": + $ref: '#/components/responses/ServerErrorResponse' + /v1/{prefix}/databases/{database}/tables/{table}/commit-lock/renew: + post: + tags: + - table + summary: Renew a table commit lease + operationId: renewCommitLock + parameters: + - name: prefix + in: path + required: true + schema: + type: string + - name: database + in: path + required: true + schema: + type: string + - name: table + in: path + required: true + schema: + type: string + description: >- + Requires the same permission as committing snapshots. Validate the table incarnation, + branch, authenticated caller and exact commitUser against the unexpired lease. + Return acquired=false when ownership is lost or the lease has expired; renewal must + never acquire a new lease or revive an expired one. Keep leaseMillis fixed within + the scope. Each successful new snapshot publication ends its lease. + requestBody: + required: true + content: + application/json: + schema: + $ref: '#/components/schemas/CommitLockRequest' + responses: + "200": + description: OK + content: + application/json: + schema: + $ref: '#/components/schemas/CommitLockResponse' + "400": + $ref: '#/components/responses/BadRequestErrorResponse' + "401": + $ref: '#/components/responses/UnauthorizedErrorResponse' + "403": + $ref: '#/components/responses/ForbiddenErrorResponse' + "404": + $ref: '#/components/responses/TableNotExistErrorResponse' + "500": + $ref: '#/components/responses/ServerErrorResponse' /v1/{prefix}/databases/{database}/tables/{table}/rollback: post: tags: @@ -3499,6 +3611,39 @@ components: $ref: '#/components/schemas/Identifier' destination: $ref: '#/components/schemas/Identifier' + CommitLockRequest: + type: object + required: [tableId, commitUser] + properties: + tableId: + type: string + description: Stable resolved table or branch UUID; a recreated table cannot inherit a lease. + commitUser: + type: string + minLength: 1 + maxLength: 255 + description: Exact commitUser that will appear in the submitted snapshot. + CommitLockResponse: + type: object + required: [acquired, expiresAtMillis, leaseMillis] + properties: + acquired: + type: boolean + commitUser: + type: [string, "null"] + expiresAtMillis: + type: integer + format: int64 + description: Server database clock expiration; zero when acquired is false. + leaseMillis: + type: integer + format: int64 + description: Lease duration in milliseconds, fixed within a scope; zero when acquired is false. + snapshot: + anyOf: + - $ref: '#/components/schemas/Snapshot' + - type: "null" + description: Authoritative head at grant time; null for an empty table. CommitTableRequest: type: object properties: @@ -3516,21 +3661,32 @@ components: type: object description: >- tableId identifies the expected table incarnation independently of lock ownership. - For rollback within a writer's commit lock, also supply commitUser and fromSnapshot so - the server can validate the expected head and owner. Omit commitUser when no writer - lock is used, and omit tableId only when its UUID is unavailable. + Writer rollback supplies commitUser, fromSnapshot and fromSnapshotUuid regardless of + whether the writer acquired a lease. commitUser identifies the writer, not a requirement + for a lease to exist. A live lease requires the exact owner and authenticated caller; + without a live lease, conditional optimistic rollback is allowed. Snapshot IDs can be + reused after rollback, so both expected head fields must be checked against the current + head inside the mutation transaction before partition changes or file cleanup. The server + must validate table incarnation, admission and expected head, then roll back and clear + the lease atomically. Manual rollback may omit writer identity, but cannot change a + leased head. Omit tableId only when its UUID is unavailable. properties: instant: $ref: '#/components/schemas/Instant' fromSnapshot: type: [ integer, "null" ] format: int64 + description: Expected latest snapshot ID, required for writer rollback. + fromSnapshotUuid: + type: [ string, "null" ] + minLength: 1 + description: Expected latest snapshot UUID; requires fromSnapshot and is required for writer rollback. tableId: type: [ string, "null" ] description: Expected table or branch UUID, independently of commitUser. commitUser: type: [ string, "null" ] - description: Exact owner of the writer's lock. + description: Writer identity for admission against any active lease; requires both expected head fields. RollbackSchemaRequest: type: object required: diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java index 98f0dfe048e6..d44370d760fc 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java @@ -973,6 +973,23 @@ public InlineElement getDescription() { "Specify how to initialize the next sequence number for primary key " + "table writers."); + public static final ConfigOption REST_COMMIT_LOCK_ENABLED = + key("rest.commit.lock-enabled") + .booleanType() + .defaultValue(false) + .withDescription( + "Enable the REST catalog's server-side table commit lease capability. " + + "Writers acquire a lease only when rest.commit.acquire-lock is enabled."); + + public static final ConfigOption REST_COMMIT_ACQUIRE_LOCK = + key("rest.commit.acquire-lock") + .booleanType() + .defaultValue(false) + .withDescription( + "Whether this writer acquires a REST commit lease for each publication " + + "attempt. Can be set dynamically for a writer and requires " + + "rest.commit.lock-enabled on the server-side table."); + public static final ConfigOption COMMIT_TIMEOUT = key("commit.timeout") .durationType() @@ -3900,6 +3917,14 @@ public SequenceNumberInitMode writeSequenceNumberInitMode() { return options.get(WRITE_SEQUENCE_NUMBER_INIT_MODE); } + public boolean restCommitLockEnabled() { + return options.get(REST_COMMIT_LOCK_ENABLED); + } + + public boolean restCommitAcquireLock() { + return options.get(REST_COMMIT_ACQUIRE_LOCK); + } + public long commitTimeout() { return options.get(COMMIT_TIMEOUT) == null ? Long.MAX_VALUE diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java b/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java index c454f2634170..f97425a16b89 100644 --- a/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java +++ b/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java @@ -45,6 +45,7 @@ import org.apache.paimon.rest.requests.AlterTableRequest; import org.apache.paimon.rest.requests.AlterViewRequest; import org.apache.paimon.rest.requests.AuthTableQueryRequest; +import org.apache.paimon.rest.requests.CommitLockRequest; import org.apache.paimon.rest.requests.CommitTableRequest; import org.apache.paimon.rest.requests.CreateBranchRequest; import org.apache.paimon.rest.requests.CreateDatabaseRequest; @@ -72,6 +73,7 @@ import org.apache.paimon.rest.requests.UpsertSemanticViewRequest; import org.apache.paimon.rest.responses.AlterDatabaseResponse; import org.apache.paimon.rest.responses.AuthTableQueryResponse; +import org.apache.paimon.rest.responses.CommitLockResponse; import org.apache.paimon.rest.responses.CommitTableResponse; import org.apache.paimon.rest.responses.ConfigResponse; import org.apache.paimon.rest.responses.CreatePartitionsResponse; @@ -734,6 +736,25 @@ public boolean commitSnapshot( return response.isSuccess(); } + public CommitLockResponse acquireCommitLock( + Identifier identifier, String tableUuid, String commitUser) { + return client.post( + resourcePaths.commitLock(identifier.getDatabaseName(), identifier.getObjectName()), + new CommitLockRequest(tableUuid, commitUser), + CommitLockResponse.class, + restAuthFunction); + } + + public CommitLockResponse renewCommitLock( + Identifier identifier, String tableUuid, String commitUser) { + return client.post( + resourcePaths.commitLock(identifier.getDatabaseName(), identifier.getObjectName()) + + "/renew", + new CommitLockRequest(tableUuid, commitUser), + CommitLockResponse.class, + restAuthFunction); + } + /** * Rollback instant for table. * @@ -742,7 +763,8 @@ public boolean commitSnapshot( * @param instant instant to rollback * @param fromSnapshot snapshot from, success only occurs when the latest snapshot is this * snapshot. - * @param commitUser owner of the writer lock, or null without a writer lock + * @param fromSnapshotUuid expected snapshot UUID, checked together with fromSnapshot + * @param commitUser writer identity for active-lock admission, or null without a writer * @throws NoSuchResourceException Exception thrown on HTTP 404 means the table or the snapshot * or the tag not exists * @throws ForbiddenException Exception thrown on HTTP 403 means don't have the permission for @@ -753,9 +775,11 @@ public void rollbackTo( @Nullable String tableUuid, Instant instant, @Nullable Long fromSnapshot, + @Nullable String fromSnapshotUuid, @Nullable String commitUser) { RollbackTableRequest request = - new RollbackTableRequest(instant, fromSnapshot, tableUuid, commitUser); + new RollbackTableRequest( + instant, fromSnapshot, fromSnapshotUuid, tableUuid, commitUser); client.post( resourcePaths.rollbackTable( identifier.getDatabaseName(), identifier.getObjectName()), diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/ResourcePaths.java b/paimon-api/src/main/java/org/apache/paimon/rest/ResourcePaths.java index f3065708da9c..90ca5e9012cf 100644 --- a/paimon-api/src/main/java/org/apache/paimon/rest/ResourcePaths.java +++ b/paimon-api/src/main/java/org/apache/paimon/rest/ResourcePaths.java @@ -196,6 +196,10 @@ public String commitTable(String databaseName, String objectName) { "commit"); } + public String commitLock(String databaseName, String objectName) { + return table(databaseName, objectName) + "/commit-lock"; + } + public String rollbackTable(String databaseName, String objectName) { return SLASH.join( V1, diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/requests/CommitLockRequest.java b/paimon-api/src/main/java/org/apache/paimon/rest/requests/CommitLockRequest.java new file mode 100644 index 000000000000..9a12f8272cd3 --- /dev/null +++ b/paimon-api/src/main/java/org/apache/paimon/rest/requests/CommitLockRequest.java @@ -0,0 +1,55 @@ +/* + * 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.paimon.rest.requests; + +import org.apache.paimon.rest.RESTRequest; + +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty; + +import java.beans.ConstructorProperties; + +/** Request to acquire or renew a table commit lease owned by a writer. */ +@JsonIgnoreProperties(ignoreUnknown = true) +public class CommitLockRequest implements RESTRequest { + + @JsonProperty("tableId") + private final String tableId; + + @JsonProperty("commitUser") + private final String commitUser; + + @JsonCreator + @ConstructorProperties({"tableId", "commitUser"}) + public CommitLockRequest( + @JsonProperty("tableId") String tableId, + @JsonProperty("commitUser") String commitUser) { + this.tableId = tableId; + this.commitUser = commitUser; + } + + public String getTableId() { + return tableId; + } + + public String getCommitUser() { + return commitUser; + } +} diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/requests/RollbackTableRequest.java b/paimon-api/src/main/java/org/apache/paimon/rest/requests/RollbackTableRequest.java index 77fc97049c95..b76747ef2ad8 100644 --- a/paimon-api/src/main/java/org/apache/paimon/rest/requests/RollbackTableRequest.java +++ b/paimon-api/src/main/java/org/apache/paimon/rest/requests/RollbackTableRequest.java @@ -46,6 +46,11 @@ public class RollbackTableRequest implements RESTRequest { @Nullable private final Long fromSnapshot; + @JsonProperty("fromSnapshotUuid") + @JsonInclude(JsonInclude.Include.NON_NULL) + @Nullable + private final String fromSnapshotUuid; + @JsonProperty("tableId") @JsonInclude(JsonInclude.Include.NON_NULL) @Nullable @@ -57,14 +62,16 @@ public class RollbackTableRequest implements RESTRequest { private final String commitUser; @JsonCreator - @ConstructorProperties({"instant", "fromSnapshot", "tableId", "commitUser"}) + @ConstructorProperties({"instant", "fromSnapshot", "fromSnapshotUuid", "tableId", "commitUser"}) public RollbackTableRequest( @JsonProperty(FIELD_INSTANT) Instant instant, @JsonProperty(FIELD_FROM_SNAPSHOT) @Nullable Long fromSnapshot, + @JsonProperty("fromSnapshotUuid") @Nullable String fromSnapshotUuid, @JsonProperty("tableId") @Nullable String tableId, @JsonProperty("commitUser") @Nullable String commitUser) { this.instant = instant; this.fromSnapshot = fromSnapshot; + this.fromSnapshotUuid = fromSnapshotUuid; this.tableId = tableId; this.commitUser = commitUser; } @@ -91,4 +98,10 @@ public Instant getInstant() { public Long getFromSnapshot() { return fromSnapshot; } + + @JsonGetter("fromSnapshotUuid") + @Nullable + public String getFromSnapshotUuid() { + return fromSnapshotUuid; + } } diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/responses/CommitLockResponse.java b/paimon-api/src/main/java/org/apache/paimon/rest/responses/CommitLockResponse.java new file mode 100644 index 000000000000..94c2f45810c1 --- /dev/null +++ b/paimon-api/src/main/java/org/apache/paimon/rest/responses/CommitLockResponse.java @@ -0,0 +1,90 @@ +/* + * 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.paimon.rest.responses; + +import org.apache.paimon.Snapshot; +import org.apache.paimon.rest.RESTResponse; + +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty; + +import javax.annotation.Nullable; + +/** Result of acquiring or renewing a commit lease; a null snapshot denotes an empty table. */ +@JsonIgnoreProperties(ignoreUnknown = true) +public class CommitLockResponse implements RESTResponse { + + @JsonProperty("acquired") + private final boolean acquired; + + @JsonProperty("commitUser") + @Nullable + private final String commitUser; + + @JsonProperty("expiresAtMillis") + private final long expiresAtMillis; + + @JsonProperty("leaseMillis") + private final long leaseMillis; + + @JsonProperty("snapshot") + @Nullable + private final Snapshot snapshot; + + @JsonCreator + public CommitLockResponse( + @JsonProperty("acquired") boolean acquired, + @JsonProperty("commitUser") @Nullable String commitUser, + @JsonProperty("expiresAtMillis") long expiresAtMillis, + @JsonProperty("leaseMillis") long leaseMillis, + @JsonProperty("snapshot") @Nullable Snapshot snapshot) { + this.acquired = acquired; + this.commitUser = commitUser; + this.expiresAtMillis = expiresAtMillis; + this.leaseMillis = leaseMillis; + this.snapshot = snapshot; + } + + public static CommitLockResponse unavailable() { + return new CommitLockResponse(false, null, 0, 0, null); + } + + public boolean isAcquired() { + return acquired; + } + + @Nullable + public String getCommitUser() { + return commitUser; + } + + public long getExpiresAtMillis() { + return expiresAtMillis; + } + + public long getLeaseMillis() { + return leaseMillis; + } + + @Nullable + public Snapshot getSnapshot() { + return snapshot; + } +} diff --git a/paimon-api/src/test/java/org/apache/paimon/rest/requests/RequestJacksonCompatibilityTest.java b/paimon-api/src/test/java/org/apache/paimon/rest/requests/RequestJacksonCompatibilityTest.java index e7e8a5517e94..64dae0cb5ead 100644 --- a/paimon-api/src/test/java/org/apache/paimon/rest/requests/RequestJacksonCompatibilityTest.java +++ b/paimon-api/src/test/java/org/apache/paimon/rest/requests/RequestJacksonCompatibilityTest.java @@ -72,6 +72,15 @@ public class RequestJacksonCompatibilityTest { "{\"select\":[\"id\",\"name\"]}", request -> assertThat(request.select()).containsExactly("id", "name"), "select"), + requestCase( + CommitLockRequest.class, + "{\"tableId\":\"table-id\",\"commitUser\":\"morax-job\"}", + request -> { + assertThat(request.getTableId()).isEqualTo("table-id"); + assertThat(request.getCommitUser()).isEqualTo("morax-job"); + }, + "tableId", + "commitUser"), requestCase( CreateBranchRequest.class, "{\"branch\":\"audit\",\"fromTag\":\"v1\"}", 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 8dd60a760368..15ae1627ad12 100644 --- a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java +++ b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java @@ -307,7 +307,8 @@ public FileStoreCommitImpl newCommit(String commitUser, FileStoreTable table) { snapshotManager, scanner); CommitRollback rollback = null; - TableRollback tableRollback = catalogEnvironment.catalogTableRollback(null, null); + TableRollback tableRollback = + catalogEnvironment.catalogTableRollback(options.branch(), commitUser); if (tableRollback != null) { rollback = new CommitRollback(tableRollback); } diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java index d9503d43d93b..49eabacbf06d 100644 --- a/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java +++ b/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java @@ -717,6 +717,7 @@ public void rollbackTo( @Nullable String tableUuid, Instant instant, @Nullable Long fromSnapshot, + @Nullable String fromSnapshotUuid, @Nullable String commitUser) throws Catalog.TableNotExistException { throw new UnsupportedOperationException(); diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java index 02e318a13ee5..beb43b59d5e2 100644 --- a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java +++ b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java @@ -715,7 +715,7 @@ default boolean supportsListTableByType() { *
    *
  • {@link #commitSnapshot(Identifier, String, String, Snapshot, List)}. *
  • {@link #loadSnapshot(Identifier)}. - *
  • {@link #rollbackTo(Identifier, String, Instant, Long, String)}. + *
  • {@link #rollbackTo(Identifier, String, Instant, Long, String, String)}. *
  • {@link #createBranch(Identifier, String, String)}. *
  • {@link #dropBranch(Identifier, String)}. *
  • {@link #renameBranch(Identifier, String, String)}. @@ -847,7 +847,8 @@ default void resetConsumer( * @param instant like snapshotId or tagName * @param fromSnapshot snapshot from, success only occurs when the latest snapshot is this * snapshot - * @param commitUser owner of the writer's lock, or null when no writer lock is used + * @param fromSnapshotUuid expected head UUID; distinguishes snapshots that reuse the same ID + * @param commitUser writer identity for active-lock admission, or null when unavailable * @throws Catalog.TableNotExistException if the table does not exist * @throws UnsupportedOperationException if the catalog does not {@link * #supportsVersionManagement()} @@ -857,6 +858,7 @@ void rollbackTo( @Nullable String tableUuid, Instant instant, @Nullable Long fromSnapshot, + @Nullable String fromSnapshotUuid, @Nullable String commitUser) throws Catalog.TableNotExistException; diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java index 89159783fbf2..1aeeff926d07 100644 --- a/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java +++ b/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java @@ -254,9 +254,11 @@ public void rollbackTo( @Nullable String tableUuid, Instant instant, @Nullable Long fromSnapshot, + @Nullable String fromSnapshotUuid, @Nullable String commitUser) throws Catalog.TableNotExistException { - wrapped.rollbackTo(identifier, tableUuid, instant, fromSnapshot, commitUser); + wrapped.rollbackTo( + identifier, tableUuid, instant, fromSnapshot, fromSnapshotUuid, commitUser); } @Override diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/TableRollback.java b/paimon-core/src/main/java/org/apache/paimon/catalog/TableRollback.java index 5cb2a521924d..bbe16ec3bbec 100644 --- a/paimon-core/src/main/java/org/apache/paimon/catalog/TableRollback.java +++ b/paimon-core/src/main/java/org/apache/paimon/catalog/TableRollback.java @@ -25,5 +25,6 @@ /** Rollback table to instant from snapshot. */ public interface TableRollback { - void rollbackTo(Instant instant, @Nullable Long fromSnapshot); + void rollbackTo( + Instant instant, @Nullable Long fromSnapshot, @Nullable String fromSnapshotUuid); } diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/CommitRollback.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/CommitRollback.java index 683b6555a651..ead30bc8f878 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/CommitRollback.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/commit/CommitRollback.java @@ -32,10 +32,12 @@ public CommitRollback(TableRollback tableRollback) { } public boolean tryToRollback(Snapshot latestSnapshot) { - if (latestSnapshot.commitKind() == Snapshot.CommitKind.COMPACT) { + if (latestSnapshot.commitKind() == Snapshot.CommitKind.COMPACT + && latestSnapshot.uuid() != null) { long latest = latestSnapshot.id(); try { - tableRollback.rollbackTo(Instant.snapshot(latest - 1), latest); + tableRollback.rollbackTo( + Instant.snapshot(latest - 1), latest, latestSnapshot.uuid()); return true; } catch (Exception ignored) { } diff --git a/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java b/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java index 56939c3a6183..07e4a4ae18f8 100644 --- a/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java +++ b/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java @@ -26,6 +26,7 @@ import org.apache.paimon.annotation.VisibleForTesting; import org.apache.paimon.catalog.Catalog; import org.apache.paimon.catalog.CatalogContext; +import org.apache.paimon.catalog.CatalogLockContext; import org.apache.paimon.catalog.CatalogUtils; import org.apache.paimon.catalog.Database; import org.apache.paimon.catalog.Identifier; @@ -44,6 +45,7 @@ import org.apache.paimon.management.PermissionManagement; import org.apache.paimon.management.PolicyManagement; import org.apache.paimon.management.SemanticViewManagement; +import org.apache.paimon.operation.Lock; import org.apache.paimon.options.Options; import org.apache.paimon.partition.Partition; import org.apache.paimon.partition.PartitionStatistics; @@ -402,6 +404,21 @@ public Table getTable(Identifier identifier) throws TableNotExistException { true); } + @Override + public Lock createLock( + Identifier identifier, + @Nullable String tableUuid, + @Nullable String commitUser, + Options tableOptions) { + return new RESTCatalogLockFactory() + .createLock( + CatalogLockContext.fromOptions(context.options()), + identifier, + tableUuid, + commitUser, + tableOptions); + } + @Override public Optional loadSnapshot(Identifier identifier) throws TableNotExistException { @@ -545,10 +562,12 @@ public void rollbackTo( @Nullable String tableUuid, Instant instant, @Nullable Long fromSnapshot, + @Nullable String fromSnapshotUuid, @Nullable String commitUser) throws Catalog.TableNotExistException { try { - api.rollbackTo(identifier, tableUuid, instant, fromSnapshot, commitUser); + api.rollbackTo( + identifier, tableUuid, instant, fromSnapshot, fromSnapshotUuid, commitUser); } catch (NoSuchResourceException e) { if (StringUtils.equals(e.resourceType(), ErrorResponse.RESOURCE_TYPE_SNAPSHOT)) { throw new IllegalArgumentException( diff --git a/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalogLock.java b/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalogLock.java new file mode 100644 index 000000000000..581881048462 --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalogLock.java @@ -0,0 +1,279 @@ +/* + * 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.paimon.rest; + +import org.apache.paimon.catalog.Catalog; +import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.operation.Lock; +import org.apache.paimon.rest.exceptions.BadRequestException; +import org.apache.paimon.rest.exceptions.ForbiddenException; +import org.apache.paimon.rest.exceptions.NoSuchResourceException; +import org.apache.paimon.rest.responses.CommitLockResponse; +import org.apache.paimon.utils.ExecutorThreadFactory; + +import javax.annotation.Nullable; + +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.FutureTask; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.LongSupplier; + +import static org.apache.paimon.utils.Preconditions.checkArgument; +import static org.apache.paimon.utils.Preconditions.checkNotNull; + +/** A renewable REST commit lease bound to a table, branch and writer. */ +public class RESTCatalogLock implements Lock { + + private static final ScheduledExecutorService RENEWAL_SCHEDULER = createRenewalScheduler(); + private static final ExecutorService RENEWAL_WORKERS = + Executors.newCachedThreadPool(new ExecutorThreadFactory("rest-commit-lock-renew")); + + private final RESTApi api; + private final Identifier identifier; + private final String tableUuid; + private final String commitUser; + private final ScheduledExecutorService scheduler; + private final ExecutorService workers; + private final LongSupplier nanoTime; + private final long acquireTimeoutMillis; + private final ThreadLocal currentLease = new ThreadLocal<>(); + + public RESTCatalogLock( + RESTApi api, + Identifier identifier, + String tableUuid, + String commitUser, + long acquireTimeoutMillis) { + this( + api, + identifier, + tableUuid, + commitUser, + acquireTimeoutMillis, + RENEWAL_SCHEDULER, + RENEWAL_WORKERS, + System::nanoTime); + } + + RESTCatalogLock( + RESTApi api, + Identifier identifier, + String tableUuid, + String commitUser, + long acquireTimeoutMillis, + ScheduledExecutorService scheduler, + ExecutorService workers, + LongSupplier nanoTime) { + checkArgument( + tableUuid != null && !tableUuid.isEmpty(), "A commit lease requires a table UUID."); + checkArgument( + commitUser != null && !commitUser.isEmpty(), + "A commit lease requires a commit user."); + checkArgument( + acquireTimeoutMillis >= 0, "The lock acquisition timeout cannot be negative."); + this.api = checkNotNull(api); + this.identifier = checkNotNull(identifier); + this.tableUuid = tableUuid; + this.commitUser = commitUser; + this.acquireTimeoutMillis = acquireTimeoutMillis; + this.scheduler = checkNotNull(scheduler); + this.workers = checkNotNull(workers); + this.nanoTime = nanoTime; + } + + static ScheduledThreadPoolExecutor createRenewalScheduler() { + ScheduledThreadPoolExecutor executor = + new ScheduledThreadPoolExecutor( + 1, new ExecutorThreadFactory("rest-commit-lock-timer")); + executor.setRemoveOnCancelPolicy(true); + executor.setKeepAliveTime(1, TimeUnit.MINUTES); + executor.allowCoreThreadTimeOut(true); + return executor; + } + + @Override + public AutoCloseable acquire() throws Exception { + checkArgument(currentLease.get() == null, "This thread already holds a commit lease."); + long startedAt = nanoTime.getAsLong(); + while (true) { + long requestedAt = nanoTime.getAsLong(); + CommitLockResponse response = acquireLease(); + if (response.isAcquired()) { + validateGrant(response); + Lease lease = new Lease(response.getLeaseMillis(), requestedAt); + try { + lease.checkValid(); + currentLease.set(lease); + return () -> { + if (currentLease.get() == lease) { + currentLease.remove(); + } + lease.close(); + }; + } catch (Exception | Error e) { + Lock.release(lease, e); + throw e; + } + } + if (TimeUnit.NANOSECONDS.toMillis(nanoTime.getAsLong() - startedAt) + >= acquireTimeoutMillis) { + throw new IllegalStateException( + "Timed out acquiring commit lock for " + identifier); + } + Thread.sleep(Math.min(100, acquireTimeoutMillis)); + } + } + + private CommitLockResponse acquireLease() throws Catalog.TableNotExistException { + try { + return api.acquireCommitLock(identifier, tableUuid, commitUser); + } catch (NoSuchResourceException e) { + throw new Catalog.TableNotExistException(identifier, e); + } catch (ForbiddenException e) { + throw new Catalog.TableNoPermissionException(identifier, e); + } catch (BadRequestException e) { + throw new IllegalArgumentException(e.getMessage(), e); + } + } + + private void validateGrant(CommitLockResponse response) { + if (!commitUser.equals(response.getCommitUser()) || response.getLeaseMillis() <= 0) { + throw new IllegalStateException("The server returned an invalid commit lease."); + } + } + + @Override + public void ensureValid() { + Lease lease = currentLease.get(); + if (lease == null) { + throw new IllegalStateException("No commit lease is held by this thread."); + } + lease.checkValid(); + } + + private class Lease implements AutoCloseable { + private final ScheduledFuture renewal; + private final AtomicBoolean closed = new AtomicBoolean(); + private final AtomicBoolean renewing = new AtomicBoolean(); + private final long leaseMillis; + private volatile long renewedAt; + @Nullable private volatile Throwable renewalFailure; + @Nullable private volatile FutureTask renewalRequest; + + private Lease(long leaseMillis, long requestedAt) { + this.leaseMillis = leaseMillis; + this.renewedAt = requestedAt; + long interval = Math.max(1, leaseMillis / 3); + this.renewal = + scheduler.scheduleWithFixedDelay( + this::submitRenewal, interval, interval, TimeUnit.MILLISECONDS); + } + + private void submitRenewal() { + if (closed.get() || renewalFailure != null || !renewing.compareAndSet(false, true)) { + return; + } + FutureTask request = + new FutureTask<>( + () -> { + try { + renew(); + } finally { + renewing.set(false); + } + }, + null); + renewalRequest = request; + // Closing can race with dispatch. Publish the request before rechecking closed so + // either close or this dispatcher cancels it before it can start another RPC. + if (closed.get()) { + request.cancel(true); + renewing.set(false); + return; + } + try { + workers.execute(request); + } catch (RuntimeException | Error e) { + renewalFailure = e; + request.cancel(false); + renewing.set(false); + } + } + + private void renew() { + if (closed.get() || renewalFailure != null) { + return; + } + try { + // Renewal workers validate this lease without the acquiring thread's context. + checkValid(); + long requestedAt = nanoTime.getAsLong(); + CommitLockResponse response = + api.renewCommitLock(identifier, tableUuid, commitUser); + if (!response.isAcquired()) { + throw new IllegalStateException("The commit lease has expired."); + } + validateGrant(response); + if (response.getLeaseMillis() != leaseMillis) { + throw new IllegalStateException( + "The commit lease duration changed during renewal."); + } + renewedAt = requestedAt; + } catch (Exception | Error e) { + renewalFailure = e; + } + } + + private void checkValid() { + // Include response time conservatively without relying on the renewal worker running. + if (TimeUnit.NANOSECONDS.toMillis(nanoTime.getAsLong() - renewedAt) >= leaseMillis + && renewalFailure == null) { + renewalFailure = new IllegalStateException("The commit lease has expired locally."); + } + if (closed.get() || renewalFailure != null) { + throw new IllegalStateException( + "The commit lease is no longer usable.", renewalFailure); + } + } + + @Override + public void close() { + if (closed.compareAndSet(false, true)) { + try { + renewal.cancel(false); + } finally { + FutureTask request = renewalRequest; + if (request != null) { + request.cancel(true); + } + } + } + } + } + + @Override + public void close() { + // Lease scopes cancel their own tasks; shared executors outlive individual locks. + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalogLockFactory.java b/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalogLockFactory.java new file mode 100644 index 000000000000..bb91fd016b3d --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalogLockFactory.java @@ -0,0 +1,63 @@ +/* + * 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.paimon.rest; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.catalog.CatalogLockContext; +import org.apache.paimon.catalog.CatalogLockFactory; +import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.operation.Lock; +import org.apache.paimon.options.Options; + +import javax.annotation.Nullable; + +import static org.apache.paimon.utils.Preconditions.checkArgument; + +/** Creates REST commit lease clients from serializable catalog options. */ +public class RESTCatalogLockFactory implements CatalogLockFactory { + + private static final long serialVersionUID = 1L; + + @Override + public String identifier() { + return RESTCatalogFactory.IDENTIFIER; + } + + @Override + public Lock createLock( + CatalogLockContext context, + Identifier identifier, + @Nullable String tableUuid, + @Nullable String commitUser, + Options tableOptions) { + CoreOptions options = new CoreOptions(tableOptions); + if (!options.restCommitAcquireLock()) { + return Lock.empty(); + } + checkArgument( + options.restCommitLockEnabled(), + "rest.commit.acquire-lock requires rest.commit.lock-enabled on the table."); + return new RESTCatalogLock( + new RESTApi(context.options(), false), + identifier, + tableUuid, + commitUser, + options.commitTimeout()); + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java b/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java index 5988ee128f1c..7cea98df0069 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java @@ -180,7 +180,7 @@ private Lock createFactoryLock( lockContext, lockIdentifier, uuid, commitUser, tableOptions); } - /** Create rollback with the table incarnation and optional writer lock context. */ + /** Create rollback with the table incarnation and optional writer identity. */ @Nullable public TableRollback catalogTableRollback( @Nullable String branch, @Nullable String commitUser) { @@ -193,9 +193,15 @@ public TableRollback catalogTableRollback( identifier.getDatabaseName(), identifier.getTableName(), branch); - return (instant, fromSnapshot) -> { + return (instant, fromSnapshot, fromSnapshotUuid) -> { try { - catalog.rollbackTo(rollbackIdentifier, uuid, instant, fromSnapshot, commitUser); + catalog.rollbackTo( + rollbackIdentifier, + uuid, + instant, + fromSnapshot, + fromSnapshotUuid, + commitUser); } catch (Catalog.TableNotExistException e) { throw new RuntimeException(e); } diff --git a/paimon-core/src/main/java/org/apache/paimon/tag/SnapshotLoaderImpl.java b/paimon-core/src/main/java/org/apache/paimon/tag/SnapshotLoaderImpl.java index 8a47e88b1b93..3aab3235aec8 100644 --- a/paimon-core/src/main/java/org/apache/paimon/tag/SnapshotLoaderImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/tag/SnapshotLoaderImpl.java @@ -59,7 +59,7 @@ public Optional load() throws IOException { @Override public void rollback(Instant instant) throws IOException { try (Catalog catalog = catalogLoader.load()) { - catalog.rollbackTo(identifier, uuid, instant, null, null); + catalog.rollbackTo(identifier, uuid, instant, null, null, null); } catch (RuntimeException e) { throw e; } catch (Exception e) { diff --git a/paimon-core/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory b/paimon-core/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory index 3e678972d8e8..2e39575c5fe9 100644 --- a/paimon-core/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory +++ b/paimon-core/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory @@ -39,3 +39,4 @@ org.apache.paimon.mergetree.compact.aggregate.factory.FieldRoaringBitmap64AggFac org.apache.paimon.mergetree.compact.aggregate.factory.FieldSumAggFactory org.apache.paimon.mergetree.compact.aggregate.factory.FieldThetaSketchAggFactory org.apache.paimon.rest.RESTCatalogFactory +org.apache.paimon.rest.RESTCatalogLockFactory diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/commit/CommitRollbackTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/commit/CommitRollbackTest.java new file mode 100644 index 000000000000..6e0c4e50dc97 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/operation/commit/CommitRollbackTest.java @@ -0,0 +1,60 @@ +/* + * 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.paimon.operation.commit; + +import org.apache.paimon.Snapshot; +import org.apache.paimon.catalog.TableRollback; +import org.apache.paimon.table.Instant; + +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +class CommitRollbackTest { + + @Test + void legacyCompactionWithoutUuidCannotBeRolledBackAutomatically() { + Snapshot head = mock(Snapshot.class); + when(head.commitKind()).thenReturn(Snapshot.CommitKind.COMPACT); + TableRollback rollback = mock(TableRollback.class); + + assertThat(new CommitRollback(rollback).tryToRollback(head)).isFalse(); + verifyNoInteractions(rollback); + } + + @Test + void changedHeadRejectsRollbackAndLeavesTheConflictUnresolved() { + Snapshot head = mock(Snapshot.class); + when(head.commitKind()).thenReturn(Snapshot.CommitKind.COMPACT); + when(head.id()).thenReturn(6L); + when(head.uuid()).thenReturn("head-uuid"); + TableRollback rollback = mock(TableRollback.class); + doThrow(new IllegalStateException("Head UUID changed")) + .when(rollback) + .rollbackTo(any(Instant.class), eq(6L), eq("head-uuid")); + + assertThat(new CommitRollback(rollback).tryToRollback(head)).isFalse(); + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/rest/CommitLockProtocolTest.java b/paimon-core/src/test/java/org/apache/paimon/rest/CommitLockProtocolTest.java new file mode 100644 index 000000000000..5c046d80f85b --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/rest/CommitLockProtocolTest.java @@ -0,0 +1,266 @@ +/* + * 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.paimon.rest; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.Snapshot; +import org.apache.paimon.catalog.Catalog; +import org.apache.paimon.catalog.CatalogContext; +import org.apache.paimon.catalog.CatalogLoader; +import org.apache.paimon.catalog.CatalogLockFactory; +import org.apache.paimon.catalog.DelegateCatalog; +import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.catalog.RenamingSnapshotCommit; +import org.apache.paimon.factories.FactoryUtil; +import org.apache.paimon.operation.Lock; +import org.apache.paimon.options.Options; +import org.apache.paimon.rest.requests.CommitLockRequest; +import org.apache.paimon.rest.requests.CommitTableRequest; +import org.apache.paimon.rest.requests.RollbackTableRequest; +import org.apache.paimon.rest.responses.CommitLockResponse; +import org.apache.paimon.rest.responses.GetTableResponse; +import org.apache.paimon.schema.Schema; +import org.apache.paimon.table.CatalogEnvironment; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.Instant; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.utils.InstantiationUtil; + +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.node.ObjectNode; + +import okhttp3.mockwebserver.MockResponse; +import okhttp3.mockwebserver.MockWebServer; +import okhttp3.mockwebserver.RecordedRequest; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.nio.file.Path; +import java.util.Collections; +import java.util.concurrent.TimeUnit; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests the owner lease wire protocol without changing legacy snapshot requests. */ +class CommitLockProtocolTest { + @Test + void acquireRenewAndCommitUseTheSameOwnerWithoutToken() throws Exception { + try (MockWebServer server = new MockWebServer()) { + server.start(); + Options options = new Options(); + options.set(RESTCatalogOptions.URI, server.url("/").toString()); + options.set(RESTCatalogOptions.TOKEN_PROVIDER, "bear"); + options.set(RESTCatalogOptions.TOKEN, "token"); + options.set(RESTCatalogInternalOptions.PREFIX, "catalog"); + RESTApi api = new RESTApi(options, false); + Identifier identifier = Identifier.create("db", "table"); + CommitLockResponse grant = + new CommitLockResponse(true, "morax-job", 100000, 60000, snapshot(1)); + server.enqueue(json(RESTApi.toJson(grant))); + CommitLockResponse acquired = + api.acquireCommitLock(identifier, "table-id", "morax-job"); + assertThat(acquired.isAcquired()).isTrue(); + assertThat(acquired.getSnapshot().id()).isEqualTo(1); + RecordedRequest acquire = server.takeRequest(10, TimeUnit.SECONDS); + assertThat(acquire.getPath()) + .isEqualTo("/v1/catalog/databases/db/tables/table/commit-lock"); + CommitLockRequest request = + RESTApi.fromJson(acquire.getBody().readUtf8(), CommitLockRequest.class); + assertThat(request.getTableId()).isEqualTo("table-id"); + assertThat(request.getCommitUser()).isEqualTo("morax-job"); + server.enqueue(json(RESTApi.toJson(grant))); + assertThat(api.renewCommitLock(identifier, "table-id", "morax-job").isAcquired()) + .isTrue(); + assertThat(server.takeRequest(10, TimeUnit.SECONDS).getPath()) + .endsWith("/commit-lock/renew"); + server.enqueue(json("{\"success\":true}")); + assertThat( + api.commitSnapshot( + identifier, + "table-id", + null, + snapshot(2), + Collections.emptyList())) + .isTrue(); + String commitBody = server.takeRequest(10, TimeUnit.SECONDS).getBody().readUtf8(); + ObjectNode payload = RESTApi.fromJson(commitBody, ObjectNode.class); + assertThat(payload.has("lockToken")).isFalse(); + assertThat(payload.has("commitUser")).isFalse(); + assertThat( + RESTApi.fromJson(commitBody, CommitTableRequest.class) + .getSnapshot() + .commitUser()) + .isEqualTo("morax-job"); + assertThat(RESTApi.fromJson(commitBody, CommitTableRequest.class).getTableId()) + .isEqualTo("table-id"); + } + } + + @Test + void rollbackWithinLeaseCarriesOwnerAndLegacyPayloadStaysUnchanged() throws Exception { + try (MockWebServer server = new MockWebServer()) { + server.start(); + Options options = new Options(); + options.set(RESTCatalogOptions.URI, server.url("/").toString()); + options.set(RESTCatalogOptions.TOKEN_PROVIDER, "bear"); + options.set(RESTCatalogOptions.TOKEN, "token"); + options.set(RESTCatalogInternalOptions.PREFIX, "catalog"); + Catalog catalog = + new DelegateCatalog(new RESTCatalog(CatalogContext.create(options), false)) { + @Override + public CatalogLoader catalogLoader() { + return wrapped.catalogLoader(); + } + }; + Identifier identifier = new Identifier("db", "table", "dev"); + server.enqueue(json("{}")); + catalog.rollbackTo( + identifier, "table-id", Instant.snapshot(1L), 2L, "head-uuid", "morax-job"); + RecordedRequest owned = server.takeRequest(10, TimeUnit.SECONDS); + assertThat(owned.getPath()).endsWith("/rollback"); + RollbackTableRequest request = + RESTApi.fromJson(owned.getBody().readUtf8(), RollbackTableRequest.class); + assertThat(request.getTableId()).isEqualTo("table-id"); + assertThat(request.getCommitUser()).isEqualTo("morax-job"); + assertThat(request.getFromSnapshot()).isEqualTo(2L); + server.enqueue(json("{}")); + catalog.rollbackTo(identifier, null, Instant.snapshot(1L), 2L, null, null); + ObjectNode legacy = + RESTApi.fromJson( + server.takeRequest(10, TimeUnit.SECONDS).getBody().readUtf8(), + ObjectNode.class); + assertThat(legacy.has("tableId")).isFalse(); + assertThat(legacy.has("commitUser")).isFalse(); + assertThat(legacy.has("fromSnapshotUuid")).isFalse(); + } + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void restTableCarriesSerializableCatalogLockEntry(boolean external, @TempDir Path directory) + throws Exception { + try (MockWebServer server = new MockWebServer()) { + server.start(); + Options options = new Options(); + options.set(RESTCatalogOptions.URI, server.url("/").toString()); + options.set(RESTCatalogOptions.TOKEN_PROVIDER, "bear"); + options.set(RESTCatalogOptions.TOKEN, "token"); + options.set(RESTCatalogInternalOptions.PREFIX, "catalog"); + options.set(RESTTokenFileIO.DATA_TOKEN_ENABLED, false); + GetTableResponse tableResponse = + new GetTableResponse( + "table-id", + "db", + "table$branch_dev", + directory.toUri().toString(), + external, + 0, + Schema.newBuilder().column("id", DataTypes.INT()).build(), + null, + 0, + null, + 0, + null); + server.enqueue(json(RESTApi.toJson(tableResponse))); + options.set(RESTCatalogInternalOptions.PREFIX, "catalog"); + RESTCatalog catalog = new RESTCatalog(CatalogContext.create(options), false); + Identifier identifier = new Identifier("db", "table", "dev"); + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier); + assertThat(server.takeRequest(10, TimeUnit.SECONDS).getPath()).contains("table"); + CatalogEnvironment environment = InstantiationUtil.clone(table.catalogEnvironment()); + assertThat(environment.uuid()).isEqualTo("table-id"); + if (external) { + assertThat(environment.catalogLoader()).isNull(); + assertThat(environment.lockFactory()).isNull(); + assertThat(environment.lockContext()).isNull(); + assertThat( + environment.snapshotCommit( + table.snapshotManager(), table.coreOptions(), "morax-job")) + .isInstanceOf(RenamingSnapshotCommit.class); + return; + } + assertThat(environment.lockFactory()).isNull(); + assertThat( + FactoryUtil.discoverFactory( + getClass().getClassLoader(), CatalogLockFactory.class, "rest")) + .isInstanceOf(RESTCatalogLockFactory.class); + server.enqueue( + json( + RESTApi.toJson( + new CommitLockResponse( + true, "morax-job", 100000, 60000, snapshot(1))))); + Options tableOptions = Options.fromMap(table.options()); + tableOptions.set(CoreOptions.REST_COMMIT_LOCK_ENABLED, true); + tableOptions.set(CoreOptions.REST_COMMIT_ACQUIRE_LOCK, true); + try (Catalog writerCatalog = environment.catalogLoader().load(); + Lock lock = + writerCatalog.createLock( + identifier, environment.uuid(), "morax-job", tableOptions)) { + assertThat( + lock.runWithLock( + () -> { + lock.ensureValid(); + return "published"; + })) + .isEqualTo("published"); + RecordedRequest acquire = server.takeRequest(10, TimeUnit.SECONDS); + assertThat(acquire.getPath()).endsWith("/commit-lock"); + assertThat(acquire.getHeader("Authorization")).isEqualTo("Bearer token"); + CommitLockRequest request = + RESTApi.fromJson(acquire.getBody().readUtf8(), CommitLockRequest.class); + assertThat(request.getTableId()).isEqualTo("table-id"); + assertThat(request.getCommitUser()).isEqualTo("morax-job"); + } + } + } + + private static Snapshot snapshot(long id) { + return new Snapshot( + id, + 0L, + null, + null, + null, + null, + null, + null, + null, + "morax-job", + null, + 0L, + Snapshot.CommitKind.APPEND, + 1000L, + 0L, + 0L, + null, + null, + null, + null, + null, + null); + } + + private static MockResponse json(String body) { + return new MockResponse() + .setResponseCode(200) + .setBody(body) + .addHeader("Content-Type", "application/json"); + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTMessage.java b/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTMessage.java index dafeae833a03..caa1e5c08d56 100644 --- a/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTMessage.java +++ b/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTMessage.java @@ -52,6 +52,7 @@ import org.apache.paimon.schema.Schema; import org.apache.paimon.schema.SchemaChange; import org.apache.paimon.table.Instant; +import org.apache.paimon.table.TableSnapshot; import org.apache.paimon.types.DataField; import org.apache.paimon.types.DataType; import org.apache.paimon.types.DataTypes; @@ -66,6 +67,8 @@ import org.apache.paimon.shade.guava30.com.google.common.collect.Lists; import org.apache.paimon.shade.guava30.com.google.common.collect.Maps; +import javax.annotation.Nullable; + import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -79,6 +82,23 @@ /** Mock REST message. */ public class MockRESTMessage { + public static void validateRollbackHead( + RollbackTableRequest request, @Nullable TableSnapshot current) { + if (request.getFromSnapshot() != null + && (current == null || request.getFromSnapshot() != current.snapshot().id())) { + throw new IllegalStateException( + String.format( + "Latest snapshot %s is not %s", + current == null ? null : current.snapshot().id(), + request.getFromSnapshot())); + } + if (request.getFromSnapshotUuid() != null + && (current == null + || !request.getFromSnapshotUuid().equals(current.snapshot().uuid()))) { + throw new IllegalStateException("Latest snapshot UUID changed"); + } + } + public static String databaseName() { return "database"; } @@ -274,11 +294,11 @@ public static GetTableTokenResponse getTableCredentialsResponse() { } public static RollbackTableRequest rollbackTableRequestBySnapshot(long snapshotId) { - return new RollbackTableRequest(Instant.snapshot(snapshotId), null, null, null); + return new RollbackTableRequest(Instant.snapshot(snapshotId), null, null, null, null); } public static RollbackTableRequest rollbackTableRequestByTag(String tagName) { - return new RollbackTableRequest(Instant.tag(tagName), null, null, null); + return new RollbackTableRequest(Instant.tag(tagName), null, null, null, null); } public static AlterViewRequest alterViewRequest() { diff --git a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogLockTest.java b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogLockTest.java new file mode 100644 index 000000000000..91856529bba3 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogLockTest.java @@ -0,0 +1,845 @@ +/* + * 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.paimon.rest; + +import org.apache.paimon.catalog.Catalog; +import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.operation.Lock; +import org.apache.paimon.rest.exceptions.ForbiddenException; +import org.apache.paimon.rest.exceptions.NoSuchResourceException; +import org.apache.paimon.rest.responses.CommitLockResponse; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import org.mockito.ArgumentCaptor; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.Callable; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +/** Tests renewal, expiry and scoped execution of REST catalog locks. */ +class RESTCatalogLockTest { + private final RESTApi api = mock(RESTApi.class); + private final ScheduledExecutorService renewer = mock(ScheduledExecutorService.class); + private final ExecutorService workers = mock(ExecutorService.class); + private final ScheduledFuture future = mock(ScheduledFuture.class); + private final Identifier identifier = new Identifier("db", "table", "dev"); + private final AtomicLong clock = new AtomicLong(); + private final RESTCatalogLock lock = + new RESTCatalogLock( + api, identifier, "table-id", "morax-job", 0, renewer, workers, clock::get); + + @Test + void idleSchedulerCanRestartAfterTheLastLeaseEnds() throws Exception { + grant("morax-job", 60000); + ScheduledThreadPoolExecutor scheduler = RESTCatalogLock.createRenewalScheduler(); + scheduler.setKeepAliveTime(10, TimeUnit.MILLISECONDS); + RESTCatalogLock realLock = + new RESTCatalogLock( + api, + identifier, + "table-id", + "morax-job", + 0, + scheduler, + workers, + clock::get); + try { + Thread firstTimer; + try (AutoCloseable ignored = realLock.acquire()) { + firstTimer = scheduler.submit(Thread::currentThread).get(10, TimeUnit.SECONDS); + realLock.ensureValid(); + } + firstTimer.join(2000); + assertThat(firstTimer.isAlive()).isFalse(); + assertThat(scheduler.getPoolSize()).isZero(); + + Thread nextTimer; + try (AutoCloseable ignored = realLock.acquire()) { + nextTimer = scheduler.submit(Thread::currentThread).get(10, TimeUnit.SECONDS); + assertThat(nextTimer).isNotSameAs(firstTimer); + realLock.ensureValid(); + } + nextTimer.join(2000); + assertThat(nextTimer.isAlive()).isFalse(); + } finally { + scheduler.shutdownNow(); + assertThat(scheduler.awaitTermination(10, TimeUnit.SECONDS)).isTrue(); + } + } + + @Test + void closingOneLeaseKeepsTheSharedSchedulerRenewingOtherLeases() throws Exception { + grant("morax-job", 3000); + CommitLockResponse otherGrant = + new CommitLockResponse(true, "other-job", 100000, 3000, null); + when(api.acquireCommitLock(identifier, "table-id", "other-job")).thenReturn(otherGrant); + CountDownLatch otherRenewed = new CountDownLatch(1); + AtomicBoolean firstClosed = new AtomicBoolean(); + when(api.renewCommitLock(identifier, "table-id", "other-job")) + .thenAnswer( + invocation -> { + if (firstClosed.get()) { + otherRenewed.countDown(); + } + return otherGrant; + }); + ScheduledThreadPoolExecutor scheduler = RESTCatalogLock.createRenewalScheduler(); + scheduler.setKeepAliveTime(10, TimeUnit.MILLISECONDS); + RESTCatalogLock firstLock = + new RESTCatalogLock( + api, + identifier, + "table-id", + "morax-job", + 0, + scheduler, + workers, + clock::get); + RESTCatalogLock otherLock = + new RESTCatalogLock( + api, + identifier, + "table-id", + "other-job", + 0, + scheduler, + workers, + clock::get); + Thread timer; + try { + try (AutoCloseable first = firstLock.acquire(); + AutoCloseable other = otherLock.acquire()) { + timer = scheduler.submit(Thread::currentThread).get(10, TimeUnit.SECONDS); + first.close(); + firstClosed.set(true); + assertThat(otherRenewed.await(10, TimeUnit.SECONDS)).isTrue(); + otherLock.ensureValid(); + assertThat(timer.isAlive()).isTrue(); + } + timer.join(2000); + assertThat(timer.isAlive()).isFalse(); + assertThat(scheduler.getQueue()).isEmpty(); + } finally { + scheduler.shutdownNow(); + assertThat(scheduler.awaitTermination(10, TimeUnit.SECONDS)).isTrue(); + } + } + + @Test + void callbackIsProtectedAndLeaseEndsOnReturn() throws Exception { + grant("morax-job", 60000); + assertThat( + run( + () -> { + lock.ensureValid(); + return "published"; + })) + .isEqualTo("published"); + verify(future).cancel(false); + assertThatThrownBy(lock::ensureValid).hasMessageContaining("No commit lease"); + } + + @Test + void callbackFailureStopsRenewal() { + grant("morax-job", 60000); + RuntimeException failure = new RuntimeException("Preparation failed"); + assertThatThrownBy( + () -> + run( + () -> { + throw failure; + })) + .isSameAs(failure); + verify(future).cancel(false); + } + + @Test + void renewalOnAnotherThreadExtendsTheAcquiringThreadsLease() throws Exception { + grant("morax-job", 60000); + ExecutorService renewalWorker = Executors.newSingleThreadExecutor(); + RESTCatalogLock realLock = + new RESTCatalogLock( + api, + identifier, + "table-id", + "morax-job", + 0, + renewer, + renewalWorker, + clock::get); + Thread acquiringThread = Thread.currentThread(); + when(api.renewCommitLock(identifier, "table-id", "morax-job")) + .thenAnswer( + invocation -> { + assertThat(Thread.currentThread()).isNotSameAs(acquiringThread); + assertThatThrownBy(realLock::ensureValid) + .hasMessageContaining("No commit lease"); + return new CommitLockResponse(true, "morax-job", 140000, 60000, null); + }); + try { + try (AutoCloseable ignored = realLock.acquire()) { + clock.set(TimeUnit.MILLISECONDS.toNanos(40000)); + renewal().run(); + // A FIFO barrier waits until the renewal and its deadline update have completed. + renewalWorker.submit(() -> {}).get(10, TimeUnit.SECONDS); + verify(api).renewCommitLock(identifier, "table-id", "morax-job"); + + clock.set(TimeUnit.MILLISECONDS.toNanos(70000)); + realLock.ensureValid(); + clock.set(TimeUnit.MILLISECONDS.toNanos(100000)); + assertThatThrownBy(realLock::ensureValid) + .hasRootCauseMessage("The commit lease has expired locally."); + } + verify(future).cancel(false); + } finally { + renewalWorker.shutdownNow(); + assertThat(renewalWorker.awaitTermination(10, TimeUnit.SECONDS)).isTrue(); + } + } + + @Test + void renewalUsesExactOwnerAndStopsAfterClose() throws Exception { + grant("morax-job", 60000); + when(api.renewCommitLock(identifier, "table-id", "morax-job")) + .thenReturn(new CommitLockResponse(true, "morax-job", 100000, 60000, null)); + Runnable callback = + run( + () -> { + Runnable renewal = renewal(); + renewal.run(); + lock.ensureValid(); + return renewal; + }); + callback.run(); + verify(api).renewCommitLock(identifier, "table-id", "morax-job"); + verify(future).cancel(false); + } + + @Test + void rejectedRenewalInvalidatesLease() throws Exception { + grant("morax-job", 60000); + when(api.renewCommitLock(identifier, "table-id", "morax-job")) + .thenReturn(CommitLockResponse.unavailable()); + run( + () -> { + renewal().run(); + renewal().run(); + assertThatThrownBy(lock::ensureValid) + .hasCauseInstanceOf(IllegalStateException.class); + return null; + }); + verify(api).renewCommitLock(identifier, "table-id", "morax-job"); + } + + @Test + void renewalExceptionPreservesCause() throws Exception { + grant("morax-job", 60000); + RuntimeException failure = new RuntimeException("Disconnected"); + when(api.renewCommitLock(identifier, "table-id", "morax-job")).thenThrow(failure); + run( + () -> { + renewal().run(); + assertThatThrownBy(lock::ensureValid).hasCause(failure); + return null; + }); + } + + @Test + void busyTimeoutDoesNotExecuteCallbackOrStartRenewal() { + when(api.acquireCommitLock(identifier, "table-id", "morax-job")) + .thenReturn(CommitLockResponse.unavailable()); + assertThatThrownBy( + () -> + run( + () -> { + throw new AssertionError(); + })) + .isInstanceOf(Lock.AcquisitionException.class) + .hasRootCauseMessage("Timed out acquiring commit lock for " + identifier); + verifyNoInteractions(renewer); + } + + @Test + void invalidGrantDoesNotStartRenewal() { + grant("other-job", 60000); + assertThatThrownBy(() -> run(() -> null)).isInstanceOf(Lock.AcquisitionException.class); + grant("morax-job", 0); + assertThatThrownBy(() -> run(() -> null)).isInstanceOf(Lock.AcquisitionException.class); + verifyNoInteractions(renewer); + } + + @Test + void missingIdentityAndGeneralLockAreRejectedBeforeRequest() { + assertThatThrownBy(() -> new RESTCatalogLock(api, identifier, "", "morax-job", 0)) + .isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy(() -> new RESTCatalogLock(api, identifier, "table-id", "", 0)) + .isInstanceOf(IllegalArgumentException.class); + verifyNoInteractions(api, renewer); + } + + @Test + void expiryIsDetectedEvenWhenRenewerHasNotRun() throws Exception { + grant("morax-job", 60000); + run( + () -> { + clock.set(TimeUnit.MILLISECONDS.toNanos(60000)); + assertThatThrownBy(lock::ensureValid).hasMessageContaining("no longer usable"); + renewal().run(); + verify(api, never()).renewCommitLock(identifier, "table-id", "morax-job"); + return null; + }); + } + + @Test + void renewalExtendsLocalLifetimeAndIncludesResponseTime() throws Exception { + grant("morax-job", 60000); + when(api.renewCommitLock(identifier, "table-id", "morax-job")) + .thenAnswer( + invocation -> { + clock.addAndGet(TimeUnit.MILLISECONDS.toNanos(10000)); + return new CommitLockResponse(true, "morax-job", 100000, 60000, null); + }); + run( + () -> { + clock.set(TimeUnit.MILLISECONDS.toNanos(20000)); + renewal().run(); + clock.set(TimeUnit.MILLISECONDS.toNanos(79999)); + lock.ensureValid(); + clock.set(TimeUnit.MILLISECONDS.toNanos(80000)); + assertThatThrownBy(lock::ensureValid).isInstanceOf(IllegalStateException.class); + return null; + }); + } + + @Test + void acquirePreservesCatalogPermissionAndMissingTableErrors() { + ForbiddenException denied = new ForbiddenException("Denied"); + when(api.acquireCommitLock(identifier, "table-id", "morax-job")).thenThrow(denied); + assertThatThrownBy(() -> run(() -> null)) + .isInstanceOf(Lock.AcquisitionException.class) + .hasCauseInstanceOf(Catalog.TableNoPermissionException.class) + .hasRootCause(denied); + NoSuchResourceException missing = new NoSuchResourceException("table", "table", "Missing"); + doThrow(missing).when(api).acquireCommitLock(identifier, "table-id", "morax-job"); + assertThatThrownBy(() -> run(() -> null)) + .isInstanceOf(Lock.AcquisitionException.class) + .hasCauseInstanceOf(Catalog.TableNotExistException.class) + .hasRootCause(missing); + verifyNoInteractions(renewer); + } + + @Test + void slowAcquisitionCannotStartProtectedOperation() { + grant("morax-job", 60000); + when(api.acquireCommitLock(identifier, "table-id", "morax-job")) + .thenAnswer( + invocation -> { + clock.addAndGet(TimeUnit.MILLISECONDS.toNanos(60000)); + return new CommitLockResponse(true, "morax-job", 100000, 60000, null); + }); + assertThatThrownBy( + () -> + run( + () -> { + throw new AssertionError(); + })) + .isInstanceOf(Lock.AcquisitionException.class) + .hasRootCauseMessage("The commit lease has expired locally."); + verify(future).cancel(false); + verify(renewer, never()).shutdownNow(); + assertThatThrownBy(lock::ensureValid).hasMessageContaining("No commit lease"); + } + + @Test + void schedulerFailureKeepsSharedExecutorsUsableForFreshAcquisition() throws Exception { + grant("morax-job", 60000); + RejectedExecutionException failure = new RejectedExecutionException("Stopped worker"); + when(renewer.scheduleWithFixedDelay(any(Runnable.class), anyLong(), anyLong(), any())) + .thenThrow(failure); + assertThatThrownBy( + () -> + run( + () -> { + throw new AssertionError(); + })) + .isInstanceOf(Lock.AcquisitionException.class) + .hasCause(failure); + verify(renewer, never()).shutdownNow(); + assertThatThrownBy(lock::ensureValid).hasMessageContaining("No commit lease"); + doReturn(future) + .when(renewer) + .scheduleWithFixedDelay(any(Runnable.class), anyLong(), anyLong(), any()); + assertThat(run(() -> "published")).isEqualTo("published"); + verify(renewer, never()).shutdownNow(); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void taskCancellationFailurePreservesOperationOutcome(boolean failOperation) throws Exception { + grant("morax-job", 60000); + RuntimeException releaseFailure = new RuntimeException("Task cancellation failed"); + RuntimeException operationFailure = new RuntimeException("Preparation failed"); + doThrow(releaseFailure).when(future).cancel(false); + Callable operation = + () -> { + if (failOperation) { + throw operationFailure; + } + return "published"; + }; + if (failOperation) { + assertThatThrownBy(() -> run(operation)) + .isSameAs(operationFailure) + .satisfies(e -> assertThat(e.getSuppressed()).containsExactly(releaseFailure)); + } else { + assertThat(run(operation)).isEqualTo("published"); + } + verify(future).cancel(false); + assertThatThrownBy(lock::ensureValid).hasMessageContaining("No commit lease"); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void invalidRenewalGrantCannotExtendLease(boolean differentOwner) throws Exception { + grant("morax-job", 60000); + when(api.renewCommitLock(identifier, "table-id", "morax-job")) + .thenReturn( + new CommitLockResponse( + true, + differentOwner ? "other-job" : "morax-job", + 100000, + differentOwner ? 60000 : 120000, + null)); + run( + () -> { + renewal().run(); + assertThatThrownBy(lock::ensureValid) + .isInstanceOf(IllegalStateException.class) + .hasRootCauseMessage( + differentOwner + ? "The server returned an invalid commit lease." + : "The commit lease duration changed during renewal."); + return null; + }); + } + + @Test + void closingAScopeKeepsSharedRenewalExecutorsUsable() throws Exception { + grant("morax-job", 60000); + ScheduledFuture secondRenewal = mock(ScheduledFuture.class); + when(renewer.scheduleWithFixedDelay(any(Runnable.class), anyLong(), anyLong(), any())) + .thenAnswer(invocation -> future) + .thenAnswer(invocation -> secondRenewal); + assertThat(run(() -> "first")).isEqualTo("first"); + assertThat(run(() -> "second")).isEqualTo("second"); + verify(future).cancel(false); + verify(secondRenewal).cancel(false); + verify(renewer, never()).shutdownNow(); + verify(workers, never()).shutdownNow(); + } + + @Test + void queuedRenewalIsCancelledAndCannotAffectANewScope() throws Exception { + grant("morax-job", 60000); + when(api.renewCommitLock(identifier, "table-id", "morax-job")) + .thenReturn(new CommitLockResponse(true, "morax-job", 100000, 60000, null)); + List requests = queueRenewals(); + AutoCloseable first = lock.acquire(); + Runnable oldTick = renewal(); + oldTick.run(); + oldTick.run(); + assertThat(requests).hasSize(1); + first.close(); + requests.get(0).run(); + verify(api, never()).renewCommitLock(identifier, "table-id", "morax-job"); + try (AutoCloseable second = lock.acquire()) { + first.close(); + oldTick.run(); + assertThat(requests).hasSize(1); + renewal(2).run(); + assertThat(requests).hasSize(2); + requests.get(1).run(); + lock.ensureValid(); + } + verify(api).renewCommitLock(identifier, "table-id", "morax-job"); + verify(workers, never()).shutdownNow(); + } + + @Test + void queuedRenewalCannotReviveALocallyExpiredLease() throws Exception { + grant("morax-job", 60000); + List requests = queueRenewals(); + run( + () -> { + renewal().run(); + clock.set(TimeUnit.MILLISECONDS.toNanos(60000)); + requests.get(0).run(); + assertThatThrownBy(lock::ensureValid) + .hasRootCauseMessage("The commit lease has expired locally."); + return null; + }); + verify(api, never()).renewCommitLock(identifier, "table-id", "morax-job"); + } + + @Test + void workerDispatchFailureInvalidatesOnlyItsScope() throws Exception { + grant("morax-job", 60000); + RejectedExecutionException failure = new RejectedExecutionException("Dispatch rejected"); + doThrow(failure).when(workers).execute(any(Runnable.class)); + run( + () -> { + renewal().run(); + assertThatThrownBy(lock::ensureValid).hasCause(failure); + return null; + }); + grant("morax-job", 60000); + when(api.renewCommitLock(identifier, "table-id", "morax-job")) + .thenReturn(new CommitLockResponse(true, "morax-job", 100000, 60000, null)); + run( + () -> { + renewal(2).run(); + lock.ensureValid(); + return null; + }); + verify(api).renewCommitLock(identifier, "table-id", "morax-job"); + verify(workers, never()).shutdownNow(); + } + + @Test + void renewalErrorInvalidatesLeaseAndStopsFurtherDispatch() throws Exception { + grant("morax-job", 60000); + AssertionError failure = new AssertionError("RPC failed"); + when(api.renewCommitLock(identifier, "table-id", "morax-job")).thenThrow(failure); + run( + () -> { + Runnable tick = renewal(); + tick.run(); + tick.run(); + assertThatThrownBy(lock::ensureValid).hasCause(failure); + return null; + }); + verify(api).renewCommitLock(identifier, "table-id", "morax-job"); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void lateOldRenewalCannotChangeANewScope(boolean failOldRenewal) throws Exception { + grant("morax-job", 60000); + List requests = queueRenewals(); + CountDownLatch started = new CountDownLatch(1); + CountDownLatch resume = new CountDownLatch(1); + CommitLockResponse renewed = new CommitLockResponse(true, "morax-job", 100000, 60000, null); + when(api.renewCommitLock(identifier, "table-id", "morax-job")) + .thenAnswer( + invocation -> { + started.countDown(); + awaitIgnoringInterrupts(resume); + if (failOldRenewal) { + throw new IllegalStateException("Old renewal failed"); + } + return renewed; + }) + .thenReturn(renewed); + ExecutorService runner = Executors.newSingleThreadExecutor(); + try (AutoCloseable first = lock.acquire()) { + clock.set(TimeUnit.MILLISECONDS.toNanos(20000)); + renewal().run(); + Future oldResponse = runner.submit(requests.get(0)); + assertThat(started.await(10, TimeUnit.SECONDS)).isTrue(); + first.close(); + try (AutoCloseable second = lock.acquire()) { + clock.set(TimeUnit.MILLISECONDS.toNanos(30000)); + renewal(2).run(); + requests.get(1).run(); + resume.countDown(); + oldResponse.get(10, TimeUnit.SECONDS); + // Only the new scope's renewal at 30s keeps it valid beyond 80s. + clock.set(TimeUnit.MILLISECONDS.toNanos(85000)); + lock.ensureValid(); + } + } finally { + resume.countDown(); + runner.shutdownNow(); + assertThat(runner.awaitTermination(10, TimeUnit.SECONDS)).isTrue(); + } + verify(api, times(2)).renewCommitLock(identifier, "table-id", "morax-job"); + } + + @Test + void lateSuccessfulRenewalCannotReviveAnExpiredScope() throws Exception { + grant("morax-job", 60000); + List requests = queueRenewals(); + CountDownLatch started = new CountDownLatch(1); + CountDownLatch resume = new CountDownLatch(1); + when(api.renewCommitLock(identifier, "table-id", "morax-job")) + .thenAnswer( + invocation -> { + started.countDown(); + assertThat(resume.await(10, TimeUnit.SECONDS)).isTrue(); + return new CommitLockResponse(true, "morax-job", 100000, 60000, null); + }); + ExecutorService runner = Executors.newSingleThreadExecutor(); + try (AutoCloseable scope = lock.acquire()) { + clock.set(TimeUnit.MILLISECONDS.toNanos(20000)); + renewal().run(); + Future response = runner.submit(requests.get(0)); + assertThat(started.await(10, TimeUnit.SECONDS)).isTrue(); + clock.set(TimeUnit.MILLISECONDS.toNanos(60000)); + assertThatThrownBy(lock::ensureValid) + .hasRootCauseMessage("The commit lease has expired locally."); + resume.countDown(); + response.get(10, TimeUnit.SECONDS); + assertThatThrownBy(lock::ensureValid) + .hasRootCauseMessage("The commit lease has expired locally."); + } finally { + resume.countDown(); + runner.shutdownNow(); + assertThat(runner.awaitTermination(10, TimeUnit.SECONDS)).isTrue(); + } + } + + @Test + void blockedRenewalDoesNotBlockAnotherLeaseAndCloseInterruptsOnlyItsRequest() throws Exception { + grant("morax-job", 60000); + ScheduledFuture secondRenewal = mock(ScheduledFuture.class); + when(renewer.scheduleWithFixedDelay(any(Runnable.class), anyLong(), anyLong(), any())) + .thenAnswer(invocation -> future) + .thenAnswer(invocation -> secondRenewal); + when(api.acquireCommitLock(identifier, "table-id", "other-job")) + .thenReturn(new CommitLockResponse(true, "other-job", 100000, 60000, null)); + CountDownLatch firstStarted = new CountDownLatch(1); + CountDownLatch resumeFirst = new CountDownLatch(1); + CountDownLatch firstInterrupted = new CountDownLatch(1); + CountDownLatch secondRenewed = new CountDownLatch(1); + when(api.renewCommitLock(identifier, "table-id", "morax-job")) + .thenAnswer( + invocation -> { + firstStarted.countDown(); + try { + assertThat(resumeFirst.await(10, TimeUnit.SECONDS)).isTrue(); + } catch (InterruptedException e) { + firstInterrupted.countDown(); + throw e; + } + return new CommitLockResponse(true, "morax-job", 100000, 60000, null); + }); + when(api.renewCommitLock(identifier, "table-id", "other-job")) + .thenAnswer( + invocation -> { + secondRenewed.countDown(); + return new CommitLockResponse(true, "other-job", 100000, 60000, null); + }); + ExecutorService sharedWorkers = Executors.newCachedThreadPool(); + RESTCatalogLock firstLock = + new RESTCatalogLock( + api, + identifier, + "table-id", + "morax-job", + 0, + renewer, + sharedWorkers, + clock::get); + RESTCatalogLock secondLock = + new RESTCatalogLock( + api, + identifier, + "table-id", + "other-job", + 0, + renewer, + sharedWorkers, + clock::get); + try (AutoCloseable first = firstLock.acquire(); + AutoCloseable second = secondLock.acquire()) { + ArgumentCaptor ticks = ArgumentCaptor.forClass(Runnable.class); + verify(renewer, times(2)) + .scheduleWithFixedDelay( + ticks.capture(), eq(20000L), eq(20000L), eq(TimeUnit.MILLISECONDS)); + Runnable firstTick = ticks.getAllValues().get(0); + firstTick.run(); + assertThat(firstStarted.await(10, TimeUnit.SECONDS)).isTrue(); + firstTick.run(); + firstTick.run(); + ticks.getAllValues().get(1).run(); + assertThat(secondRenewed.await(10, TimeUnit.SECONDS)).isTrue(); + first.close(); + assertThat(firstInterrupted.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(sharedWorkers.isShutdown()).isFalse(); + secondLock.ensureValid(); + firstTick.run(); + verify(api).renewCommitLock(identifier, "table-id", "morax-job"); + verify(api).renewCommitLock(identifier, "table-id", "other-job"); + verify(secondRenewal, never()).cancel(anyBoolean()); + } finally { + resumeFirst.countDown(); + sharedWorkers.shutdownNow(); + assertThat(sharedWorkers.awaitTermination(10, TimeUnit.SECONDS)).isTrue(); + } + } + + @Test + void nestedPublicationReusesLeaseAndRechecksValidity() throws Exception { + grant("morax-job", 60000); + Lock publication = Lock.reentrant(lock); + assertThat(publication.runWithLock(() -> publication.runWithLock(() -> "published"))) + .isEqualTo("published"); + verify(api).acquireCommitLock(identifier, "table-id", "morax-job"); + verify(future).cancel(false); + publication.runWithLock( + () -> { + clock.addAndGet(TimeUnit.MILLISECONDS.toNanos(60000)); + assertThatThrownBy( + () -> + publication.runWithLock( + () -> { + throw new AssertionError(); + })) + .hasMessageContaining("no longer usable"); + return null; + }); + verify(api, times(2)).acquireCommitLock(identifier, "table-id", "morax-job"); + } + + @Test + void closingAnOldScopeTwiceCannotInvalidateTheNewLease() throws Exception { + grant("morax-job", 60000); + AutoCloseable first = lock.acquire(); + first.close(); + try (AutoCloseable second = lock.acquire()) { + first.close(); + lock.ensureValid(); + } + verify(api, times(2)).acquireCommitLock(identifier, "table-id", "morax-job"); + verify(future, times(2)).cancel(false); + } + + @Test + void interruptedContentionStopsAcquisition() { + grant("morax-job", 60000); + when(api.acquireCommitLock(identifier, "table-id", "morax-job")) + .thenReturn(CommitLockResponse.unavailable()); + RESTCatalogLock waiting = + new RESTCatalogLock( + api, + identifier, + "table-id", + "morax-job", + 1000, + renewer, + workers, + clock::get); + Thread.currentThread().interrupt(); + try { + assertThatThrownBy( + () -> + waiting.runWithLock( + () -> { + throw new AssertionError(); + })) + .isInstanceOf(InterruptedException.class); + assertThat(Thread.currentThread().isInterrupted()).isTrue(); + } finally { + Thread.interrupted(); + } + verifyNoInteractions(renewer); + } + + private T run(Callable callable) throws Exception { + return lock.runWithLock(callable); + } + + private void grant(String owner, long leaseMillis) { + doAnswer( + invocation -> { + ((Runnable) invocation.getArgument(0)).run(); + return null; + }) + .when(workers) + .execute(any(Runnable.class)); + when(api.acquireCommitLock(identifier, "table-id", "morax-job")) + .thenReturn(new CommitLockResponse(true, owner, 100000, leaseMillis, null)); + when(renewer.scheduleWithFixedDelay(any(Runnable.class), anyLong(), anyLong(), any())) + .thenAnswer(invocation -> future); + } + + private Runnable renewal() { + return renewal(1); + } + + private Runnable renewal(int scopes) { + ArgumentCaptor callback = ArgumentCaptor.forClass(Runnable.class); + verify(renewer, times(scopes)) + .scheduleWithFixedDelay( + callback.capture(), eq(20000L), eq(20000L), eq(TimeUnit.MILLISECONDS)); + return callback.getValue(); + } + + private List queueRenewals() { + List requests = new ArrayList<>(); + doAnswer( + invocation -> { + requests.add(invocation.getArgument(0)); + return null; + }) + .when(workers) + .execute(any(Runnable.class)); + return requests; + } + + private static void awaitIgnoringInterrupts(CountDownLatch latch) { + while (true) { + try { + assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + return; + } catch (InterruptedException ignored) { + // Emulate a transport that completes its RPC despite scope cancellation. + } + } + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java index b4afe6adaff4..98449676be11 100644 --- a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java +++ b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java @@ -727,12 +727,14 @@ && isTableByIdRequest(request.getPath())) { if (!tableMetadataStore.containsKey(identifier.getFullName())) { throw new Catalog.TableNotExistException(identifier); } + MockRESTMessage.validateRollbackHead( + requestBody, + tableLatestSnapshotStore.get(identifier.getFullName())); if (requestBody.getInstant() instanceof Instant.SnapshotInstant) { long snapshotId = ((Instant.SnapshotInstant) requestBody.getInstant()) .getSnapshotId(); - return rollbackTableByIdHandle( - identifier, snapshotId, requestBody.getFromSnapshot()); + return rollbackTableByIdHandle(identifier, snapshotId); } else if (requestBody.getInstant() instanceof Instant.TagInstant) { String tagName = ((Instant.TagInstant) requestBody.getInstant()) @@ -1195,8 +1197,8 @@ private MockResponse commitTableHandle(Identifier identifier, String data) throw requestBody.getStatistics()); } - private MockResponse rollbackTableByIdHandle( - Identifier identifier, long snapshotId, @Nullable Long fromSnapshot) throws Exception { + private MockResponse rollbackTableByIdHandle(Identifier identifier, long snapshotId) + throws Exception { FileStoreTable table = getFileTable(identifier); String identifierWithSnapshotId = geTableFullNameWithSnapshotId(identifier, snapshotId); TableSnapshot toSnapshot = tableWithSnapshotId2SnapshotStore.get(identifierWithSnapshotId); @@ -1207,16 +1209,6 @@ private MockResponse rollbackTableByIdHandle( 404); } long latestSnapshotId = table.snapshotManager().latestSnapshotId(); - if (fromSnapshot != null && fromSnapshot != latestSnapshotId) { - return mockResponse( - new ErrorResponse( - null, - null, - String.format( - "Latest snapshot %s is not %s", latestSnapshotId, fromSnapshot), - 500), - 500); - } table = table.copy( Collections.singletonMap(SNAPSHOT_CLEAN_EMPTY_DIRECTORIES.key(), "true")); diff --git a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java index 67106e1616a1..9a5e1af6ed8a 100644 --- a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java @@ -2378,9 +2378,22 @@ public void testTableRollback() throws Exception { // rollback to snapshot 2 from snapshot assertThatThrownBy( - () -> catalog.rollbackTo(identifier, null, Instant.snapshot(2L), 4L, null)) + () -> + catalog.rollbackTo( + identifier, + null, + Instant.snapshot(2L), + 3L, + "stale-head-uuid", + "writer")) + .hasMessageContaining("Latest snapshot UUID changed"); + assertThat(table.latestSnapshot().get().id()).isEqualTo(3); + assertThatThrownBy( + () -> + catalog.rollbackTo( + identifier, null, Instant.snapshot(2L), 4L, null, null)) .hasMessageContaining("Latest snapshot 3 is not 4"); - catalog.rollbackTo(identifier, null, Instant.snapshot(2L), 3L, null); + catalog.rollbackTo(identifier, null, Instant.snapshot(2L), 3L, null, null); assertThat(table.latestSnapshot().get().id()).isEqualTo(2); } diff --git a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCommitLockIntegrationTest.java b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCommitLockIntegrationTest.java new file mode 100644 index 000000000000..e6a236283f05 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCommitLockIntegrationTest.java @@ -0,0 +1,481 @@ +/* + * 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.paimon.rest; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.Snapshot; +import org.apache.paimon.append.AppendCompactTask; +import org.apache.paimon.catalog.CatalogContext; +import org.apache.paimon.catalog.CatalogLockContext; +import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.io.CompactIncrement; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.io.DataIncrement; +import org.apache.paimon.manifest.ManifestCommittable; +import org.apache.paimon.operation.BaseAppendFileStoreWrite; +import org.apache.paimon.operation.FileStoreCommit; +import org.apache.paimon.operation.FileStoreWrite; +import org.apache.paimon.operation.Lock; +import org.apache.paimon.options.Options; +import org.apache.paimon.reader.RecordReader; +import org.apache.paimon.rest.requests.CommitLockRequest; +import org.apache.paimon.rest.requests.CommitTableRequest; +import org.apache.paimon.rest.requests.RollbackTableRequest; +import org.apache.paimon.rest.responses.CommitLockResponse; +import org.apache.paimon.rest.responses.GetTableResponse; +import org.apache.paimon.rest.responses.GetTableSnapshotResponse; +import org.apache.paimon.schema.FileSystemSchemaManager; +import org.apache.paimon.schema.Schema; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.Instant; +import org.apache.paimon.table.TableSnapshot; +import org.apache.paimon.table.sink.BatchTableCommit; +import org.apache.paimon.table.sink.BatchTableWrite; +import org.apache.paimon.table.sink.BatchWriteBuilder; +import org.apache.paimon.table.sink.CommitMessage; +import org.apache.paimon.table.sink.CommitMessageImpl; +import org.apache.paimon.table.sink.TableWriteImpl; +import org.apache.paimon.table.source.ReadBuilder; +import org.apache.paimon.types.DataTypes; + +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.JsonProcessingException; + +import okhttp3.mockwebserver.Dispatcher; +import okhttp3.mockwebserver.MockResponse; +import okhttp3.mockwebserver.MockWebServer; +import okhttp3.mockwebserver.RecordedRequest; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; +import org.junit.jupiter.params.provider.ValueSource; + +import java.nio.file.Path; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import static org.apache.paimon.data.BinaryRow.EMPTY_ROW; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Exercises REST leases through the real file-store commit path. */ +class RESTCommitLockIntegrationTest { + + @Test + void disabledFactoryDoesNotRequireRestConfigurationOrWriterIdentity() { + Lock lock = + new RESTCatalogLockFactory() + .createLock( + CatalogLockContext.fromOptions(new Options()), + null, + null, + null, + new Options()); + assertThat(lock).isInstanceOf(Lock.EmptyLock.class); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void tableCapabilityDoesNotMakeAClientAcquireByDefault(boolean capability) { + Options options = new Options(); + options.set(CoreOptions.REST_COMMIT_LOCK_ENABLED, capability); + Lock lock = + new RESTCatalogLockFactory() + .createLock( + CatalogLockContext.fromOptions(new Options()), + null, + null, + null, + options); + assertThat(lock).isInstanceOf(Lock.EmptyLock.class); + } + + @Test + void acquisitionRequiresTheTableCapabilityBeforeRestRequests() { + Options options = new Options(); + options.set("rest.commit.acquire-lock", "true"); + assertThatThrownBy( + () -> + new RESTCatalogLockFactory() + .createLock( + CatalogLockContext.fromOptions(new Options()), + null, + null, + null, + options)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("rest.commit.lock-enabled"); + } + + @ParameterizedTest + @CsvSource({ + "false,false,false", "false,false,true", "true,false,false", + "true,false,true", "true,true,false", "true,true,true" + }) + void commitsAcquireBeforeHeadAndReacquireAfterConflict( + boolean capability, + boolean locking, + boolean firstPublicationConflicts, + @TempDir Path directory) + throws Exception { + Options tableOptions = new Options(); + tableOptions.set(CoreOptions.REST_COMMIT_LOCK_ENABLED, capability); + tableOptions.set(CoreOptions.REST_COMMIT_ACQUIRE_LOCK, locking); + tableOptions.set(CoreOptions.COMMIT_MAX_RETRIES, 2); + tableOptions.set(CoreOptions.COMMIT_MIN_RETRY_WAIT, Duration.ofMillis(1)); + tableOptions.set(CoreOptions.COMMIT_MAX_RETRY_WAIT, Duration.ofMillis(1)); + Schema schema = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .options(tableOptions.toMap()) + .build(); + GetTableResponse tableResponse = + new GetTableResponse( + "table-id", + "db", + "table$branch_dev", + directory.toUri().toString(), + false, + 0, + schema, + null, + 0, + null, + 0, + null); + List events = Collections.synchronizedList(new ArrayList<>()); + AtomicReference head = new AtomicReference<>(); + AtomicBoolean lease = new AtomicBoolean(); + AtomicInteger publications = new AtomicInteger(); + try (MockWebServer server = new MockWebServer()) { + server.setDispatcher( + new Dispatcher() { + @Override + public MockResponse dispatch(RecordedRequest request) { + String path = request.getPath(); + assertThat(request.getRequestUrl().pathSegments()) + .contains("table$branch_dev"); + if (path.endsWith("/commit-lock")) { + assertThat(locking).isTrue(); + CommitLockRequest acquire = + fromJson(request, CommitLockRequest.class); + assertThat(acquire.getTableId()).isEqualTo("table-id"); + assertThat(acquire.getCommitUser()).isEqualTo("morax-job"); + events.add("acquire"); + lease.set(true); + return json( + new CommitLockResponse( + true, "morax-job", 100000, 60000, head.get())); + } + if (path.endsWith("/snapshot")) { + if (locking) { + assertThat(lease.get()).isTrue(); + } + events.add("head"); + Snapshot snapshot = head.get(); + return json( + new GetTableSnapshotResponse( + snapshot == null + ? null + : new TableSnapshot(snapshot, 0, 0, 0, 0))); + } + if (path.endsWith("/commit")) { + if (locking) { + assertThat(lease.get()).isTrue(); + } + CommitTableRequest commit = + fromJson(request, CommitTableRequest.class); + assertThat(commit.getTableId()).isEqualTo("table-id"); + assertThat(commit.getSnapshot().commitUser()) + .isEqualTo("morax-job"); + Snapshot previous = head.get(); + assertThat(commit.getBaseSnapshotUuid()) + .isEqualTo(previous == null ? null : previous.uuid()); + events.add("publish"); + if (firstPublicationConflicts + && publications.getAndIncrement() == 0) { + return json(Collections.singletonMap("success", false)); + } + head.set(commit.getSnapshot()); + lease.set(false); + return json(Collections.singletonMap("success", true)); + } + if ("GET".equals(request.getMethod()) && !path.contains("/schemas/")) { + return json(tableResponse); + } + return new MockResponse().setResponseCode(400); + } + }); + server.start(); + Options catalogOptions = new Options(); + catalogOptions.set(RESTCatalogOptions.URI, server.url("/").toString()); + catalogOptions.set(RESTCatalogOptions.TOKEN_PROVIDER, "bear"); + catalogOptions.set(RESTCatalogOptions.TOKEN, "token"); + catalogOptions.set(RESTTokenFileIO.DATA_TOKEN_ENABLED, false); + catalogOptions.set(RESTCatalogInternalOptions.PREFIX, "catalog"); + try (RESTCatalog catalog = + new RESTCatalog(CatalogContext.create(catalogOptions), false)) { + FileStoreTable table = + (FileStoreTable) catalog.getTable(new Identifier("db", "table", "dev")); + new FileSystemSchemaManager(table.fileIO(), table.location(), "dev") + .createTable(schema); + try (FileStoreCommit commit = + table.store().newCommit("morax-job", table).ignoreEmptyCommit(false)) { + assertThat(commit.commit(new ManifestCommittable(1L), false)).isEqualTo(1); + assertThat(head.get().id()).isEqualTo(1L); + assertThat(events) + .containsExactlyElementsOf( + expected(locking, firstPublicationConflicts)); + events.clear(); + assertThat(commit.commit(new ManifestCommittable(2L), false)).isEqualTo(1); + assertThat(head.get().id()).isEqualTo(2L); + assertThat(events).containsExactlyElementsOf(expected(locking, false)); + } + } + } + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void compactionConflictRollsBackWithWriterContextAndReacquires( + boolean locking, @TempDir Path directory) throws Exception { + Schema schema = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .option(CoreOptions.REST_COMMIT_LOCK_ENABLED.key(), "true") + .option(CoreOptions.REST_COMMIT_ACQUIRE_LOCK.key(), String.valueOf(locking)) + .option(CoreOptions.COMMIT_MIN_RETRY_WAIT.key(), "1 ms") + .option(CoreOptions.COMMIT_MAX_RETRY_WAIT.key(), "1 ms") + .build(); + GetTableResponse tableResponse = + new GetTableResponse( + "branch-id", + "db", + "table$branch_dev", + directory.toUri().toString(), + false, + 0, + schema, + null, + 0, + null, + 0, + null); + AtomicReference head = new AtomicReference<>(); + Map snapshots = new ConcurrentHashMap<>(); + AtomicReference owner = new AtomicReference<>(); + AtomicReference rollback = new AtomicReference<>(); + List events = Collections.synchronizedList(new ArrayList<>()); + try (MockWebServer server = new MockWebServer()) { + server.setDispatcher( + new Dispatcher() { + @Override + public MockResponse dispatch(RecordedRequest request) { + String path = request.getPath(); + assertThat(request.getRequestUrl().pathSegments()) + .contains("table$branch_dev"); + if (path.endsWith("/commit-lock")) { + CommitLockRequest acquire = + fromJson(request, CommitLockRequest.class); + assertThat(acquire.getTableId()).isEqualTo("branch-id"); + assertThat(locking).isTrue(); + owner.set(acquire.getCommitUser()); + events.add("acquire"); + return json( + new CommitLockResponse( + true, owner.get(), 100000, 60000, head.get())); + } + if (path.endsWith("/snapshot")) { + events.add("head"); + Snapshot snapshot = head.get(); + return json( + new GetTableSnapshotResponse( + snapshot == null + ? null + : new TableSnapshot(snapshot, 0, 0, 0, 0))); + } + if (path.endsWith("/rollback")) { + RollbackTableRequest requestBody = + fromJson(request, RollbackTableRequest.class); + assertThat(requestBody.getTableId()).isEqualTo("branch-id"); + assertThat(requestBody.getFromSnapshot()) + .isEqualTo(head.get().id()); + if (locking) { + assertThat(owner.get()).isNotNull(); + assertThat(requestBody.getCommitUser()).isEqualTo(owner.get()); + } + head.set( + snapshots.get( + ((Instant.SnapshotInstant) requestBody.getInstant()) + .getSnapshotId())); + rollback.set(requestBody); + owner.set(null); + events.add("rollback"); + return json(Collections.emptyMap()); + } + if (path.endsWith("/commit")) { + CommitTableRequest commit = + fromJson(request, CommitTableRequest.class); + assertThat(commit.getTableId()).isEqualTo("branch-id"); + if (locking) { + assertThat(commit.getSnapshot().commitUser()) + .isEqualTo(owner.get()); + } + Snapshot previous = head.get(); + assertThat(commit.getBaseSnapshotUuid()) + .isEqualTo(previous == null ? null : previous.uuid()); + head.set(commit.getSnapshot()); + snapshots.put(commit.getSnapshot().id(), commit.getSnapshot()); + owner.set(null); + events.add("publish"); + return json(Collections.singletonMap("success", true)); + } + if ("GET".equals(request.getMethod()) && !path.contains("/schemas/")) { + return json(tableResponse); + } + return new MockResponse().setResponseCode(400); + } + }); + server.start(); + Options catalogOptions = new Options(); + catalogOptions.set(RESTCatalogOptions.URI, server.url("/").toString()); + catalogOptions.set(RESTCatalogOptions.TOKEN_PROVIDER, "bear"); + catalogOptions.set(RESTCatalogOptions.TOKEN, "token"); + catalogOptions.set(RESTTokenFileIO.DATA_TOKEN_ENABLED, false); + catalogOptions.set(RESTCatalogInternalOptions.PREFIX, "catalog"); + try (RESTCatalog catalog = + new RESTCatalog(CatalogContext.create(catalogOptions), false)) { + FileStoreTable table = + (FileStoreTable) catalog.getTable(new Identifier("db", "table", "dev")); + new FileSystemSchemaManager(table.fileIO(), table.location(), "dev") + .createTable(schema); + BatchWriteBuilder builder = table.newBatchWriteBuilder(); + List files = new ArrayList<>(); + for (int i = 0; i < 5; i++) { + try (BatchTableWrite write = builder.newWrite(); + BatchTableCommit commit = builder.newCommit()) { + write.write(GenericRow.of(i)); + List messages = write.prepareCommit(); + commit.commit(messages); + files.add( + ((CommitMessageImpl) messages.get(0)) + .newFilesIncrement() + .newFiles() + .get(0)); + } + } + CommitMessageImpl deletion = + new CommitMessageImpl( + EMPTY_ROW, + 0, + -1, + new DataIncrement( + Collections.emptyList(), + Collections.singletonList(files.get(0)), + Collections.emptyList()), + new CompactIncrement( + Collections.emptyList(), + Collections.emptyList(), + Collections.emptyList())); + CommitMessage compacted; + try (BatchTableWrite write = builder.newWrite()) { + FileStoreWrite storeWrite = ((TableWriteImpl) write).getWrite(); + compacted = + new AppendCompactTask(EMPTY_ROW, files) + .doCompact(table, (BaseAppendFileStoreWrite) storeWrite); + } + try (BatchTableCommit commit = builder.newCommit()) { + commit.commit(Collections.singletonList(compacted)); + } + Snapshot compactHead = head.get(); + assertThat(compactHead.id()).isEqualTo(6L); + assertThat(compactHead.commitKind()).isEqualTo(Snapshot.CommitKind.COMPACT); + events.clear(); + try (BatchTableCommit commit = builder.newCommit()) { + commit.commit(Collections.singletonList(deletion)); + } + Snapshot replacement = head.get(); + assertThat(replacement.id()).isEqualTo(6L); + assertThat(replacement.uuid()).isNotEqualTo(compactHead.uuid()); + assertThat(replacement.commitKind()).isEqualTo(Snapshot.CommitKind.OVERWRITE); + assertThat(rollback.get().getFromSnapshot()).isEqualTo(6L); + assertThat(rollback.get().getFromSnapshotUuid()).isEqualTo(compactHead.uuid()); + assertThat(((Instant.SnapshotInstant) rollback.get().getInstant()).getSnapshotId()) + .isEqualTo(5L); + assertThat(rollback.get().getCommitUser()).isEqualTo(replacement.commitUser()); + if (locking) { + assertThat(events) + .containsSubsequence( + "acquire", "head", "rollback", "acquire", "head", "publish"); + } else { + assertThat(events) + .containsSubsequence("head", "rollback", "head", "publish") + .doesNotContain("acquire"); + } + List values = new ArrayList<>(); + ReadBuilder read = table.newReadBuilder(); + try (RecordReader reader = + read.newRead().createReader(read.newScan().plan())) { + reader.forEachRemaining(row -> values.add(row.getInt(0))); + } + assertThat(values).containsExactlyInAnyOrder(1, 2, 3, 4); + } + } + } + + private static List expected(boolean locking, boolean retry) { + List events = new ArrayList<>(); + for (int attempt = 0; attempt < (retry ? 2 : 1); attempt++) { + if (locking) { + events.add("acquire"); + } + events.add("head"); + events.add("publish"); + } + return events; + } + + private static T fromJson(RecordedRequest request, Class type) { + try { + return RESTApi.fromJson(request.getBody().readUtf8(), type); + } catch (JsonProcessingException e) { + throw new AssertionError("Invalid request JSON", e); + } + } + + private static MockResponse json(Object body) { + try { + return new MockResponse() + .setResponseCode(200) + .setBody(RESTApi.toJson(body)) + .addHeader("Content-Type", "application/json"); + } catch (JsonProcessingException e) { + throw new AssertionError("Invalid response JSON", e); + } + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/rest/RollbackProtocolTest.java b/paimon-core/src/test/java/org/apache/paimon/rest/RollbackProtocolTest.java index b26aad689960..4e710c0f1cf9 100644 --- a/paimon-core/src/test/java/org/apache/paimon/rest/RollbackProtocolTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/rest/RollbackProtocolTest.java @@ -72,7 +72,7 @@ public CatalogLoader catalogLoader() { .setBody("{}")); String tableUuid = "without-uuid".equals(context) ? null : "table-id"; String commitUser = "owned".equals(context) ? "Writer-1" : null; - catalog.rollbackTo(identifier, tableUuid, instant, 2L, commitUser); + catalog.rollbackTo(identifier, tableUuid, instant, 2L, "head-uuid", commitUser); RecordedRequest recorded = server.takeRequest(10, TimeUnit.SECONDS); assertThat(recorded.getMethod()).isEqualTo("POST"); assertThat(recorded.getPath()) @@ -82,6 +82,7 @@ public CatalogLoader catalogLoader() { assertThat(payload.get("instant").get("type").asText()).isEqualTo(kind); RollbackTableRequest request = RESTApi.fromJson(body, RollbackTableRequest.class); assertThat(request.getFromSnapshot()).isEqualTo(2L); + assertThat(request.getFromSnapshotUuid()).isEqualTo("head-uuid"); if ("snapshot".equals(kind)) { assertThat(((Instant.SnapshotInstant) request.getInstant()).getSnapshotId()) .isEqualTo(1L); @@ -100,7 +101,7 @@ public CatalogLoader catalogLoader() { assertThat(payload.has("commitUser")).isFalse(); } assertThat(payload.size()) - .isEqualTo(2 + (tableUuid == null ? 0 : 1) + (commitUser == null ? 0 : 1)); + .isEqualTo(3 + (tableUuid == null ? 0 : 1) + (commitUser == null ? 0 : 1)); } } } diff --git a/paimon-core/src/test/java/org/apache/paimon/table/CatalogEnvironmentTest.java b/paimon-core/src/test/java/org/apache/paimon/table/CatalogEnvironmentTest.java index 1b18b7039b4a..3489b8e50604 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/CatalogEnvironmentTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/CatalogEnvironmentTest.java @@ -80,10 +80,15 @@ void testCatalogTableRollbackUsesRuntimeBranchAndOwner(String branch) throws Exc true, false); Instant instant = Instant.snapshot(1L); - environment.catalogTableRollback(branch, "writer").rollbackTo(instant, 2L); + environment.catalogTableRollback(branch, "writer").rollbackTo(instant, 2L, "head-uuid"); verify(catalog) .rollbackTo( - new Identifier("db", "table", branch), "table-id", instant, 2L, "writer"); + new Identifier("db", "table", branch), + "table-id", + instant, + 2L, + "head-uuid", + "writer"); } @ParameterizedTest @@ -102,9 +107,15 @@ void testCatalogTableRollbackWithoutOwnerPreservesTableUuid(String branch, Strin true, false); Instant instant = Instant.tag("tag-1"); - environment.catalogTableRollback(null, null).rollbackTo(instant, null); + environment.catalogTableRollback(null, null).rollbackTo(instant, null, null); verify(catalog) - .rollbackTo(new Identifier("db", "table", branch), tableUuid, instant, null, null); + .rollbackTo( + new Identifier("db", "table", branch), + tableUuid, + instant, + null, + null, + null); } @ParameterizedTest @@ -131,12 +142,12 @@ void testCopiedEnvironmentRollbackUsesTargetIdentity( CatalogEnvironment copied = environment.copy(targetIdentifier); Instant snapshot = Instant.snapshot(1L); - copied.catalogTableRollback(null, null).rollbackTo(snapshot, 2L); - verify(catalog).rollbackTo(targetIdentifier, expectedUuid, snapshot, 2L, null); + copied.catalogTableRollback(null, null).rollbackTo(snapshot, 2L, "head-uuid"); + verify(catalog).rollbackTo(targetIdentifier, expectedUuid, snapshot, 2L, "head-uuid", null); Instant tag = Instant.tag("checkpoint"); copied.snapshotLoader().rollback(tag); - verify(catalog).rollbackTo(targetIdentifier, expectedUuid, tag, null, null); + verify(catalog).rollbackTo(targetIdentifier, expectedUuid, tag, null, null, null); assertThat(environment.uuid()).isEqualTo(sourceUuid); } diff --git a/paimon-core/src/test/java/org/apache/paimon/table/FallbackReadFileStoreTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/FallbackReadFileStoreTableTest.java index 78ac14e2683c..213de33561d1 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/FallbackReadFileStoreTableTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/FallbackReadFileStoreTableTest.java @@ -579,6 +579,7 @@ void testBranchViewsRollbackWithoutSourceUuid(String branchOption) throws Except .getSnapshotId() == 1L), Mockito.isNull(), + Mockito.isNull(), Mockito.isNull()); table.switchToBranch("bc").rollbackTo("checkpoint"); @@ -594,6 +595,7 @@ void testBranchViewsRollbackWithoutSourceUuid(String branchOption) throws Except ((Instant.TagInstant) instant) .getTagName())), Mockito.isNull(), + Mockito.isNull(), Mockito.isNull()); assertThat(table.catalogEnvironment().uuid()).isEqualTo("main-uuid"); } diff --git a/paimon-core/src/test/java/org/apache/paimon/tag/SnapshotLoaderImplTest.java b/paimon-core/src/test/java/org/apache/paimon/tag/SnapshotLoaderImplTest.java index 772564281b18..0346690bad54 100644 --- a/paimon-core/src/test/java/org/apache/paimon/tag/SnapshotLoaderImplTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/tag/SnapshotLoaderImplTest.java @@ -43,7 +43,7 @@ void testRollbackPreservesTableUuid(String branch, String uuid) throws Exception environment.snapshotLoader().rollback(instant); - verify(catalog).rollbackTo(identifier, uuid, instant, null, null); + verify(catalog).rollbackTo(identifier, uuid, instant, null, null, null); } @ParameterizedTest @@ -62,7 +62,12 @@ void testCopyWithinBranchKeepsTableUuid(String branch) throws Exception { verify(catalog) .rollbackTo( - new Identifier("db", "table", branch), "table-uuid", instant, null, null); + new Identifier("db", "table", branch), + "table-uuid", + instant, + null, + null, + null); } @ParameterizedTest @@ -78,7 +83,7 @@ void testCopyToAnotherBranchDoesNotReuseSourceUuid(String source, String target) loader.copyWithBranch(target).rollback(instant); verify(catalog) - .rollbackTo(new Identifier("db", "table", target), null, instant, null, null); + .rollbackTo(new Identifier("db", "table", target), null, instant, null, null, null); } private CatalogEnvironment environment(Catalog catalog, Identifier identifier, String uuid) {