You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@amoro.apache.org by "majin1102 (via GitHub)" <gi...@apache.org> on 2025/03/25 13:59:01 UTC
[PR] [AMORO-3485][WIP] Introduce scheduler module and external resource container [amoro]
majin1102 opened a new pull request, #3486:
URL: https://github.com/apache/amoro/pull/3486
<!--
Thanks for sending a pull request!
Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://amoro.apache.org/how-to-contribute/
2. If the PR is related to an issue in https://github.com/apache/amoro/issues, add '[AMORO-XXXX]' in your PR title, e.g., '[AMORO-XXXX] Your PR title ...'.
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][AMORO-XXXX] Your PR title ...'.
-->
## Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you add a feature, you can talk about its use case.
2. If you fix a bug, you can clarify why it is a bug.
3. Use Fix/Resolve/Close #{ISSUE_NUMBER} to link this PR to its related issue
-->
POC #3485.
## Brief change log
<!--
Clearly describe the changes made in modules, classes, methods, etc.
-->
-
## How was this patch tested?
- [ ] Add some test cases that check the changes thoroughly including negative and positive cases if possible
- [ ] Add screenshots for manual tests if appropriate
- [ ] Run test locally before making a pull request
## Documentation
- Does this pull request introduce a new feature? (yes / no)
- If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-3485] Introduce scheduler module and external resource container [amoro]
Posted by "majin1102 (via GitHub)" <gi...@apache.org>.
majin1102 commented on code in PR #3486:
URL: https://github.com/apache/amoro/pull/3486#discussion_r2084018754
##########
amoro-common/src/main/java/org/apache/amoro/process/TableProcessState.java:
##########
@@ -30,6 +30,7 @@ public class TableProcessState implements ProcessState {
@StateField private volatile long id;
private final Action action;
Review Comment:
Yes
What's the problem?
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-3485] Introduce scheduler module and external resource container [amoro]
Posted by "majin1102 (via GitHub)" <gi...@apache.org>.
majin1102 commented on code in PR #3486:
URL: https://github.com/apache/amoro/pull/3486#discussion_r2084755236
##########
amoro-common/src/main/java/org/apache/amoro/process/ManagedProcess.java:
##########
@@ -0,0 +1,42 @@
+/*
+ * 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.amoro.process;
+
+/**
+ * ManagedProcess is the interface for AMS modules to manage process.
+ *
+ * @param <T>
+ */
+public interface ManagedProcess<T extends ProcessState> extends AmoroProcess<T> {
+
+ /** Submit this process to ResourceContainer. */
+ void submit();
+
+ /** Mark this process as completed, trigger callbacks */
+ void complete();
+
+ /** Mark this process as failed, trigger callbacks */
+ void complete(String failedReason);
Review Comment:
What's the point?
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-3485] Introduce scheduler module and external resource container [amoro]
Posted by "majin1102 (via GitHub)" <gi...@apache.org>.
majin1102 commented on PR #3486:
URL: https://github.com/apache/amoro/pull/3486#issuecomment-2911878207
> I left some comments, PTAL
Any updates?
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-3485][WIP] Introduce scheduler module and external resource container [amoro]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #3486:
URL: https://github.com/apache/amoro/pull/3486#issuecomment-2829097340
This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@amoro.apache.org list. Thank you for your contributions.
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-3485] Introduce scheduler module and external resource container [amoro]
Posted by "XBaith (via GitHub)" <gi...@apache.org>.
XBaith commented on code in PR #3486:
URL: https://github.com/apache/amoro/pull/3486#discussion_r2077092667
##########
amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java:
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.amoro.server.table;
+
+import org.apache.amoro.Action;
+import org.apache.amoro.ManagedTableRuntime;
+import org.apache.amoro.ServerTableIdentifier;
+import org.apache.amoro.config.TableConfiguration;
+import org.apache.amoro.process.AmoroProcess;
+import org.apache.amoro.process.ProcessFactory;
+import org.apache.amoro.process.TableProcessState;
+import org.apache.amoro.server.metrics.MetricRegistry;
+import org.apache.amoro.server.optimizing.OptimizingProcess;
+import org.apache.amoro.server.persistence.StatedPersistentBase;
+import org.apache.amoro.server.persistence.TableRuntimeMeta;
+import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.stream.Collectors;
+
+public class DefaultTableRuntime extends StatedPersistentBase implements ManagedTableRuntime {
+
+ private static final Logger LOG = LoggerFactory.getLogger(DefaultTableRuntime.class);
+
+ private final Lock processLock = new ReentrantLock();
+ private final ServerTableIdentifier tableIdentifier;
+ private final DefaultOptimizingState optimizingState;
+ private final Map<Action, ProcessFactory<? extends TableProcessState>> processFactoryMap =
Review Comment:
Is it thread safe? I believe processFactoryMap is used by multiple threads.
We can remove the reentrant lock and use `ConcurrentHashMap` directly to have better performance.
##########
amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java:
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.amoro.server.table;
+
+import org.apache.amoro.Action;
+import org.apache.amoro.ManagedTableRuntime;
+import org.apache.amoro.ServerTableIdentifier;
+import org.apache.amoro.config.TableConfiguration;
+import org.apache.amoro.process.AmoroProcess;
+import org.apache.amoro.process.ProcessFactory;
+import org.apache.amoro.process.TableProcessState;
+import org.apache.amoro.server.metrics.MetricRegistry;
+import org.apache.amoro.server.optimizing.OptimizingProcess;
+import org.apache.amoro.server.persistence.StatedPersistentBase;
+import org.apache.amoro.server.persistence.TableRuntimeMeta;
+import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.stream.Collectors;
+
+public class DefaultTableRuntime extends StatedPersistentBase implements ManagedTableRuntime {
+
+ private static final Logger LOG = LoggerFactory.getLogger(DefaultTableRuntime.class);
+
+ private final Lock processLock = new ReentrantLock();
+ private final ServerTableIdentifier tableIdentifier;
+ private final DefaultOptimizingState optimizingState;
+ private final Map<Action, ProcessFactory<? extends TableProcessState>> processFactoryMap =
+ new HashMap<>();
+ private final Map<Long, AmoroProcess<? extends TableProcessState>> processMap = new HashMap<>();
+
+ public DefaultTableRuntime(
+ ServerTableIdentifier tableIdentifier,
+ TableRuntimeHandler tableHandler,
+ Map<String, String> properties) {
+ Preconditions.checkNotNull(tableIdentifier, "ServerTableIdentifier must not be null.");
+ Preconditions.checkNotNull(tableHandler, "TableRuntimeHandler must not be null.");
+ this.tableIdentifier = tableIdentifier;
+ this.optimizingState = new DefaultOptimizingState(this, tableHandler, properties);
+ }
+
+ public DefaultTableRuntime(TableRuntimeMeta tableRuntimeMeta, TableRuntimeHandler tableHandler) {
+ Preconditions.checkNotNull(tableRuntimeMeta, "TableRuntimeMeta must not be null.");
+ Preconditions.checkNotNull(tableHandler, "TableRuntimeHandler must not be null.");
+
+ this.tableIdentifier =
+ ServerTableIdentifier.of(
+ tableRuntimeMeta.getTableId(),
+ tableRuntimeMeta.getCatalogName(),
+ tableRuntimeMeta.getDbName(),
+ tableRuntimeMeta.getTableName(),
+ tableRuntimeMeta.getFormat());
+ this.optimizingState = new DefaultOptimizingState(this, tableRuntimeMeta, tableHandler);
+ }
+
+ public DefaultOptimizingState getOptimizingState() {
+ return optimizingState;
+ }
+
+ public void recover(OptimizingProcess optimizingProcess) {
Review Comment:
Expose addiational methods: `recover`, `registerMetric`, `dispose`.
How abuot invoke thoese methods via `getOptimizingState().xxx`?
##########
amoro-common/src/main/java/org/apache/amoro/ManagedTableRuntime.java:
##########
@@ -0,0 +1,50 @@
+/*
+ * 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.amoro;
+
+import org.apache.amoro.process.AmoroProcess;
+import org.apache.amoro.process.ProcessFactory;
+import org.apache.amoro.process.TableProcessState;
+
+public interface ManagedTableRuntime extends TableRuntime {
Review Comment:
The functions in this interface are all about actions and processes.
I think we can split it into an interface called `SupportsAmoroProcess`. So, `DefaultTableRuntime` implements `TableRuntime` and `SupportsAmoroProcess`
##########
amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java:
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.amoro.server.table;
+
+import org.apache.amoro.Action;
+import org.apache.amoro.ManagedTableRuntime;
+import org.apache.amoro.ServerTableIdentifier;
+import org.apache.amoro.config.TableConfiguration;
+import org.apache.amoro.process.AmoroProcess;
+import org.apache.amoro.process.ProcessFactory;
+import org.apache.amoro.process.TableProcessState;
+import org.apache.amoro.server.metrics.MetricRegistry;
+import org.apache.amoro.server.optimizing.OptimizingProcess;
+import org.apache.amoro.server.persistence.StatedPersistentBase;
+import org.apache.amoro.server.persistence.TableRuntimeMeta;
+import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.stream.Collectors;
+
+public class DefaultTableRuntime extends StatedPersistentBase implements ManagedTableRuntime {
+
+ private static final Logger LOG = LoggerFactory.getLogger(DefaultTableRuntime.class);
+
+ private final Lock processLock = new ReentrantLock();
+ private final ServerTableIdentifier tableIdentifier;
+ private final DefaultOptimizingState optimizingState;
+ private final Map<Action, ProcessFactory<? extends TableProcessState>> processFactoryMap =
+ new HashMap<>();
+ private final Map<Long, AmoroProcess<? extends TableProcessState>> processMap = new HashMap<>();
Review Comment:
same as processFactoryMap
##########
amoro-common/src/main/java/org/apache/amoro/resource/ExternalResourceContainer.java:
##########
@@ -0,0 +1,45 @@
+/*
+ * 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.amoro.resource;
+
+import org.apache.amoro.process.AmoroProcess;
+import org.apache.amoro.process.TableProcessState;
+
+/**
+ * ExternalResourceContainer is the key interface for the AMS framework to interact with the
+ * external resource. Typically, it is used to submit and release resource on YARN/K8S.
+ */
+public interface ExternalResourceContainer extends ResourceContainer {
Review Comment:
Do you mean the life cycle of `ExternalResourceContainer` is not managed by AMS?
> Typically, it is used to submit and release resource on YARN/K8S.
I don't quite understand here. If AMS itself runs on K8S, then `InternalResourceContainer` will also run on K8S?What is the difference between `InternalResourceContainer` and `ExternalResourceContainer`?
##########
amoro-common/src/main/java/org/apache/amoro/process/ManagedProcess.java:
##########
@@ -0,0 +1,42 @@
+/*
+ * 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.amoro.process;
+
+/**
+ * ManagedProcess is the interface for AMS modules to manage process.
+ *
+ * @param <T>
+ */
+public interface ManagedProcess<T extends ProcessState> extends AmoroProcess<T> {
+
+ /** Submit this process to ResourceContainer. */
+ void submit();
+
+ /** Mark this process as completed, trigger callbacks */
+ void complete();
+
+ /** Mark this process as failed, trigger callbacks */
+ void complete(String failedReason);
Review Comment:
```suggestion
void failed(String failedReason);
```
##########
amoro-common/src/main/java/org/apache/amoro/process/TableProcessState.java:
##########
@@ -30,6 +30,7 @@ public class TableProcessState implements ProcessState {
@StateField private volatile long id;
private final Action action;
Review Comment:
I see the action field is materialized in the DB?
##########
amoro-common/src/main/java/org/apache/amoro/resource/InternalResourceContainer.java:
##########
@@ -0,0 +1,41 @@
+/*
+ * 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.amoro.resource;
+
+/**
+ * Interface of internal resource container. For InternalResourceContainer, resources are managed by
+ * AMS. Resources are decoupled from processes. For ExternalResourceContainer, resources are managed
+ * outside, and each single resource belonged to a process.
+ */
+public interface InternalResourceContainer extends ResourceContainer {
Review Comment:
Do you mean the life cycle is managed by AMS?
Maybe remove the description about "external resource container"
##########
amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableService.java:
##########
@@ -71,7 +71,7 @@ public class DefaultTableService extends PersistentBase implements TableService
public static final Logger LOG = LoggerFactory.getLogger(DefaultTableService.class);
private final long externalCatalogRefreshingInterval;
- private final Map<Long, TableRuntime> tableRuntimeMap = new ConcurrentHashMap<>();
+ private final Map<Long, DefaultTableRuntime> tableRuntimeMap = new ConcurrentHashMap<>();
Review Comment:
why use `DefaultTableRuntime` instead of `TableRuntime`. I don't see any methods which only belong to `DefaultTableRuntime`
##########
amoro-common/src/main/java/org/apache/amoro/IcebergActions.java:
##########
@@ -0,0 +1,31 @@
+/*
+ * 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.amoro;
+
+public class IcebergActions {
+
+ private static final TableFormat[] DEFAULT_FORMATS =
+ new TableFormat[] {TableFormat.ICEBERG, TableFormat.MIXED_ICEBERG, TableFormat.MIXED_HIVE};
+
+ public static final Action SYSTEM = new Action(DEFAULT_FORMATS, 0, "system");
Review Comment:
What is `SYSTEM` meanning
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-3485] Introduce scheduler module and external resource container [amoro]
Posted by "sqd (via GitHub)" <gi...@apache.org>.
sqd commented on code in PR #3486:
URL: https://github.com/apache/amoro/pull/3486#discussion_r2087333206
##########
amoro-ams/src/main/java/org/apache/amoro/server/scheduler/TimelyExternalScheduler.java:
##########
@@ -0,0 +1,231 @@
+/*
+ * 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.amoro.server.scheduler;
+
+import org.apache.amoro.Action;
+import org.apache.amoro.TableRuntime;
+import org.apache.amoro.process.AmoroProcess;
+import org.apache.amoro.process.ManagedProcess;
+import org.apache.amoro.process.ProcessFactory;
+import org.apache.amoro.process.ProcessStatus;
+import org.apache.amoro.process.SimpleFuture;
+import org.apache.amoro.process.TableProcess;
+import org.apache.amoro.process.TableProcessState;
+import org.apache.amoro.resource.ExternalResourceContainer;
+import org.apache.amoro.resource.Resource;
+import org.apache.amoro.resource.ResourceManager;
+import org.apache.amoro.server.persistence.PersistentBase;
+import org.apache.amoro.server.persistence.mapper.ProcessStateMapper;
+import org.apache.amoro.server.table.DefaultTableRuntime;
+import org.apache.amoro.server.table.TableService;
+
+import java.util.List;
+
+public abstract class TimelyExternalScheduler extends TimelyTableScheduler {
+
+ private final ExternalResourceContainer resourceContainer;
+ private final ResourceManager resourceManager;
+ private final ProcessFactory<? extends TableProcessState> processFactory;
+
+ public TimelyExternalScheduler(
+ ResourceManager resourceManager,
+ ExternalResourceContainer resourceContainer,
+ Action action,
+ TableService tableService,
+ int poolSize) {
+ super(action, tableService, poolSize);
+ this.resourceContainer = resourceContainer;
+ this.resourceManager = resourceManager;
+ this.processFactory = generateProcessFactory();
+ }
+
+ @Override
+ protected void initHandler(List<DefaultTableRuntime> tableRuntimeList) {
+ tableRuntimeList.forEach(tableRuntime -> tableRuntime.install(getAction(), processFactory));
+ super.initHandler(tableRuntimeList);
+ }
+
+ @Override
+ protected long getNextExecutingTime(DefaultTableRuntime tableRuntime) {
+ return 0;
+ }
Review Comment:
This essentially schedules the task in a while(true) loop. Intentional?
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-3485] Introduce scheduler module and external resource container [amoro]
Posted by "xxubai (via GitHub)" <gi...@apache.org>.
xxubai commented on code in PR #3486:
URL: https://github.com/apache/amoro/pull/3486#discussion_r2084810119
##########
amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java:
##########
@@ -94,48 +91,35 @@ public void dispose() {
@Override
public AmoroProcess<? extends TableProcessState> trigger(Action action) {
- processLock.lock();
- try {
- AmoroProcess<? extends TableProcessState> process =
- Optional.ofNullable(processFactoryMap.get(action))
- .map(factory -> factory.create(this, action))
- // Define a related exception
- .orElseThrow(
- () -> new IllegalArgumentException("No ProcessFactory for action " + action));
- processMap.put(process.getId(), process);
- process.getCompleteFuture().whenCompleted(() -> processMap.remove(process.getId()));
- return process;
- } finally {
- processLock.unlock();
- }
+ return Optional.ofNullable(processContainerMap.get(action))
+ .map(container -> container.trigger(action))
+ // Define a related exception
+ .orElseThrow(() -> new IllegalArgumentException("No ProcessFactory for action " + action));
}
@Override
public void install(Action action, ProcessFactory<? extends TableProcessState> processFactory) {
- processLock.lock();
- try {
- if (processFactoryMap.containsKey(action)) {
- throw new IllegalStateException("ProcessFactory for action " + action + " already exists");
- }
- processFactoryMap.put(action, processFactory);
- } finally {
- processLock.unlock();
+ if (processContainerMap.putIfAbsent(action, new TableProcessContainer(processFactory))
+ == null) {
+ throw new IllegalStateException("ProcessFactory for action " + action + " already exists");
}
}
@Override
public boolean enabled(Action action) {
- return processFactoryMap.get(action) != null;
+ return processContainerMap.get(action) != null;
Review Comment:
processContainerMap.containsKey(action)
##########
amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java:
##########
@@ -94,48 +91,35 @@ public void dispose() {
@Override
public AmoroProcess<? extends TableProcessState> trigger(Action action) {
- processLock.lock();
- try {
- AmoroProcess<? extends TableProcessState> process =
- Optional.ofNullable(processFactoryMap.get(action))
- .map(factory -> factory.create(this, action))
- // Define a related exception
- .orElseThrow(
- () -> new IllegalArgumentException("No ProcessFactory for action " + action));
- processMap.put(process.getId(), process);
- process.getCompleteFuture().whenCompleted(() -> processMap.remove(process.getId()));
- return process;
- } finally {
- processLock.unlock();
- }
+ return Optional.ofNullable(processContainerMap.get(action))
+ .map(container -> container.trigger(action))
+ // Define a related exception
+ .orElseThrow(() -> new IllegalArgumentException("No ProcessFactory for action " + action));
}
@Override
public void install(Action action, ProcessFactory<? extends TableProcessState> processFactory) {
- processLock.lock();
- try {
- if (processFactoryMap.containsKey(action)) {
- throw new IllegalStateException("ProcessFactory for action " + action + " already exists");
- }
- processFactoryMap.put(action, processFactory);
- } finally {
- processLock.unlock();
+ if (processContainerMap.putIfAbsent(action, new TableProcessContainer(processFactory))
+ == null) {
+ throw new IllegalStateException("ProcessFactory for action " + action + " already exists");
}
}
@Override
public boolean enabled(Action action) {
- return processFactoryMap.get(action) != null;
+ return processContainerMap.get(action) != null;
Review Comment:
processContainerMap.containsKey(action)
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-3485] Introduce scheduler module and external resource container [amoro]
Posted by "xxubai (via GitHub)" <gi...@apache.org>.
xxubai commented on code in PR #3486:
URL: https://github.com/apache/amoro/pull/3486#discussion_r2084796484
##########
amoro-common/src/main/java/org/apache/amoro/process/TableProcessState.java:
##########
@@ -30,6 +30,7 @@ public class TableProcessState implements ProcessState {
@StateField private volatile long id;
private final Action action;
Review Comment:
I see
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-3485] Introduce scheduler module and external resource container [amoro]
Posted by "majin1102 (via GitHub)" <gi...@apache.org>.
majin1102 commented on code in PR #3486:
URL: https://github.com/apache/amoro/pull/3486#discussion_r2084721416
##########
amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java:
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.amoro.server.table;
+
+import org.apache.amoro.Action;
+import org.apache.amoro.ManagedTableRuntime;
+import org.apache.amoro.ServerTableIdentifier;
+import org.apache.amoro.config.TableConfiguration;
+import org.apache.amoro.process.AmoroProcess;
+import org.apache.amoro.process.ProcessFactory;
+import org.apache.amoro.process.TableProcessState;
+import org.apache.amoro.server.metrics.MetricRegistry;
+import org.apache.amoro.server.optimizing.OptimizingProcess;
+import org.apache.amoro.server.persistence.StatedPersistentBase;
+import org.apache.amoro.server.persistence.TableRuntimeMeta;
+import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.stream.Collectors;
+
+public class DefaultTableRuntime extends StatedPersistentBase implements ManagedTableRuntime {
+
+ private static final Logger LOG = LoggerFactory.getLogger(DefaultTableRuntime.class);
+
+ private final Lock processLock = new ReentrantLock();
+ private final ServerTableIdentifier tableIdentifier;
+ private final DefaultOptimizingState optimizingState;
+ private final Map<Action, ProcessFactory<? extends TableProcessState>> processFactoryMap =
Review Comment:
It is thread safe by processLock.
We could improve this by ConcurrentHashMap, still the processLock is necessary to handle read-and-create atomicity
The code has committed, PTAL
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-3485] Introduce scheduler module and external resource container [amoro]
Posted by "majin1102 (via GitHub)" <gi...@apache.org>.
majin1102 commented on code in PR #3486:
URL: https://github.com/apache/amoro/pull/3486#discussion_r2108733328
##########
amoro-ams/src/main/java/org/apache/amoro/server/persistence/mapper/ProcessStateMapper.java:
##########
@@ -0,0 +1,87 @@
+/*
+ * 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.amoro.server.persistence.mapper;
+
+import org.apache.amoro.process.TableProcessState;
+import org.apache.ibatis.annotations.Insert;
+import org.apache.ibatis.annotations.Options;
+import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.Result;
+import org.apache.ibatis.annotations.ResultMap;
+import org.apache.ibatis.annotations.Results;
+import org.apache.ibatis.annotations.Select;
+import org.apache.ibatis.annotations.Update;
+
+import java.util.Map;
+
+public interface ProcessStateMapper {
+
+ @Insert(
+ "INSERT INTO table_process_state "
+ + "(process_id, action, table_id, retry_num, status, start_time, end_time, fail_reason, summary) "
+ + "VALUES "
+ + "(#{id}, #{action}, #{tableId}, #{retryNumber}, #{status}, #{startTime}, #{endTime}, #{failedReason}, #{summary})")
+ @Options(useGeneratedKeys = true, keyProperty = "id")
+ void createProcessState(TableProcessState state);
+
+ @Update(
+ "UPDATE table_process_state "
+ + "SET status = #{status}, start_time = #{startTime} "
+ + "WHERE process_id = #{id} and retry_num = #{retryNumber}")
+ void updateProcessRunning(TableProcessState state);
Review Comment:
I think update is better, and upateXXX is all around.
Maybe you could raise a PR to meet this globally, Let's focus on Process API itself in this PR
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-3485] Introduce scheduler module and external resource container [amoro]
Posted by "majin1102 (via GitHub)" <gi...@apache.org>.
majin1102 commented on code in PR #3486:
URL: https://github.com/apache/amoro/pull/3486#discussion_r2083987674
##########
amoro-common/src/main/java/org/apache/amoro/IcebergActions.java:
##########
@@ -0,0 +1,31 @@
+/*
+ * 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.amoro;
+
+public class IcebergActions {
+
+ private static final TableFormat[] DEFAULT_FORMATS =
+ new TableFormat[] {TableFormat.ICEBERG, TableFormat.MIXED_ICEBERG, TableFormat.MIXED_HIVE};
+
+ public static final Action SYSTEM = new Action(DEFAULT_FORMATS, 0, "system");
Review Comment:
It's the legacy process type like expiring process records, or just use it to take place all types of actions which don't have an action name
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-3485] Introduce scheduler module and external resource container [amoro]
Posted by "majin1102 (via GitHub)" <gi...@apache.org>.
majin1102 commented on code in PR #3486:
URL: https://github.com/apache/amoro/pull/3486#discussion_r2084751638
##########
amoro-common/src/main/java/org/apache/amoro/process/TableProcessState.java:
##########
@@ -30,6 +30,7 @@ public class TableProcessState implements ProcessState {
@StateField private volatile long id;
private final Action action;
Review Comment:
The field shoud never change. so action and identifer are both unnecessary for @StateField
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-3485] Introduce scheduler module and external resource container [amoro]
Posted by "majin1102 (via GitHub)" <gi...@apache.org>.
majin1102 commented on code in PR #3486:
URL: https://github.com/apache/amoro/pull/3486#discussion_r2084022779
##########
amoro-common/src/main/java/org/apache/amoro/process/ManagedProcess.java:
##########
@@ -0,0 +1,42 @@
+/*
+ * 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.amoro.process;
+
+/**
+ * ManagedProcess is the interface for AMS modules to manage process.
+ *
+ * @param <T>
+ */
+public interface ManagedProcess<T extends ProcessState> extends AmoroProcess<T> {
+
+ /** Submit this process to ResourceContainer. */
+ void submit();
+
+ /** Mark this process as completed, trigger callbacks */
+ void complete();
+
+ /** Mark this process as failed, trigger callbacks */
+ void complete(String failedReason);
Review Comment:
This is quite different.
The complete(String failReason) is called by task completion phase. and failing could happen at any phase(like submitting), so I think complete(String failReason) is more reasonable here which means only completion callback would use this method
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-3485] Introduce scheduler module and external resource container [amoro]
Posted by "majin1102 (via GitHub)" <gi...@apache.org>.
majin1102 commented on code in PR #3486:
URL: https://github.com/apache/amoro/pull/3486#discussion_r2084004522
##########
amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java:
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.amoro.server.table;
+
+import org.apache.amoro.Action;
+import org.apache.amoro.ManagedTableRuntime;
+import org.apache.amoro.ServerTableIdentifier;
+import org.apache.amoro.config.TableConfiguration;
+import org.apache.amoro.process.AmoroProcess;
+import org.apache.amoro.process.ProcessFactory;
+import org.apache.amoro.process.TableProcessState;
+import org.apache.amoro.server.metrics.MetricRegistry;
+import org.apache.amoro.server.optimizing.OptimizingProcess;
+import org.apache.amoro.server.persistence.StatedPersistentBase;
+import org.apache.amoro.server.persistence.TableRuntimeMeta;
+import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.stream.Collectors;
+
+public class DefaultTableRuntime extends StatedPersistentBase implements ManagedTableRuntime {
+
+ private static final Logger LOG = LoggerFactory.getLogger(DefaultTableRuntime.class);
+
+ private final Lock processLock = new ReentrantLock();
+ private final ServerTableIdentifier tableIdentifier;
+ private final DefaultOptimizingState optimizingState;
+ private final Map<Action, ProcessFactory<? extends TableProcessState>> processFactoryMap =
+ new HashMap<>();
+ private final Map<Long, AmoroProcess<? extends TableProcessState>> processMap = new HashMap<>();
+
+ public DefaultTableRuntime(
+ ServerTableIdentifier tableIdentifier,
+ TableRuntimeHandler tableHandler,
+ Map<String, String> properties) {
+ Preconditions.checkNotNull(tableIdentifier, "ServerTableIdentifier must not be null.");
+ Preconditions.checkNotNull(tableHandler, "TableRuntimeHandler must not be null.");
+ this.tableIdentifier = tableIdentifier;
+ this.optimizingState = new DefaultOptimizingState(this, tableHandler, properties);
+ }
+
+ public DefaultTableRuntime(TableRuntimeMeta tableRuntimeMeta, TableRuntimeHandler tableHandler) {
+ Preconditions.checkNotNull(tableRuntimeMeta, "TableRuntimeMeta must not be null.");
+ Preconditions.checkNotNull(tableHandler, "TableRuntimeHandler must not be null.");
+
+ this.tableIdentifier =
+ ServerTableIdentifier.of(
+ tableRuntimeMeta.getTableId(),
+ tableRuntimeMeta.getCatalogName(),
+ tableRuntimeMeta.getDbName(),
+ tableRuntimeMeta.getTableName(),
+ tableRuntimeMeta.getFormat());
+ this.optimizingState = new DefaultOptimizingState(this, tableRuntimeMeta, tableHandler);
+ }
+
+ public DefaultOptimizingState getOptimizingState() {
+ return optimizingState;
+ }
+
+ public void recover(OptimizingProcess optimizingProcess) {
Review Comment:
Agree
This could be introduced by another PR. I did't involve these to keep PR relatively small
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-3485] Introduce scheduler module and external resource container [amoro]
Posted by "majin1102 (via GitHub)" <gi...@apache.org>.
majin1102 commented on code in PR #3486:
URL: https://github.com/apache/amoro/pull/3486#discussion_r2083998401
##########
amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableService.java:
##########
@@ -71,7 +71,7 @@ public class DefaultTableService extends PersistentBase implements TableService
public static final Logger LOG = LoggerFactory.getLogger(DefaultTableService.class);
private final long externalCatalogRefreshingInterval;
- private final Map<Long, TableRuntime> tableRuntimeMap = new ConcurrentHashMap<>();
+ private final Map<Long, DefaultTableRuntime> tableRuntimeMap = new ConcurrentHashMap<>();
Review Comment:
like getOptimizingState()?
May be we can use ManagedTableRuntime to take place of DefaultTableRuntime, which would cause quite a lot of work. The previos POJO here is old TableRuntime, which is replaced by DefaultTableRuntime.getOptimizingState(), which has not been asbtracted to TableRuntime
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-3485] Introduce scheduler module and external resource container [amoro]
Posted by "majin1102 (via GitHub)" <gi...@apache.org>.
majin1102 commented on code in PR #3486:
URL: https://github.com/apache/amoro/pull/3486#discussion_r2084721416
##########
amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java:
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.amoro.server.table;
+
+import org.apache.amoro.Action;
+import org.apache.amoro.ManagedTableRuntime;
+import org.apache.amoro.ServerTableIdentifier;
+import org.apache.amoro.config.TableConfiguration;
+import org.apache.amoro.process.AmoroProcess;
+import org.apache.amoro.process.ProcessFactory;
+import org.apache.amoro.process.TableProcessState;
+import org.apache.amoro.server.metrics.MetricRegistry;
+import org.apache.amoro.server.optimizing.OptimizingProcess;
+import org.apache.amoro.server.persistence.StatedPersistentBase;
+import org.apache.amoro.server.persistence.TableRuntimeMeta;
+import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.stream.Collectors;
+
+public class DefaultTableRuntime extends StatedPersistentBase implements ManagedTableRuntime {
+
+ private static final Logger LOG = LoggerFactory.getLogger(DefaultTableRuntime.class);
+
+ private final Lock processLock = new ReentrantLock();
+ private final ServerTableIdentifier tableIdentifier;
+ private final DefaultOptimizingState optimizingState;
+ private final Map<Action, ProcessFactory<? extends TableProcessState>> processFactoryMap =
Review Comment:
It is thread safe by processLock.
We could improve this by ConcurrentHashMap, still the processLock is necessary to handle read-and-create process atomicity
The code has committed, PTAL
##########
amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java:
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.amoro.server.table;
+
+import org.apache.amoro.Action;
+import org.apache.amoro.ManagedTableRuntime;
+import org.apache.amoro.ServerTableIdentifier;
+import org.apache.amoro.config.TableConfiguration;
+import org.apache.amoro.process.AmoroProcess;
+import org.apache.amoro.process.ProcessFactory;
+import org.apache.amoro.process.TableProcessState;
+import org.apache.amoro.server.metrics.MetricRegistry;
+import org.apache.amoro.server.optimizing.OptimizingProcess;
+import org.apache.amoro.server.persistence.StatedPersistentBase;
+import org.apache.amoro.server.persistence.TableRuntimeMeta;
+import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.stream.Collectors;
+
+public class DefaultTableRuntime extends StatedPersistentBase implements ManagedTableRuntime {
+
+ private static final Logger LOG = LoggerFactory.getLogger(DefaultTableRuntime.class);
+
+ private final Lock processLock = new ReentrantLock();
+ private final ServerTableIdentifier tableIdentifier;
+ private final DefaultOptimizingState optimizingState;
+ private final Map<Action, ProcessFactory<? extends TableProcessState>> processFactoryMap =
+ new HashMap<>();
+ private final Map<Long, AmoroProcess<? extends TableProcessState>> processMap = new HashMap<>();
Review Comment:
as above
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-3485] Introduce scheduler module and external resource container [amoro]
Posted by "xxubai (via GitHub)" <gi...@apache.org>.
xxubai commented on code in PR #3486:
URL: https://github.com/apache/amoro/pull/3486#discussion_r2084731249
##########
amoro-common/src/main/java/org/apache/amoro/process/TableProcessState.java:
##########
@@ -30,6 +30,7 @@ public class TableProcessState implements ProcessState {
@StateField private volatile long id;
private final Action action;
Review Comment:
Need to add `@StateField` for action field?
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-3485] Introduce scheduler module and external resource container [amoro]
Posted by "xxubai (via GitHub)" <gi...@apache.org>.
xxubai commented on code in PR #3486:
URL: https://github.com/apache/amoro/pull/3486#discussion_r2084740696
##########
amoro-common/src/main/java/org/apache/amoro/resource/ExternalResourceContainer.java:
##########
@@ -0,0 +1,45 @@
+/*
+ * 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.amoro.resource;
+
+import org.apache.amoro.process.AmoroProcess;
+import org.apache.amoro.process.TableProcessState;
+
+/**
+ * ExternalResourceContainer is the key interface for the AMS framework to interact with the
+ * external resource. Typically, it is used to submit and release resource on YARN/K8S.
+ */
+public interface ExternalResourceContainer extends ResourceContainer {
Review Comment:
I think we can make that clear in the comments.
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-3485] Introduce scheduler module and external resource container [amoro]
Posted by "majin1102 (via GitHub)" <gi...@apache.org>.
majin1102 commented on code in PR #3486:
URL: https://github.com/apache/amoro/pull/3486#discussion_r2084740735
##########
amoro-common/src/main/java/org/apache/amoro/resource/InternalResourceContainer.java:
##########
@@ -0,0 +1,41 @@
+/*
+ * 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.amoro.resource;
+
+/**
+ * Interface of internal resource container. For InternalResourceContainer, resources are managed by
+ * AMS. Resources are decoupled from processes. For ExternalResourceContainer, resources are managed
+ * outside, and each single resource belonged to a process.
+ */
+public interface InternalResourceContainer extends ResourceContainer {
Review Comment:
I've improved the description. PTAL
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-3485] Introduce scheduler module and external resource container [amoro]
Posted by "majin1102 (via GitHub)" <gi...@apache.org>.
majin1102 commented on code in PR #3486:
URL: https://github.com/apache/amoro/pull/3486#discussion_r2084751638
##########
amoro-common/src/main/java/org/apache/amoro/process/TableProcessState.java:
##########
@@ -30,6 +30,7 @@ public class TableProcessState implements ProcessState {
@StateField private volatile long id;
private final Action action;
Review Comment:
The field shoud never change. so action and identifer is unnecessary for @StateField
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-3485] Introduce scheduler module and external resource container [amoro]
Posted by "majin1102 (via GitHub)" <gi...@apache.org>.
majin1102 commented on code in PR #3486:
URL: https://github.com/apache/amoro/pull/3486#discussion_r2084755236
##########
amoro-common/src/main/java/org/apache/amoro/process/ManagedProcess.java:
##########
@@ -0,0 +1,42 @@
+/*
+ * 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.amoro.process;
+
+/**
+ * ManagedProcess is the interface for AMS modules to manage process.
+ *
+ * @param <T>
+ */
+public interface ManagedProcess<T extends ProcessState> extends AmoroProcess<T> {
+
+ /** Submit this process to ResourceContainer. */
+ void submit();
+
+ /** Mark this process as completed, trigger callbacks */
+ void complete();
+
+ /** Mark this process as failed, trigger callbacks */
+ void complete(String failedReason);
Review Comment:
What's the point or difference?
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-3485] Introduce scheduler module and external resource container [amoro]
Posted by "majin1102 (via GitHub)" <gi...@apache.org>.
majin1102 commented on code in PR #3486:
URL: https://github.com/apache/amoro/pull/3486#discussion_r2084030393
##########
amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java:
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.amoro.server.table;
+
+import org.apache.amoro.Action;
+import org.apache.amoro.ManagedTableRuntime;
+import org.apache.amoro.ServerTableIdentifier;
+import org.apache.amoro.config.TableConfiguration;
+import org.apache.amoro.process.AmoroProcess;
+import org.apache.amoro.process.ProcessFactory;
+import org.apache.amoro.process.TableProcessState;
+import org.apache.amoro.server.metrics.MetricRegistry;
+import org.apache.amoro.server.optimizing.OptimizingProcess;
+import org.apache.amoro.server.persistence.StatedPersistentBase;
+import org.apache.amoro.server.persistence.TableRuntimeMeta;
+import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.stream.Collectors;
+
+public class DefaultTableRuntime extends StatedPersistentBase implements ManagedTableRuntime {
+
+ private static final Logger LOG = LoggerFactory.getLogger(DefaultTableRuntime.class);
+
+ private final Lock processLock = new ReentrantLock();
+ private final ServerTableIdentifier tableIdentifier;
+ private final DefaultOptimizingState optimizingState;
+ private final Map<Action, ProcessFactory<? extends TableProcessState>> processFactoryMap =
+ new HashMap<>();
+ private final Map<Long, AmoroProcess<? extends TableProcessState>> processMap = new HashMap<>();
Review Comment:
as above
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-3485] Introduce scheduler module and external resource container [amoro]
Posted by "sqd (via GitHub)" <gi...@apache.org>.
sqd commented on code in PR #3486:
URL: https://github.com/apache/amoro/pull/3486#discussion_r2087320641
##########
amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java:
##########
@@ -0,0 +1,162 @@
+/*
+ * 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.amoro.server.table;
+
+import org.apache.amoro.Action;
+import org.apache.amoro.ServerTableIdentifier;
+import org.apache.amoro.SupportsProcessPlugins;
+import org.apache.amoro.config.TableConfiguration;
+import org.apache.amoro.process.AmoroProcess;
+import org.apache.amoro.process.ProcessFactory;
+import org.apache.amoro.process.TableProcessState;
+import org.apache.amoro.server.metrics.MetricRegistry;
+import org.apache.amoro.server.optimizing.OptimizingProcess;
+import org.apache.amoro.server.persistence.StatedPersistentBase;
+import org.apache.amoro.server.persistence.TableRuntimeMeta;
+import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
+import org.apache.amoro.shade.zookeeper3.org.apache.curator.shaded.com.google.common.collect.Maps;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.stream.Collectors;
+
+public class DefaultTableRuntime extends StatedPersistentBase implements SupportsProcessPlugins {
+
+ private static final Logger LOG = LoggerFactory.getLogger(DefaultTableRuntime.class);
+
+ private final ServerTableIdentifier tableIdentifier;
+ private final DefaultOptimizingState optimizingState;
+ private final Map<Action, TableProcessContainer> processContainerMap = Maps.newConcurrentMap();
+
+ public DefaultTableRuntime(
+ ServerTableIdentifier tableIdentifier,
+ TableRuntimeHandler tableHandler,
+ Map<String, String> properties) {
+ Preconditions.checkNotNull(tableIdentifier, "ServerTableIdentifier must not be null.");
+ Preconditions.checkNotNull(tableHandler, "TableRuntimeHandler must not be null.");
+ this.tableIdentifier = tableIdentifier;
+ this.optimizingState = new DefaultOptimizingState(this, tableHandler, properties);
+ }
+
+ public DefaultTableRuntime(TableRuntimeMeta tableRuntimeMeta, TableRuntimeHandler tableHandler) {
+ Preconditions.checkNotNull(tableRuntimeMeta, "TableRuntimeMeta must not be null.");
+ Preconditions.checkNotNull(tableHandler, "TableRuntimeHandler must not be null.");
+
+ this.tableIdentifier =
+ ServerTableIdentifier.of(
+ tableRuntimeMeta.getTableId(),
+ tableRuntimeMeta.getCatalogName(),
+ tableRuntimeMeta.getDbName(),
+ tableRuntimeMeta.getTableName(),
+ tableRuntimeMeta.getFormat());
+ this.optimizingState = new DefaultOptimizingState(this, tableRuntimeMeta, tableHandler);
+ }
+
+ public DefaultOptimizingState getOptimizingState() {
+ return optimizingState;
+ }
+
+ public void recover(OptimizingProcess optimizingProcess) {
+ optimizingState.recover(optimizingProcess);
+ }
+
+ public void registerMetric(MetricRegistry metricRegistry) {
+ optimizingState.registerMetric(metricRegistry);
+ }
+
+ public void dispose() {
+ optimizingState.dispose();
+ }
+
+ @Override
+ public AmoroProcess<? extends TableProcessState> trigger(Action action) {
+ return Optional.ofNullable(processContainerMap.get(action))
+ .map(container -> container.trigger(action))
+ // Define a related exception
+ .orElseThrow(() -> new IllegalArgumentException("No ProcessFactory for action " + action));
+ }
+
+ @Override
+ public void install(Action action, ProcessFactory<? extends TableProcessState> processFactory) {
+ if (processContainerMap.putIfAbsent(action, new TableProcessContainer(processFactory))
+ == null) {
+ throw new IllegalStateException("ProcessFactory for action " + action + " already exists");
Review Comment:
I think this is inverted. Should be `!= null` instead.
```suggestion
if (processContainerMap.putIfAbsent(action, new TableProcessContainer(processFactory))
!= null) {
throw new IllegalStateException("ProcessFactory for action " + action + " already exists");
```
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-3485] Introduce scheduler module and external resource container [amoro]
Posted by "majin1102 (via GitHub)" <gi...@apache.org>.
majin1102 commented on code in PR #3486:
URL: https://github.com/apache/amoro/pull/3486#discussion_r2084013980
##########
amoro-common/src/main/java/org/apache/amoro/resource/ExternalResourceContainer.java:
##########
@@ -0,0 +1,45 @@
+/*
+ * 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.amoro.resource;
+
+import org.apache.amoro.process.AmoroProcess;
+import org.apache.amoro.process.TableProcessState;
+
+/**
+ * ExternalResourceContainer is the key interface for the AMS framework to interact with the
+ * external resource. Typically, it is used to submit and release resource on YARN/K8S.
+ */
+public interface ExternalResourceContainer extends ResourceContainer {
Review Comment:
For InternalContainer executors，they recieve tasks scheduled by AMS, and released by AMS when necessary. For ExternalContainer executors, they recieved tasks by some JobManager or application master. The true diff is that ResourceManager(response for task scheduling and managing lifetime of executors) is inside AMS or outside
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-3485] Introduce scheduler module and external resource container [amoro]
Posted by "majin1102 (via GitHub)" <gi...@apache.org>.
majin1102 commented on code in PR #3486:
URL: https://github.com/apache/amoro/pull/3486#discussion_r2084030080
##########
amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java:
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.amoro.server.table;
+
+import org.apache.amoro.Action;
+import org.apache.amoro.ManagedTableRuntime;
+import org.apache.amoro.ServerTableIdentifier;
+import org.apache.amoro.config.TableConfiguration;
+import org.apache.amoro.process.AmoroProcess;
+import org.apache.amoro.process.ProcessFactory;
+import org.apache.amoro.process.TableProcessState;
+import org.apache.amoro.server.metrics.MetricRegistry;
+import org.apache.amoro.server.optimizing.OptimizingProcess;
+import org.apache.amoro.server.persistence.StatedPersistentBase;
+import org.apache.amoro.server.persistence.TableRuntimeMeta;
+import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.stream.Collectors;
+
+public class DefaultTableRuntime extends StatedPersistentBase implements ManagedTableRuntime {
+
+ private static final Logger LOG = LoggerFactory.getLogger(DefaultTableRuntime.class);
+
+ private final Lock processLock = new ReentrantLock();
+ private final ServerTableIdentifier tableIdentifier;
+ private final DefaultOptimizingState optimizingState;
+ private final Map<Action, ProcessFactory<? extends TableProcessState>> processFactoryMap =
Review Comment:
We could use ConcurrentMap, but processLock is still inevitable.
Use a ConcurrentMap under a lock seems a little dummy to me.
We could use different locks for each actions. I prefer raising another PR for this improvement
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-3485] Introduce scheduler module and external resource container [amoro]
Posted by "majin1102 (via GitHub)" <gi...@apache.org>.
majin1102 commented on code in PR #3486:
URL: https://github.com/apache/amoro/pull/3486#discussion_r2084030080
##########
amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java:
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.amoro.server.table;
+
+import org.apache.amoro.Action;
+import org.apache.amoro.ManagedTableRuntime;
+import org.apache.amoro.ServerTableIdentifier;
+import org.apache.amoro.config.TableConfiguration;
+import org.apache.amoro.process.AmoroProcess;
+import org.apache.amoro.process.ProcessFactory;
+import org.apache.amoro.process.TableProcessState;
+import org.apache.amoro.server.metrics.MetricRegistry;
+import org.apache.amoro.server.optimizing.OptimizingProcess;
+import org.apache.amoro.server.persistence.StatedPersistentBase;
+import org.apache.amoro.server.persistence.TableRuntimeMeta;
+import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.stream.Collectors;
+
+public class DefaultTableRuntime extends StatedPersistentBase implements ManagedTableRuntime {
+
+ private static final Logger LOG = LoggerFactory.getLogger(DefaultTableRuntime.class);
+
+ private final Lock processLock = new ReentrantLock();
+ private final ServerTableIdentifier tableIdentifier;
+ private final DefaultOptimizingState optimizingState;
+ private final Map<Action, ProcessFactory<? extends TableProcessState>> processFactoryMap =
Review Comment:
We could use ConcurrentMap, but processLock is still inevitable.
Use a ConcurrentMap under a lock seems a little dummy to me.
We could use different locks for each actions. I prefer raising another PR for this improvement
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-3485] Introduce scheduler module and external resource container [amoro]
Posted by "majin1102 (via GitHub)" <gi...@apache.org>.
majin1102 commented on code in PR #3486:
URL: https://github.com/apache/amoro/pull/3486#discussion_r2084030393
##########
amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java:
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.amoro.server.table;
+
+import org.apache.amoro.Action;
+import org.apache.amoro.ManagedTableRuntime;
+import org.apache.amoro.ServerTableIdentifier;
+import org.apache.amoro.config.TableConfiguration;
+import org.apache.amoro.process.AmoroProcess;
+import org.apache.amoro.process.ProcessFactory;
+import org.apache.amoro.process.TableProcessState;
+import org.apache.amoro.server.metrics.MetricRegistry;
+import org.apache.amoro.server.optimizing.OptimizingProcess;
+import org.apache.amoro.server.persistence.StatedPersistentBase;
+import org.apache.amoro.server.persistence.TableRuntimeMeta;
+import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.stream.Collectors;
+
+public class DefaultTableRuntime extends StatedPersistentBase implements ManagedTableRuntime {
+
+ private static final Logger LOG = LoggerFactory.getLogger(DefaultTableRuntime.class);
+
+ private final Lock processLock = new ReentrantLock();
+ private final ServerTableIdentifier tableIdentifier;
+ private final DefaultOptimizingState optimizingState;
+ private final Map<Action, ProcessFactory<? extends TableProcessState>> processFactoryMap =
+ new HashMap<>();
+ private final Map<Long, AmoroProcess<? extends TableProcessState>> processMap = new HashMap<>();
Review Comment:
as above
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-3485] Introduce scheduler module and external resource container [amoro]
Posted by "xxubai (via GitHub)" <gi...@apache.org>.
xxubai merged PR #3486:
URL: https://github.com/apache/amoro/pull/3486
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-3485] Introduce scheduler module and external resource container [amoro]
Posted by "xxubai (via GitHub)" <gi...@apache.org>.
xxubai commented on code in PR #3486:
URL: https://github.com/apache/amoro/pull/3486#discussion_r2084737155
##########
amoro-common/src/main/java/org/apache/amoro/process/ManagedProcess.java:
##########
@@ -0,0 +1,42 @@
+/*
+ * 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.amoro.process;
+
+/**
+ * ManagedProcess is the interface for AMS modules to manage process.
+ *
+ * @param <T>
+ */
+public interface ManagedProcess<T extends ProcessState> extends AmoroProcess<T> {
+
+ /** Submit this process to ResourceContainer. */
+ void submit();
+
+ /** Mark this process as completed, trigger callbacks */
+ void complete();
+
+ /** Mark this process as failed, trigger callbacks */
+ void complete(String failedReason);
Review Comment:
how about define one method with a flag indicate the process is success or not
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-3485] Introduce scheduler module and external resource container [amoro]
Posted by "majin1102 (via GitHub)" <gi...@apache.org>.
majin1102 commented on PR #3486:
URL: https://github.com/apache/amoro/pull/3486#issuecomment-2871367742
@xxubai check it out
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-3485] Introduce scheduler module and external resource container [amoro]
Posted by "majin1102 (via GitHub)" <gi...@apache.org>.
majin1102 commented on code in PR #3486:
URL: https://github.com/apache/amoro/pull/3486#discussion_r2084744218
##########
amoro-common/src/main/java/org/apache/amoro/resource/ExternalResourceContainer.java:
##########
@@ -0,0 +1,45 @@
+/*
+ * 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.amoro.resource;
+
+import org.apache.amoro.process.AmoroProcess;
+import org.apache.amoro.process.TableProcessState;
+
+/**
+ * ExternalResourceContainer is the key interface for the AMS framework to interact with the
+ * external resource. Typically, it is used to submit and release resource on YARN/K8S.
+ */
+public interface ExternalResourceContainer extends ResourceContainer {
Review Comment:
PTAL
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-3485] Introduce scheduler module and external resource container [amoro]
Posted by "xxubai (via GitHub)" <gi...@apache.org>.
xxubai commented on code in PR #3486:
URL: https://github.com/apache/amoro/pull/3486#discussion_r2084777155
##########
amoro-ams/src/main/java/org/apache/amoro/server/persistence/mapper/ProcessStateMapper.java:
##########
@@ -0,0 +1,87 @@
+/*
+ * 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.amoro.server.persistence.mapper;
+
+import org.apache.amoro.process.TableProcessState;
+import org.apache.ibatis.annotations.Insert;
+import org.apache.ibatis.annotations.Options;
+import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.Result;
+import org.apache.ibatis.annotations.ResultMap;
+import org.apache.ibatis.annotations.Results;
+import org.apache.ibatis.annotations.Select;
+import org.apache.ibatis.annotations.Update;
+
+import java.util.Map;
+
+public interface ProcessStateMapper {
+
+ @Insert(
+ "INSERT INTO table_process_state "
+ + "(process_id, action, table_id, retry_num, status, start_time, end_time, fail_reason, summary) "
+ + "VALUES "
+ + "(#{id}, #{action}, #{tableId}, #{retryNumber}, #{status}, #{startTime}, #{endTime}, #{failedReason}, #{summary})")
+ @Options(useGeneratedKeys = true, keyProperty = "id")
+ void createProcessState(TableProcessState state);
+
+ @Update(
+ "UPDATE table_process_state "
+ + "SET status = #{status}, start_time = #{startTime} "
+ + "WHERE process_id = #{id} and retry_num = #{retryNumber}")
+ void updateProcessRunning(TableProcessState state);
+
+ @Update(
+ "UPDATE table_process_state "
+ + "SET status = #{status}, end_time = #{endTime} "
+ + "WHERE process_id = #{id} and retry_num = #{retryNumber}")
+ void updateProcessCompleted(TableProcessState state);
+
+ @Update(
+ "UPDATE table_process_state "
+ + "SET status = #{status}, end_time = #{endTime}, fail_reason = #{failedReason} "
+ + "WHERE process_id = #{id} and retry_num = #{retryNumber}")
+ void updateProcessFailed(TableProcessState state);
+
+ @Select(
+ "SELECT process_id, action, table_id, retry_num, status, start_time, end_time, fail_reason, summary "
+ + "FROM table_process_state "
+ + "WHERE process_id = #{processId}")
+ @Results(
+ id = "TableProcessStateResultMap",
+ value = {
+ @Result(property = "id", column = "process_id"),
+ @Result(property = "action", column = "action"),
+ @Result(property = "tableId", column = "table_id"),
+ @Result(property = "retryNumber", column = "retry_num"),
+ @Result(property = "status", column = "status"),
+ @Result(property = "startTime", column = "start_time"),
+ @Result(property = "endTime", column = "end_time"),
+ @Result(property = "failedReason", column = "fail_reason"),
+ @Result(property = "summary", column = "summary", javaType = Map.class)
+ })
+ TableProcessState getProcessStateById(@Param("processId") long processId);
+
+ /** 根据 table_id 查询 TableProcessState */
Review Comment:
```suggestion
/** Get TableProcessState by table id*/
```
##########
amoro-common/src/main/java/org/apache/amoro/SupportsProcessPlugins.java:
##########
@@ -0,0 +1,50 @@
+/*
+ * 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.amoro;
+
+import org.apache.amoro.process.AmoroProcess;
+import org.apache.amoro.process.ProcessFactory;
+import org.apache.amoro.process.TableProcessState;
+
+public interface SupportsProcessPlugins extends TableRuntime {
Review Comment:
I think the interface can be separate, rather than inheriting from TableRuntime
##########
amoro-common/src/main/java/org/apache/amoro/process/ManagedProcess.java:
##########
@@ -0,0 +1,42 @@
+/*
+ * 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.amoro.process;
+
+/**
+ * ManagedProcess is the interface for AMS modules to manage process.
+ *
+ * @param <T>
+ */
+public interface ManagedProcess<T extends ProcessState> extends AmoroProcess<T> {
+
+ /** Submit this process to ResourceContainer. */
+ void submit();
+
+ /** Mark this process as completed, trigger callbacks */
+ void complete();
+
+ /** Mark this process as failed, trigger callbacks */
+ void complete(String failedReason);
Review Comment:
Because these two methods seem to perform different execution logic, but the two method names are the same, so I recommend differentiating the names
##########
amoro-ams/src/main/java/org/apache/amoro/server/scheduler/TimelyTableScheduler.java:
##########
@@ -39,18 +41,32 @@
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
-public abstract class BaseTableExecutor extends RuntimeHandlerChain {
+public abstract class TimelyTableScheduler extends RuntimeHandlerChain {
Review Comment:
```suggestion
public abstract class PeriodicTaskScheduler RuntimeHandlerChain {
```
The tasks execute with a fixed period, maybe `PeriodicTaskScheduler`, `FixedTableScheduler` is more fittable?
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-3485] Introduce scheduler module and external resource container [amoro]
Posted by "Aireed (via GitHub)" <gi...@apache.org>.
Aireed commented on code in PR #3486:
URL: https://github.com/apache/amoro/pull/3486#discussion_r2108997944
##########
amoro-common/src/main/java/org/apache/amoro/process/ProcessStatus.java:
##########
@@ -26,6 +26,7 @@ public enum ProcessStatus {
RUNNING,
SUCCESS,
CLOSED,
Review Comment:
> Check OptimizingQueue.java
i still can't found CLOSED using in OptimizingQueue.java. Please share me the code
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-3485] Introduce scheduler module and external resource container [amoro]
Posted by "majin1102 (via GitHub)" <gi...@apache.org>.
majin1102 commented on code in PR #3486:
URL: https://github.com/apache/amoro/pull/3486#discussion_r2084090480
##########
amoro-common/src/main/java/org/apache/amoro/ManagedTableRuntime.java:
##########
@@ -0,0 +1,50 @@
+/*
+ * 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.amoro;
+
+import org.apache.amoro.process.AmoroProcess;
+import org.apache.amoro.process.ProcessFactory;
+import org.apache.amoro.process.TableProcessState;
+
+public interface ManagedTableRuntime extends TableRuntime {
Review Comment:
Yep,
I think SupportsProcessPlugins better, means support pluggable factories
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org