-
Notifications
You must be signed in to change notification settings - Fork 1.9k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
refactor abstract AbstractJobEnvironment for JobImmutableInformationE…
…nv and JobExecutionEnvironment
- Loading branch information
Showing
3 changed files
with
102 additions
and
171 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
92 changes: 92 additions & 0 deletions
92
...ngine-core/src/main/java/org/apache/seatunnel/engine/core/job/AbstractJobEnvironment.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,92 @@ | ||
| package org.apache.seatunnel.engine.core.job; | ||
|
|
||
| import org.apache.seatunnel.api.env.EnvCommonOptions; | ||
| import org.apache.seatunnel.common.config.Common; | ||
| import org.apache.seatunnel.common.utils.FileUtils; | ||
| import org.apache.seatunnel.engine.common.config.JobConfig; | ||
| import org.apache.seatunnel.engine.common.exception.SeaTunnelEngineException; | ||
| import org.apache.seatunnel.engine.common.utils.IdGenerator; | ||
| import org.apache.seatunnel.engine.core.dag.actions.Action; | ||
| import org.apache.seatunnel.engine.core.dag.logical.LogicalDag; | ||
| import org.apache.seatunnel.engine.core.dag.logical.LogicalDagGenerator; | ||
| import org.apache.seatunnel.engine.core.parse.MultipleTableJobConfigParser; | ||
|
|
||
| import org.apache.commons.lang3.tuple.ImmutablePair; | ||
|
|
||
| import com.hazelcast.logging.ILogger; | ||
| import com.hazelcast.logging.Logger; | ||
|
|
||
| import java.io.IOException; | ||
| import java.net.MalformedURLException; | ||
| import java.net.URL; | ||
| import java.nio.file.Files; | ||
| import java.nio.file.Path; | ||
| import java.util.stream.Collectors; | ||
|
|
||
| public abstract class AbstractJobEnvironment { | ||
| protected static ILogger LOGGER = null; | ||
|
|
||
| protected final boolean isStartWithSavePoint; | ||
|
|
||
| protected final List<Action> actions = new ArrayList<>(); | ||
| protected final Set<URL> jarUrls = new HashSet<>(); | ||
|
|
||
| protected final JobConfig jobConfig; | ||
|
|
||
| protected final IdGenerator idGenerator; | ||
|
|
||
| protected final List<URL> commonPluginJars = new ArrayList<>(); | ||
|
|
||
| public AbstractJobEnvironment(JobConfig jobConfig, boolean isStartWithSavePoint) { | ||
| LOGGER = Logger.getLogger(getClass().getName()); | ||
| this.jobConfig = jobConfig; | ||
| this.isStartWithSavePoint = isStartWithSavePoint; | ||
| this.idGenerator = new IdGenerator(); | ||
| this.commonPluginJars.addAll(searchPluginJars()); | ||
| this.commonPluginJars.addAll( | ||
| new ArrayList<>( | ||
| Common.getThirdPartyJars( | ||
| jobConfig | ||
| .getEnvOptions() | ||
| .getOrDefault(EnvCommonOptions.JARS.key(), "") | ||
| .toString()) | ||
| .stream() | ||
| .map(Path::toUri) | ||
| .map( | ||
| uri -> { | ||
| try { | ||
| return uri.toURL(); | ||
| } catch (MalformedURLException e) { | ||
| throw new SeaTunnelEngineException( | ||
| "the uri of jar illegal:" + uri, e); | ||
| } | ||
| }) | ||
| .collect(Collectors.toList()))); | ||
| LOGGER.info("add common jar in plugins :" + commonPluginJars); | ||
| } | ||
|
|
||
| protected Set<URL> searchPluginJars() { | ||
| try { | ||
| if (Files.exists(Common.pluginRootDir())) { | ||
| return new HashSet<>(FileUtils.searchJarFiles(Common.pluginRootDir())); | ||
| } | ||
| } catch (IOException | SeaTunnelEngineException e) { | ||
| LOGGER.warning( | ||
| String.format("Can't search plugin jars in %s.", Common.pluginRootDir()), e); | ||
| } | ||
| return Collections.emptySet(); | ||
| } | ||
|
|
||
| protected abstract MultipleTableJobConfigParser getJobConfigParser(); | ||
|
|
||
| protected LogicalDagGenerator getLogicalDagGenerator() { | ||
| return new LogicalDagGenerator(actions, jobConfig, idGenerator); | ||
| } | ||
|
|
||
| protected LogicalDag getLogicalDag() { | ||
| ImmutablePair<List<Action>, Set<URL>> immutablePair = getJobConfigParser().parse(); | ||
| actions.addAll(immutablePair.getLeft()); | ||
| jarUrls.addAll(immutablePair.getRight()); | ||
| return getLogicalDagGenerator().generate(); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters