github-actions[bot] commented on code in PR #65670:
URL: https://github.com/apache/doris/pull/65670#discussion_r3592854751
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/Resource.java:
##########
@@ -350,12 +357,46 @@ public Resource clone() {
return copied;
}
- private void notifyUpdate(Map<String, String> properties) {
-
references.entrySet().stream().collect(Collectors.groupingBy(Entry::getValue)).forEach((type,
refs) -> {
- if (type == ReferenceType.CATALOG) {
- // No longer support resource in Catalog.
+ final Resource getCopiedResourceSnapshot() {
+ String json;
+ readLock();
Review Comment:
[P1] Pair this snapshot lock with every mutable resource writer
This read lock only excludes mutations that take `Resource.writeLock()`.
`HMSResource.modifyProperties()`, `OdbcCatalogResource.modifyProperties()`, and
`JdbcResource.modifyProperties()` all write their serialized `HashMap`s without
that lock, and all three types are reachable through `ALTER RESOURCE`. A live
`ResourceMgr.write()` can therefore hold this read lock while one of those maps
changes, so Gson can emit a mixed image or fail with
`ConcurrentModificationException`. HDFS is paired now, but the new generic
snapshot guarantee is still false for these parallel subtype paths. Please make
those writers/readers use the same lock protocol (or provide subtype snapshot
hooks) and add an image-versus-ALTER test for at least one currently unlocked
type.
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/ResourceMgr.java:
##########
@@ -145,20 +181,40 @@ public void replayDropResource(DropResourceOperationLog
operationLog) {
}
public void alterResource(String resourceName, Map<String, String>
properties) throws DdlException {
- if (!nameToResource.containsKey(resourceName)) {
+ Resource resource = nameToResource.get(resourceName);
+ if (resource == null) {
throw new DdlException("Resource(" + resourceName + ") dose not
exist.");
}
- Resource resource = nameToResource.get(resourceName);
- resource.modifyProperties(properties);
+ Resource logResource;
+ EditLog.EditLogItem logItem;
+ synchronized (resource.getAlterLock()) {
+ if (nameToResource.get(resourceName) != resource) {
+ throw new DdlException("Resource(" + resourceName + ") changed
concurrently");
+ }
+ resource.modifyProperties(properties);
Review Comment:
[P1] Alter the same Resource instance that was validated
`AlterResourceCommand.validate()` fetches one Resource and runs that
subtype's `checkProperties()`, but then calls this manager by name and this
method performs a fresh lookup. A DROP/recreate between those calls means the
new identity check only protects the replacement after the second lookup; it
does not prove this is the object that was validated. For example, an ODBC
`host` change can validate against ODBC A, then be applied and journaled on a
same-name HDFS B, whose `modifyProperties()` writes the foreign nonempty key
without revalidating it. Pass the validated identity through and recheck it
under `alterLock`, or re-run validation on the locked current object, with a
cross-type replacement test.
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/ResourceMgr.java:
##########
@@ -145,20 +181,40 @@ public void replayDropResource(DropResourceOperationLog
operationLog) {
}
public void alterResource(String resourceName, Map<String, String>
properties) throws DdlException {
- if (!nameToResource.containsKey(resourceName)) {
+ Resource resource = nameToResource.get(resourceName);
+ if (resource == null) {
throw new DdlException("Resource(" + resourceName + ") dose not
exist.");
}
- Resource resource = nameToResource.get(resourceName);
- resource.modifyProperties(properties);
+ Resource logResource;
+ EditLog.EditLogItem logItem;
+ synchronized (resource.getAlterLock()) {
+ if (nameToResource.get(resourceName) != resource) {
+ throw new DdlException("Resource(" + resourceName + ") changed
concurrently");
+ }
+ resource.modifyProperties(properties);
+ // Keep mutation, snapshot, and submission ordered without holding
the resource state lock.
+ logResource = getAlterLogResource(resource);
+ logItem =
Env.getCurrentEnv().getEditLog().submitEdit(OperationType.OP_ALTER_RESOURCE,
logResource);
+ }
- // log alter
- Env.getCurrentEnv().getEditLog().logAlterResource(resource);
- LOG.info("Alter resource success. Resource: {}", resource);
+ logItem.await();
+ LOG.info("Alter resource success. Resource: {}", logResource);
}
public void replayAlterResource(Resource resource) {
- nameToResource.put(resource.getName(), resource);
+ Resource replayed =
nameToResource.computeIfPresent(resource.getName(), (name, currentResource) -> {
Review Comment:
[P1] Preserve ALTERs for deferred restore-created resources
Absence during replay does not always mean the ALTER is stale. The earlier
PENDING `RestoreJob` record is durable, but ordinary replay only re-registers
that deferred job; it does not run the resource creation. Live PENDING
execution then publishes an ODBC resource through `createResource(Resource,
false)` without a new creation-bearing state log. That visible resource can be
altered and the ALTER acknowledged; if the FE crashes before the next
restore-state log, replay reaches this method with the resource absent and
discards the ALTER, then the resumed job recreates the original backup
resource. Please journal/register restore-created resources before exposing
them to ALTER, or defer and later apply the matching ALTER; add a recovery test
for this PENDING window.
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/ResourceMgr.java:
##########
@@ -108,35 +125,54 @@ public void replayCreateResource(Resource resource) {
public void dropResource(DropResourceCommand dropResourceCommand) throws
DdlException {
String resourceName = dropResourceCommand.getResourceName();
- if (!nameToResource.containsKey(resourceName)) {
+ Resource resource = nameToResource.get(resourceName);
+ if (resource == null) {
if (dropResourceCommand.isIfExists()) {
return;
}
throw new DdlException("Resource(" + resourceName + ") does not
exist");
}
- Resource resource = nameToResource.get(resourceName);
- resource.dropResource();
-
- // Check whether the resource is in use before deleting it, except
spark resource
- StoragePolicy checkedStoragePolicy = StoragePolicy.ofCheck(null);
- checkedStoragePolicy.setStorageResource(resourceName);
- if
(Env.getCurrentEnv().getPolicyMgr().existPolicy(checkedStoragePolicy)) {
- Policy policy =
Env.getCurrentEnv().getPolicyMgr().getPolicy(checkedStoragePolicy);
- LOG.warn("Can not drop resource, since it's used in policy {}",
policy.getPolicyName());
- throw new DdlException("Can not drop resource, since it's used in
policy " + policy.getPolicyName());
+ EditLog.EditLogItem logItem;
+ synchronized (resource.getAlterLock()) {
+ if (nameToResource.get(resourceName) != resource) {
+ if (dropResourceCommand.isIfExists() &&
!nameToResource.containsKey(resourceName)) {
+ return;
+ }
+ throw new DdlException("Resource(" + resourceName + ") changed
concurrently");
+ }
+
+ resource.dropResource();
Review Comment:
[P1] Serialize policy reference acquisition with resource DROP
The reference check here is synchronized only for the duration of
`Resource.dropResource()`, while `StoragePolicy.init()` adds its reference
before the later `PolicyMgr.createPolicy()` publish/journal. A concrete race
is: DROP observes no references; CREATE POLICY adds `p@POLICY` and pauses
before PolicyMgr insertion; DROP's policy lookup still sees nothing, submits
the DROP, and removes the resource; CREATE POLICY then journals successfully.
Replay leaves a durable policy whose resource is absent. This is distinct from
ALTER payload reference ownership. Please make policy reference
acquisition/policy journaling and resource DROP share one lifecycle ordering
protocol with an identity recheck, and cover this race with latches.
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/ResourceMgr.java:
##########
@@ -82,23 +84,38 @@ public void createResource(CreateResourceCommand command)
throws DdlException {
"Only support SPARK, ODBC_CATALOG, JDBC, S3_COOLDOWN, S3,
HDFS(JFS/JUICEFS), and HMS resource.");
}
Resource resource = Resource.fromCommand(command);
- if (createResource(resource, info.isIfNotExists())) {
- Env.getCurrentEnv().getEditLog().logCreateResource(resource);
- LOG.info("Create resource success. Resource: {}",
resource.getName());
+ Resource logResource;
+ EditLog.EditLogItem logItem;
+ synchronized (nameToResource) {
+ if (nameToResource.containsKey(resource.getName())) {
+ if (info.isIfNotExists()) {
+ return;
+ }
+ throw new DdlException("Resource(" + resource.getName() + ")
already exist");
+ }
+ logResource = getCreateLogResource(resource);
+ // Queue CREATE before publishing the live object so dependent
operations cannot journal first.
+ logItem = Env.getCurrentEnv().getEditLog()
Review Comment:
[P1] Keep live image membership aligned with submitted journal IDs
CREATE now submits `OP_CREATE_RESOURCE` before `nameToResource.put`, and
DROP submits before `remove`. In direct mode that record is already durable
when `submitEdit()` returns; in batch mode the flusher can also persist it
immediately. If `Env.dumpImage()` selects that max journal ID in the
intervening window, the body omits the CREATE (or still contains the DROP).
Recovery trusts the `image.<max-id>` filename sequence in `loadImage()`—the
older embedded header value is only checksummed—so it starts after that record
and permanently loses or resurrects the resource. Please coordinate the
live-image boundary with these lifecycle map transitions and add direct/batch
recovery tests paused after persistence but before put/remove.
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/ResourceMgr.java:
##########
@@ -108,35 +125,54 @@ public void replayCreateResource(Resource resource) {
public void dropResource(DropResourceCommand dropResourceCommand) throws
DdlException {
String resourceName = dropResourceCommand.getResourceName();
- if (!nameToResource.containsKey(resourceName)) {
+ Resource resource = nameToResource.get(resourceName);
+ if (resource == null) {
if (dropResourceCommand.isIfExists()) {
return;
}
throw new DdlException("Resource(" + resourceName + ") does not
exist");
}
- Resource resource = nameToResource.get(resourceName);
- resource.dropResource();
-
- // Check whether the resource is in use before deleting it, except
spark resource
- StoragePolicy checkedStoragePolicy = StoragePolicy.ofCheck(null);
- checkedStoragePolicy.setStorageResource(resourceName);
- if
(Env.getCurrentEnv().getPolicyMgr().existPolicy(checkedStoragePolicy)) {
- Policy policy =
Env.getCurrentEnv().getPolicyMgr().getPolicy(checkedStoragePolicy);
- LOG.warn("Can not drop resource, since it's used in policy {}",
policy.getPolicyName());
- throw new DdlException("Can not drop resource, since it's used in
policy " + policy.getPolicyName());
+ EditLog.EditLogItem logItem;
+ synchronized (resource.getAlterLock()) {
+ if (nameToResource.get(resourceName) != resource) {
+ if (dropResourceCommand.isIfExists() &&
!nameToResource.containsKey(resourceName)) {
+ return;
+ }
+ throw new DdlException("Resource(" + resourceName + ") changed
concurrently");
+ }
+
+ resource.dropResource();
+
+ // Check whether the resource is in use before deleting it, except
spark resource
+ StoragePolicy checkedStoragePolicy = StoragePolicy.ofCheck(null);
+ checkedStoragePolicy.setStorageResource(resourceName);
+ if
(Env.getCurrentEnv().getPolicyMgr().existPolicy(checkedStoragePolicy)) {
+ Policy policy =
Env.getCurrentEnv().getPolicyMgr().getPolicy(checkedStoragePolicy);
+ LOG.warn("Can not drop resource, since it's used in policy
{}", policy.getPolicyName());
+ throw new DdlException("Can not drop resource, since it's used
in policy " + policy.getPolicyName());
+ }
+
+ logItem = Env.getCurrentEnv().getEditLog().submitEdit(
+ OperationType.OP_DROP_RESOURCE, new
DropResourceOperationLog(resourceName));
+ if (!nameToResource.remove(resourceName, resource)) {
+ LOG.error("Fatal Error: failed to remove resource {} after
submitting its drop log", resourceName);
+ System.exit(-1);
+ throw new IllegalStateException("failed to remove resource " +
resourceName);
+ }
}
- nameToResource.remove(resourceName);
- // log drop
- Env.getCurrentEnv().getEditLog().logDropResource(new
DropResourceOperationLog(resourceName));
+
+ logItem.await();
LOG.info("Drop resource success. Resource resourceName: {}",
resourceName);
}
// Drop resource whether successful or not
public void dropResource(Resource resource) {
String name = resource.getName();
- if (nameToResource.remove(name) == null) {
- LOG.info("resource " + name + " does not exists.");
+ synchronized (resource.getAlterLock()) {
+ if (!nameToResource.remove(name, resource)) {
Review Comment:
[P1] Remove the logical restore-owned resource during replay
This exact-object removal cannot reproduce restore cancellation across
journal records. Replay of the DOWNLOAD `RestoreJob` inserts the Resource
objects nested in that record, but a later COMMIT record is independently
deserialized and becomes the current job without recreating those resources.
When a subsequent CANCEL record replays, `replayCancel()` cleans up through the
COMMIT job, so it passes a logically identical but different Resource instance
here and `remove(name, resource)` silently fails. The CANCEL record has already
cleared `restoredResources` and writes no separate resource DROP, leaving an
orphan resource after recovery even though live cancellation removed it. Please
match the current resource by stable restore ownership/ID under its lifecycle
lock (while still protecting a true same-name replacement), and add a DOWNLOAD
-> COMMIT -> CANCEL recovery test with a restored ODBC resource.
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/ResourceMgr.java:
##########
@@ -145,20 +181,40 @@ public void replayDropResource(DropResourceOperationLog
operationLog) {
}
public void alterResource(String resourceName, Map<String, String>
properties) throws DdlException {
- if (!nameToResource.containsKey(resourceName)) {
+ Resource resource = nameToResource.get(resourceName);
+ if (resource == null) {
throw new DdlException("Resource(" + resourceName + ") dose not
exist.");
}
- Resource resource = nameToResource.get(resourceName);
- resource.modifyProperties(properties);
+ Resource logResource;
+ EditLog.EditLogItem logItem;
+ synchronized (resource.getAlterLock()) {
Review Comment:
[P1] Recheck policy-sensitive fields after remote validation
This `alterLock` is not taken by storage-policy reference acquisition. S3
checks whether a policy reference exists before slow `pingS3()`, so ALTER can
pass that guard and block in remote I/O; CREATE POLICY can then add and journal
its reference; ALTER resumes and commits bucket/root/region/endpoint changes
that S3 explicitly forbids once a policy uses the resource. Both journal orders
replay to the same invalid combination. Please coordinate policy reference
acquisition with the ALTER commit (or recheck atomically after validation)
using a documented lock order that still keeps remote I/O outside the state
lock, and add an S3 policy-versus-blocked-ping test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
