CalvinKirs commented on code in PR #2301:
URL:
https://github.com/apache/incubator-seatunnel/pull/2301#discussion_r935160118
##########
seatunnel-engine/seatunnel-engine-client/src/main/java/org/apache/seatunnel/engine/client/JobConfigParser.java:
##########
@@ -211,43 +230,57 @@ private void initRelationMap(List<? extends Config>
sourceConfigs, List<? extend
* |
* Sink
*/
- private List<Action> sampleAnalyze(List<? extends Config> sourceConfigs,
- List<? extends Config> transformConfigs,
- List<? extends Config> sinkConfigs) {
- SeaTunnelSource seaTunnelSource =
ConnectorInstanceLoader.loadSourceInstance(sourceConfigs.get(0));
+ private void sampleAnalyze(List<? extends Config> sourceConfigs,
+ List<? extends Config> transformConfigs,
+ List<? extends Config> sinkConfigs) {
+ ImmutablePair<SeaTunnelSource, List<URL>> pair =
+ ConnectorInstanceLoader.loadSourceInstance(sourceConfigs.get(0));
SourceAction sourceAction =
- new SourceAction(idGenerator.getNextId(),
seaTunnelSource.getPluginName(), seaTunnelSource);
+ createSourceAction(idGenerator.getNextId(),
pair.getLeft().getPluginName(), pair.getLeft(),
+ pair.getRight());
sourceAction.setParallelism(getSourceParallelism(sourceConfigs.get(0)));
+
+ ImmutablePair<SeaTunnelSink<SeaTunnelRow, Serializable, Serializable,
Serializable>, List<URL>>
+ sinkListImmutablePair;
+
if (transformConfigs.size() != 0) {
- SeaTunnelTransform seaTunnelTransform =
+ ImmutablePair<SeaTunnelTransform, List<URL>>
transformListImmutablePair =
ConnectorInstanceLoader.loadTransformInstance(transformConfigs.get(0));
- TransformAction transformAction = new TransformAction(
+
+ TransformAction transformAction = createTransformAction(
idGenerator.getNextId(),
- seaTunnelTransform.getPluginName(),
+ transformListImmutablePair.getLeft().getPluginName(),
Lists.newArrayList(sourceAction),
- seaTunnelTransform);
+ transformListImmutablePair.getLeft(),
+ transformListImmutablePair.getRight());
- initTransformParallelism(transformConfigs, sourceAction,
seaTunnelTransform, transformAction);
+ initTransformParallelism(transformConfigs, sourceAction,
transformListImmutablePair.getLeft(),
+ transformAction);
- SeaTunnelSink seaTunnelSink =
ConnectorInstanceLoader.loadSinkInstance(sinkConfigs.get(0));
- SinkAction sinkAction = new SinkAction(
+ sinkListImmutablePair =
ConnectorInstanceLoader.loadSinkInstance(sinkConfigs.get(0));
+
+ SinkAction sinkAction = createSinkAction(
idGenerator.getNextId(),
- seaTunnelSink.getPluginName(),
+ sinkListImmutablePair.getLeft().getPluginName(),
Lists.newArrayList(transformAction),
- seaTunnelSink
+ sinkListImmutablePair.getLeft(),
+ sinkListImmutablePair.getRight()
);
sinkAction.setParallelism(transformAction.getParallelism());
- return Lists.newArrayList(sinkAction);
+ actions.add(sinkAction);
} else {
Review Comment:
We can return directly, so no `else` judgment is needed.
--
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]
