hailin0 commented on code in PR #4147:
URL:
https://github.com/apache/incubator-seatunnel/pull/4147#discussion_r1111400803
##########
seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/execution/ExecutionPlanGenerator.java:
##########
@@ -88,17 +96,79 @@ public class ExecutionPlanGenerator {
private final List<LogicalVertex> logicalVertices;
private final JobImmutableInformation jobImmutableInformation;
+ private final CheckpointConfig checkpointConfig;
public ExecutionPlanGenerator(@NonNull LogicalDag logicalPlan,
- @NonNull JobImmutableInformation
jobImmutableInformation) {
+ @NonNull JobImmutableInformation
jobImmutableInformation,
+ @NonNull CheckpointConfig checkpointConfig) {
checkArgument(logicalPlan.getEdges().size() > 0, "ExecutionPlan
Builder must have LogicalPlan.");
this.logicalEdges = new ArrayList<>(logicalPlan.getEdges());
this.logicalVertices = new
ArrayList<>(logicalPlan.getLogicalVertexMap().values());
this.jobImmutableInformation = jobImmutableInformation;
+ this.checkpointConfig = checkpointConfig;
+ }
+
+ @SuppressWarnings("MagicNumber")
+ private void generateShuffleEdges() {
+ List<LogicalEdge> newLogicalEdges = new ArrayList<>();
+ List<LogicalVertex> shuffleLogicalVertexs = new ArrayList<>();
+ for (LogicalEdge logicalEdge : logicalEdges) {
+ LogicalVertex inputVertex = logicalEdge.getInputVertex();
+ LogicalVertex targetVertex = logicalEdge.getTargetVertex();
+ Action inpuetAction = inputVertex.getAction();
+ Action targetAction = targetVertex.getAction();
+ if (inpuetAction instanceof SourceAction && targetAction
instanceof SinkAction) {
+ SeaTunnelSource<?, ?, ?> seaTunnelSource = ((SourceAction<?,
?, ?>) inpuetAction).getSource();
+ SeaTunnelDataType<?> sourceProducedType =
seaTunnelSource.getProducedType();
+ // cdc multiple-tables
+ if
(SqlType.MULTIPLE_ROW.equals(sourceProducedType.getSqlType())) {
+ SinkConfig sinkConfig = ((SinkAction)
targetAction).getConfig();
+ ShuffleConfig.ShuffleStrategy shuffleStrategy = new
ShuffleConfig.MultipleRowShuffleStrategy(
+ jobImmutableInformation.getJobId(),
+ inpuetAction.getParallelism(),
+ MultipleRowType.class.cast(sourceProducedType),
+ targetAction.getParallelism(),
+ sinkConfig.getUpstreamTableId(),
+ new QueueConfig()
+ .setMaxSize(ShuffleConfig.DEFAULT_QUEUE_SIZE)
+
.setBackupCount(ShuffleConfig.DEFAULT_QUEUE_BACKUP_COUNT)
+
.setAsyncBackupCount(ShuffleConfig.DEFAULT_QUEUE_ASYNC_BACKUP_COUNT)
+ .setEmptyQueueTtl((int)
(checkpointConfig.getCheckpointInterval() * 3)));
+ ShuffleConfig shuffleConfig = new ShuffleConfig()
+ .setBatchSize(ShuffleConfig.DEFAULT_BATCH_SIZE)
+
.setBatchFlushInterval(ShuffleConfig.DEFAULT_BATCH_FLUSH_INTERVAL)
+ .setShuffleStrategy(shuffleStrategy);
+
+ long shuffleVertexId = idGenerator.getNextId();
Review Comment:
Generate duplicate id, waiting for fix
--
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]
