Hisoka-X commented on code in PR #7045:
URL: https://github.com/apache/seatunnel/pull/7045#discussion_r1650271855
##########
seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalPlanGenerator.java:
##########
@@ -150,7 +151,8 @@ public PhysicalPlanGenerator(
}
public Tuple2<PhysicalPlan, Map<Integer, CheckpointPlan>> generate() {
-
+ Map<String, Object> envOptions =
jobImmutableInformation.getJobConfig().getEnvOptions();
+ Map<String, String> tags =
JsonUtils.toMap(envOptions.getOrDefault("tag", "").toString());
Review Comment:
Please add `tag` into
https://github.com/apache/seatunnel/blob/476d492165ab2747aa0bb1a5f8a6ed74bfe55f20/seatunnel-api/src/main/java/org/apache/seatunnel/api/env/EnvCommonOptions.java#L27
##########
seatunnel-engine/seatunnel-engine-common/src/main/resources/hazelcast.yaml:
##########
@@ -39,3 +39,10 @@ hazelcast:
hazelcast.tcp.join.port.try.count: 30
hazelcast.logging.type: log4j2
hazelcast.operation.generic.thread.count: 50
+ member-attributes:
+ group:
+ type: string
+ value: platform
+ team:
+ type: string
+ value: team1
Review Comment:
Hi, please do not change this. I believe this feature should not open by
default.
##########
seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/resourcemanager/AbstractResourceManager.java:
##########
@@ -129,9 +131,45 @@ public void memberRemoved(MembershipServiceEvent event) {
@Override
public CompletableFuture<List<SlotProfile>> applyResources(
- long jobId, List<ResourceProfile> resourceProfile) throws
NoEnoughResourceException {
+ long jobId, List<ResourceProfile> resourceProfile, Map<String,
String> tags)
+ throws NoEnoughResourceException {
waitingWorkerRegister();
- return new ResourceRequestHandler(jobId, resourceProfile,
registerWorker, this).request();
+ ConcurrentMap<Address, WorkerProfile> matchedWorker;
+ if (tags == null || tags.isEmpty()) {
+ matchedWorker = registerWorker;
+ } else {
+ matchedWorker =
+ registerWorker.entrySet().stream()
+ .filter(
+ e -> {
+ Map<String, String> workerAttr =
+ e.getValue().getAttributes();
+ if (workerAttr == null ||
workerAttr.isEmpty()) {
+ return true;
+ }
Review Comment:
Why workerAttr is empty it will be matched?
--
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]
