lianghuan-xatu commented on code in PR #5542:
URL: https://github.com/apache/seatunnel/pull/5542#discussion_r1366470369
##########
seatunnel-engine/seatunnel-engine-client/src/main/java/org/apache/seatunnel/engine/client/job/JobExecutionEnvironment.java:
##########
@@ -64,6 +83,92 @@ protected MultipleTableJobConfigParser getJobConfigParser() {
jobFilePath, idGenerator, jobConfig, commonPluginJars,
isStartWithSavePoint);
}
+ @Override
+ protected LogicalDag getLogicalDag() {
+ ImmutablePair<List<Action>, Set<URL>> immutablePair =
getJobConfigParser().parse();
+ actions.addAll(immutablePair.getLeft());
+ // Enable upload connector jar package to engine server, automatically
upload connector Jar
+ // packages
+ // and dependent third-party Jar packages to the server before job
execution.
+ // Enabling this configuration does not require the server to hold all
connector Jar
+ // packages.
+ boolean enableUploadConnectorJarPackage =
+
seaTunnelConfig.getEngineConfig().getConnectorJarStorageConfig().getEnable();
+ if (enableUploadConnectorJarPackage == true) {
+ /**
+ * TODO: Before uploading the Jar package file the server, first
determine whether the
+ * server holds the current Jar. If the server holds the same Jar
package file, there is
+ * no need for additional uploads.
+ */
+ Set<ConnectorJarIdentifier> commonJarIdentifiers =
+ connectorPackageClient.uploadCommonPluginJars(
+
Long.parseLong(jobConfig.getJobContext().getJobId()), commonPluginJars);
+ Set<URL> commonPluginJarUrls =
getJarUrlsFromIdentifiers(commonJarIdentifiers);
+ Set<ConnectorJarIdentifier> pluginJarIdentifiers = new HashSet<>();
+ transformActionPluginJarUrls(actions, pluginJarIdentifiers);
+ Set<URL> connectorPluginJarUrls =
getJarUrlsFromIdentifiers(pluginJarIdentifiers);
+ connectorJarIdentifiers.addAll(commonJarIdentifiers);
+ connectorJarIdentifiers.addAll(pluginJarIdentifiers);
+ jarUrls.addAll(commonPluginJarUrls);
+ jarUrls.addAll(connectorPluginJarUrls);
+ actions.forEach(
+ action -> {
+ addCommonPluginJarsToAction(
+ action, commonPluginJarUrls,
commonJarIdentifiers);
+ });
+ actions.forEach(
+ action -> {
+ org.apache.seatunnel.engine.core.dag.actions.Config
config =
+ action.getConfig();
+ });
+ } else {
+ jarUrls.addAll(commonPluginJars);
+ jarUrls.addAll(immutablePair.getRight());
+ actions.forEach(
+ action -> {
+ addCommonPluginJarsToAction(
+ action, new HashSet<>(commonPluginJars),
Collections.emptySet());
+ });
+ }
+ return getLogicalDagGenerator().generate();
+ }
+
+ protected Set<ConnectorJarIdentifier> uploadPluginJarUrls(Set<URL>
pluginJarUrls) {
Review Comment:
I have made the changes, PTAL!
##########
seatunnel-engine/seatunnel-engine-client/src/main/java/org/apache/seatunnel/engine/client/job/JobExecutionEnvironment.java:
##########
@@ -64,6 +83,92 @@ protected MultipleTableJobConfigParser getJobConfigParser() {
jobFilePath, idGenerator, jobConfig, commonPluginJars,
isStartWithSavePoint);
}
+ @Override
+ protected LogicalDag getLogicalDag() {
+ ImmutablePair<List<Action>, Set<URL>> immutablePair =
getJobConfigParser().parse();
+ actions.addAll(immutablePair.getLeft());
+ // Enable upload connector jar package to engine server, automatically
upload connector Jar
+ // packages
+ // and dependent third-party Jar packages to the server before job
execution.
+ // Enabling this configuration does not require the server to hold all
connector Jar
+ // packages.
+ boolean enableUploadConnectorJarPackage =
+
seaTunnelConfig.getEngineConfig().getConnectorJarStorageConfig().getEnable();
+ if (enableUploadConnectorJarPackage == true) {
+ /**
+ * TODO: Before uploading the Jar package file the server, first
determine whether the
+ * server holds the current Jar. If the server holds the same Jar
package file, there is
+ * no need for additional uploads.
+ */
+ Set<ConnectorJarIdentifier> commonJarIdentifiers =
+ connectorPackageClient.uploadCommonPluginJars(
+
Long.parseLong(jobConfig.getJobContext().getJobId()), commonPluginJars);
+ Set<URL> commonPluginJarUrls =
getJarUrlsFromIdentifiers(commonJarIdentifiers);
+ Set<ConnectorJarIdentifier> pluginJarIdentifiers = new HashSet<>();
+ transformActionPluginJarUrls(actions, pluginJarIdentifiers);
+ Set<URL> connectorPluginJarUrls =
getJarUrlsFromIdentifiers(pluginJarIdentifiers);
+ connectorJarIdentifiers.addAll(commonJarIdentifiers);
+ connectorJarIdentifiers.addAll(pluginJarIdentifiers);
+ jarUrls.addAll(commonPluginJarUrls);
+ jarUrls.addAll(connectorPluginJarUrls);
+ actions.forEach(
+ action -> {
+ addCommonPluginJarsToAction(
+ action, commonPluginJarUrls,
commonJarIdentifiers);
+ });
+ actions.forEach(
+ action -> {
+ org.apache.seatunnel.engine.core.dag.actions.Config
config =
+ action.getConfig();
+ });
+ } else {
+ jarUrls.addAll(commonPluginJars);
+ jarUrls.addAll(immutablePair.getRight());
+ actions.forEach(
+ action -> {
+ addCommonPluginJarsToAction(
+ action, new HashSet<>(commonPluginJars),
Collections.emptySet());
+ });
+ }
+ return getLogicalDagGenerator().generate();
+ }
+
+ protected Set<ConnectorJarIdentifier> uploadPluginJarUrls(Set<URL>
pluginJarUrls) {
+ Set<ConnectorJarIdentifier> pluginJarIdentifiers = new HashSet<>();
+ pluginJarUrls.forEach(
+ pluginJarUrl -> {
+ /**
+ * TODO: Before uploading the Jar package file the server,
first determine
+ * whether the server holds the current Jar. If the server
holds the same Jar
+ * package file, there is no need for additional uploads.
+ */
+ ConnectorJarIdentifier connectorJarIdentifier =
+ connectorPackageClient.uploadConnectorPluginJar(
+
Long.parseLong(jobConfig.getJobContext().getJobId()),
+ pluginJarUrl);
+ pluginJarIdentifiers.add(connectorJarIdentifier);
+ });
+ return pluginJarIdentifiers;
+ }
+
+ private void transformActionPluginJarUrls(
Review Comment:
I have made the changes, 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
