You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by GitBox <gi...@apache.org> on 2021/08/11 21:49:30 UTC
[GitHub] [samza] cameronlee314 opened a new pull request #1515: [SAMZA-2667] Refactor logic in JobModelManager so it is easier to use certain parts of it
cameronlee314 opened a new pull request #1515:
URL: https://github.com/apache/samza/pull/1515
Issues:
1. `JobModelManager` has a public static method just for calculating `JobModel` (i.e. `readJobModel`) but it also has a factory method (i.e. `apply`) which creates the context necessary for calculating `JobModel`. Some components only use `readJobModel` while others use `apply`. This makes the class usage somewhat inconsistent and unclean.
2. `JobModelManager.apply` is directly coupled with an `HttpServer` for communicating with workers. This means any other logic in `JobModelManager.apply` is tied to `HttpServer`, which may not be desirable in the future. There will be a follow-up PR which decouples `JobModel` creation with the communication layer.
3. `JobModelManager` is written in scala, which is harder to maintain with the significant java usage in most of the rest of Samza.
Changes:
1. Extract `JobModelManager.readJobModel` into a separate class `JobModelCalculator` which is only responsible for calculating `JobModel`. Convert this logic from scala to java.
2. Extract some of the logic of `JobModelManager.apply` (get grouper metadata, update task assignments) into a separate class `JobModelHelper` so that logic is not tied to `HttpServer` for communication. Convert this logic from scala to java.
3. Fix a bug regarding deletion of standby task assignments. Assignments should get cleared when the old set of standby tasks is not the same as the new set of standby tasks, but that was not happening because the condition was checking for equality and it wasn't comparing the same type of objects.
4. Add more tests around the job model logic.
Tests:
1. Wrote several more unit tests for the logic that got refactored. I was able to run the tests against the old logic as well to make sure they passed before and after the refactoring (the tests involving standbys did fail with the old logic due to the bug described in Change 3 above, and they pass with the new logic).
API changes and upgrade/usage instructions: N/A
--
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: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] mynameborat commented on a change in pull request #1515: [SAMZA-2667] Refactor logic in JobModelManager so it is easier to use certain parts of it
Posted by GitBox <gi...@apache.org>.
mynameborat commented on a change in pull request #1515:
URL: https://github.com/apache/samza/pull/1515#discussion_r694222949
##########
File path: samza-core/src/main/java/org/apache/samza/coordinator/JobModelCalculator.java
##########
@@ -0,0 +1,239 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.samza.coordinator;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.function.Function;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+import org.apache.samza.Partition;
+import org.apache.samza.config.ClusterManagerConfig;
+import org.apache.samza.config.Config;
+import org.apache.samza.config.ConfigException;
+import org.apache.samza.config.JobConfig;
+import org.apache.samza.config.MapConfig;
+import org.apache.samza.config.RegExTopicGenerator;
+import org.apache.samza.config.TaskConfig;
+import org.apache.samza.container.TaskName;
+import org.apache.samza.container.grouper.stream.SSPGrouperProxy;
+import org.apache.samza.container.grouper.stream.SystemStreamPartitionGrouper;
+import org.apache.samza.container.grouper.stream.SystemStreamPartitionGrouperFactory;
+import org.apache.samza.container.grouper.task.GrouperMetadata;
+import org.apache.samza.container.grouper.task.TaskNameGrouperFactory;
+import org.apache.samza.container.grouper.task.TaskNameGrouperProxy;
+import org.apache.samza.job.model.ContainerModel;
+import org.apache.samza.job.model.JobModel;
+import org.apache.samza.job.model.TaskModel;
+import org.apache.samza.system.StreamMetadataCache;
+import org.apache.samza.system.SystemStream;
+import org.apache.samza.system.SystemStreamMetadata;
+import org.apache.samza.system.SystemStreamPartition;
+import org.apache.samza.system.SystemStreamPartitionMatcher;
+import org.apache.samza.util.ConfigUtil;
+import org.apache.samza.util.ReflectionUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.collection.JavaConverters;
+
+
+public class JobModelCalculator {
Review comment:
It seemed like a utility class and imo, there isn't any reason on why this needed to be extendable (through inheritance) and hence marking it as final explicitly enforces the above.
IIRC, marking it as final is tad bit optimized because of the additional information on no extenders, which is useful to make these function calls inline.
--
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: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] mynameborat commented on a change in pull request #1515: [SAMZA-2667] Refactor logic in JobModelManager so it is easier to use certain parts of it
Posted by GitBox <gi...@apache.org>.
mynameborat commented on a change in pull request #1515:
URL: https://github.com/apache/samza/pull/1515#discussion_r694013961
##########
File path: samza-core/src/main/java/org/apache/samza/coordinator/JobModelCalculator.java
##########
@@ -0,0 +1,239 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.samza.coordinator;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.function.Function;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+import org.apache.samza.Partition;
+import org.apache.samza.config.ClusterManagerConfig;
+import org.apache.samza.config.Config;
+import org.apache.samza.config.ConfigException;
+import org.apache.samza.config.JobConfig;
+import org.apache.samza.config.MapConfig;
+import org.apache.samza.config.RegExTopicGenerator;
+import org.apache.samza.config.TaskConfig;
+import org.apache.samza.container.TaskName;
+import org.apache.samza.container.grouper.stream.SSPGrouperProxy;
+import org.apache.samza.container.grouper.stream.SystemStreamPartitionGrouper;
+import org.apache.samza.container.grouper.stream.SystemStreamPartitionGrouperFactory;
+import org.apache.samza.container.grouper.task.GrouperMetadata;
+import org.apache.samza.container.grouper.task.TaskNameGrouperFactory;
+import org.apache.samza.container.grouper.task.TaskNameGrouperProxy;
+import org.apache.samza.job.model.ContainerModel;
+import org.apache.samza.job.model.JobModel;
+import org.apache.samza.job.model.TaskModel;
+import org.apache.samza.system.StreamMetadataCache;
+import org.apache.samza.system.SystemStream;
+import org.apache.samza.system.SystemStreamMetadata;
+import org.apache.samza.system.SystemStreamPartition;
+import org.apache.samza.system.SystemStreamPartitionMatcher;
+import org.apache.samza.util.ConfigUtil;
+import org.apache.samza.util.ReflectionUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.collection.JavaConverters;
+
+
+public class JobModelCalculator {
Review comment:
mark final
##########
File path: samza-azure/src/main/java/org/apache/samza/coordinator/AzureJobCoordinator.java
##########
@@ -367,7 +367,8 @@ private void doOnProcessorChange(List<String> currentProcessorIds) {
// Generate the new JobModel
GrouperMetadata grouperMetadata = new GrouperMetadataImpl(Collections.emptyMap(), Collections.emptyMap(), Collections.emptyMap(), Collections.emptyMap());
JobModel newJobModel =
- JobModelManager.readJobModel(this.config, Collections.emptyMap(), streamMetadataCache, grouperMetadata);
+ JobModelCalculator.INSTANCE.calculateJobModel(this.config, Collections.emptyMap(), streamMetadataCache,
+ grouperMetadata);
Review comment:
Any reason behind this pattern of having an instance instead of treating this as a utility class and having static methods?
##########
File path: samza-core/src/main/java/org/apache/samza/coordinator/JobModelCalculator.java
##########
@@ -0,0 +1,239 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.samza.coordinator;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.function.Function;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+import org.apache.samza.Partition;
+import org.apache.samza.config.ClusterManagerConfig;
+import org.apache.samza.config.Config;
+import org.apache.samza.config.ConfigException;
+import org.apache.samza.config.JobConfig;
+import org.apache.samza.config.MapConfig;
+import org.apache.samza.config.RegExTopicGenerator;
+import org.apache.samza.config.TaskConfig;
+import org.apache.samza.container.TaskName;
+import org.apache.samza.container.grouper.stream.SSPGrouperProxy;
+import org.apache.samza.container.grouper.stream.SystemStreamPartitionGrouper;
+import org.apache.samza.container.grouper.stream.SystemStreamPartitionGrouperFactory;
+import org.apache.samza.container.grouper.task.GrouperMetadata;
+import org.apache.samza.container.grouper.task.TaskNameGrouperFactory;
+import org.apache.samza.container.grouper.task.TaskNameGrouperProxy;
+import org.apache.samza.job.model.ContainerModel;
+import org.apache.samza.job.model.JobModel;
+import org.apache.samza.job.model.TaskModel;
+import org.apache.samza.system.StreamMetadataCache;
+import org.apache.samza.system.SystemStream;
+import org.apache.samza.system.SystemStreamMetadata;
+import org.apache.samza.system.SystemStreamPartition;
+import org.apache.samza.system.SystemStreamPartitionMatcher;
+import org.apache.samza.util.ConfigUtil;
+import org.apache.samza.util.ReflectionUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.collection.JavaConverters;
+
+
+public class JobModelCalculator {
+ private static final Logger LOG = LoggerFactory.getLogger(JobModelCalculator.class);
+ public static final JobModelCalculator INSTANCE = new JobModelCalculator();
+
+ private JobModelCalculator() {
+ }
+
+ /**
+ * Does the following:
+ * 1. Fetches metadata of the input streams defined in configuration through {@code streamMetadataCache}.
+ * 2. Applies the SSP grouper and task name grouper defined in the configuration to build the {@link JobModel}.
+ * @param originalConfig the configuration of the job.
+ * @param changeLogPartitionMapping the task to changelog partition mapping of the job.
+ * @param streamMetadataCache the cache that holds the partition metadata of the input streams.
+ * @param grouperMetadata provides the historical metadata of the application.
+ * @return the built {@link JobModel}.
+ */
+ public JobModel calculateJobModel(Config originalConfig, Map<TaskName, Integer> changeLogPartitionMapping,
+ StreamMetadataCache streamMetadataCache, GrouperMetadata grouperMetadata) {
+ // refresh config if enabled regex topic rewriter
+ Config config = refreshConfigByRegexTopicRewriter(originalConfig);
+
+ TaskConfig taskConfig = new TaskConfig(config);
+ // Do grouping to fetch TaskName to SSP mapping
+ Set<SystemStreamPartition> allSystemStreamPartitions = getMatchedInputStreamPartitions(config, streamMetadataCache);
+
+ // processor list is required by some of the groupers. So, let's pass them as part of the config.
+ // Copy the config and add the processor list to the config copy.
+ // TODO: It is non-ideal to have config as a medium to transmit the locality information; especially, if the locality information evolves. Evaluate options on using context objects to pass dependent components.
+ Map<String, String> configMap = new HashMap<>(config);
+ configMap.put(JobConfig.PROCESSOR_LIST, String.join(",", grouperMetadata.getProcessorLocality().keySet()));
+ SystemStreamPartitionGrouper grouper = getSystemStreamPartitionGrouper(new MapConfig(configMap));
+
+ JobConfig jobConfig = new JobConfig(config);
+
+ Map<TaskName, Set<SystemStreamPartition>> groups;
+ if (jobConfig.isSSPGrouperProxyEnabled()) {
+ SSPGrouperProxy sspGrouperProxy = new SSPGrouperProxy(config, grouper);
+ groups = sspGrouperProxy.group(allSystemStreamPartitions, grouperMetadata);
+ } else {
+ LOG.warn(String.format(
+ "SSPGrouperProxy is disabled (%s = false). Stateful jobs may produce erroneous results if this is not enabled.",
+ JobConfig.SSP_INPUT_EXPANSION_ENABLED));
+ groups = grouper.group(allSystemStreamPartitions);
+ }
+ LOG.info(String.format(
+ "SystemStreamPartitionGrouper %s has grouped the SystemStreamPartitions into %d tasks with the following taskNames: %s",
+ grouper, groups.size(), groups));
+
+ // If no mappings are present (first time the job is running) we return -1, this will allow 0 to be the first change
+ // mapping.
+ int maxChangelogPartitionId = changeLogPartitionMapping.values().stream().max(Comparator.naturalOrder()).orElse(-1);
+ // Sort the groups prior to assigning the changelog mapping so that the mapping is reproducible and intuitive
+ TreeMap<TaskName, Set<SystemStreamPartition>> sortedGroups = new TreeMap<>(groups);
+ Set<TaskModel> taskModels = new HashSet<>();
+ for (Map.Entry<TaskName, Set<SystemStreamPartition>> group : sortedGroups.entrySet()) {
+ TaskName taskName = group.getKey();
+ Set<SystemStreamPartition> systemStreamPartitions = group.getValue();
+ Optional<Integer> changelogPartitionId = Optional.ofNullable(changeLogPartitionMapping.get(taskName));
+ Partition changelogPartition;
+ if (changelogPartitionId.isPresent()) {
+ changelogPartition = new Partition(changelogPartitionId.get());
+ } else {
+ // If we've never seen this TaskName before, then assign it a new changelog partition.
+ maxChangelogPartitionId++;
+ LOG.info(
+ String.format("New task %s is being assigned changelog partition %s.", taskName, maxChangelogPartitionId));
+ changelogPartition = new Partition(maxChangelogPartitionId);
+ }
+ taskModels.add(new TaskModel(taskName, systemStreamPartitions, changelogPartition));
+ }
+
+ // Here is where we should put in a pluggable option for the SSPTaskNameGrouper for locality, load-balancing, etc.
+ TaskNameGrouperFactory containerGrouperFactory =
+ ReflectionUtil.getObj(taskConfig.getTaskNameGrouperFactory(), TaskNameGrouperFactory.class);
+ boolean standbyTasksEnabled = jobConfig.getStandbyTasksEnabled();
+ int standbyTaskReplicationFactor = jobConfig.getStandbyTaskReplicationFactor();
+ TaskNameGrouperProxy taskNameGrouperProxy =
+ new TaskNameGrouperProxy(containerGrouperFactory.build(config), standbyTasksEnabled,
+ standbyTaskReplicationFactor);
+ Set<ContainerModel> containerModels;
+ boolean isHostAffinityEnabled = new ClusterManagerConfig(config).getHostAffinityEnabled();
+ if (isHostAffinityEnabled) {
+ containerModels = taskNameGrouperProxy.group(taskModels, grouperMetadata);
+ } else {
+ containerModels =
+ taskNameGrouperProxy.group(taskModels, new ArrayList<>(grouperMetadata.getProcessorLocality().keySet()));
+ }
+
+ Map<String, ContainerModel> containerMap =
+ containerModels.stream().collect(Collectors.toMap(ContainerModel::getId, Function.identity()));
+ return new JobModel(config, containerMap);
+ }
+
+ /**
+ * Refresh Kafka topic list used as input streams if enabled {@link org.apache.samza.config.RegExTopicGenerator}
+ * @param config Samza job config
+ * @return refreshed config
+ */
+ private static Config refreshConfigByRegexTopicRewriter(Config config) {
Review comment:
Just to be consistent with the above, can we rename the parameter to `originalConfig`?
##########
File path: samza-core/src/main/java/org/apache/samza/coordinator/JobModelCalculator.java
##########
@@ -0,0 +1,239 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.samza.coordinator;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.function.Function;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+import org.apache.samza.Partition;
+import org.apache.samza.config.ClusterManagerConfig;
+import org.apache.samza.config.Config;
+import org.apache.samza.config.ConfigException;
+import org.apache.samza.config.JobConfig;
+import org.apache.samza.config.MapConfig;
+import org.apache.samza.config.RegExTopicGenerator;
+import org.apache.samza.config.TaskConfig;
+import org.apache.samza.container.TaskName;
+import org.apache.samza.container.grouper.stream.SSPGrouperProxy;
+import org.apache.samza.container.grouper.stream.SystemStreamPartitionGrouper;
+import org.apache.samza.container.grouper.stream.SystemStreamPartitionGrouperFactory;
+import org.apache.samza.container.grouper.task.GrouperMetadata;
+import org.apache.samza.container.grouper.task.TaskNameGrouperFactory;
+import org.apache.samza.container.grouper.task.TaskNameGrouperProxy;
+import org.apache.samza.job.model.ContainerModel;
+import org.apache.samza.job.model.JobModel;
+import org.apache.samza.job.model.TaskModel;
+import org.apache.samza.system.StreamMetadataCache;
+import org.apache.samza.system.SystemStream;
+import org.apache.samza.system.SystemStreamMetadata;
+import org.apache.samza.system.SystemStreamPartition;
+import org.apache.samza.system.SystemStreamPartitionMatcher;
+import org.apache.samza.util.ConfigUtil;
+import org.apache.samza.util.ReflectionUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.collection.JavaConverters;
+
+
+public class JobModelCalculator {
+ private static final Logger LOG = LoggerFactory.getLogger(JobModelCalculator.class);
+ public static final JobModelCalculator INSTANCE = new JobModelCalculator();
+
+ private JobModelCalculator() {
+ }
+
+ /**
+ * Does the following:
+ * 1. Fetches metadata of the input streams defined in configuration through {@code streamMetadataCache}.
+ * 2. Applies the SSP grouper and task name grouper defined in the configuration to build the {@link JobModel}.
+ * @param originalConfig the configuration of the job.
+ * @param changeLogPartitionMapping the task to changelog partition mapping of the job.
+ * @param streamMetadataCache the cache that holds the partition metadata of the input streams.
+ * @param grouperMetadata provides the historical metadata of the application.
+ * @return the built {@link JobModel}.
+ */
+ public JobModel calculateJobModel(Config originalConfig, Map<TaskName, Integer> changeLogPartitionMapping,
+ StreamMetadataCache streamMetadataCache, GrouperMetadata grouperMetadata) {
+ // refresh config if enabled regex topic rewriter
+ Config config = refreshConfigByRegexTopicRewriter(originalConfig);
Review comment:
nit: maybe rename `expandedConfig` or `refreshedConfig`
--
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: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] alnzng commented on a change in pull request #1515: [SAMZA-2667] Refactor logic in JobModelManager so it is easier to use certain parts of it
Posted by GitBox <gi...@apache.org>.
alnzng commented on a change in pull request #1515:
URL: https://github.com/apache/samza/pull/1515#discussion_r690524892
##########
File path: samza-core/src/main/java/org/apache/samza/coordinator/JobModelHelper.java
##########
@@ -0,0 +1,217 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.samza.coordinator;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.samza.config.Config;
+import org.apache.samza.config.JobConfig;
+import org.apache.samza.container.LocalityManager;
+import org.apache.samza.container.TaskName;
+import org.apache.samza.container.grouper.task.GrouperMetadata;
+import org.apache.samza.container.grouper.task.GrouperMetadataImpl;
+import org.apache.samza.container.grouper.task.TaskAssignmentManager;
+import org.apache.samza.container.grouper.task.TaskPartitionAssignmentManager;
+import org.apache.samza.job.model.ContainerModel;
+import org.apache.samza.job.model.JobModel;
+import org.apache.samza.job.model.ProcessorLocality;
+import org.apache.samza.job.model.TaskMode;
+import org.apache.samza.job.model.TaskModel;
+import org.apache.samza.runtime.LocationId;
+import org.apache.samza.system.StreamMetadataCache;
+import org.apache.samza.system.SystemStreamPartition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class JobModelHelper {
+ private static final Logger LOG = LoggerFactory.getLogger(JobModelHelper.class);
+
+ private final LocalityManager localityManager;
+ private final TaskAssignmentManager taskAssignmentManager;
+ private final TaskPartitionAssignmentManager taskPartitionAssignmentManager;
+ private final StreamMetadataCache streamMetadataCache;
+ private final JobModelCalculator jobModelCalculator;
+
+ public JobModelHelper(LocalityManager localityManager, TaskAssignmentManager taskAssignmentManager,
+ TaskPartitionAssignmentManager taskPartitionAssignmentManager, StreamMetadataCache streamMetadataCache,
+ JobModelCalculator jobModelCalculator) {
+ this.localityManager = localityManager;
+ this.taskAssignmentManager = taskAssignmentManager;
+ this.taskPartitionAssignmentManager = taskPartitionAssignmentManager;
+ this.streamMetadataCache = streamMetadataCache;
+ this.jobModelCalculator = jobModelCalculator;
+ }
+
+ public JobModel newJobModel(Config config, Map<TaskName, Integer> changelogPartitionMapping) {
+ GrouperMetadata grouperMetadata = getGrouperMetadata(config, this.localityManager, this.taskAssignmentManager,
+ this.taskPartitionAssignmentManager);
+ JobModel jobModel =
+ this.jobModelCalculator.calculateJobModel(config, changelogPartitionMapping, this.streamMetadataCache,
+ grouperMetadata);
+ updateTaskAssignments(jobModel, this.taskAssignmentManager, this.taskPartitionAssignmentManager, grouperMetadata);
+ return jobModel;
+ }
+
+ private GrouperMetadata getGrouperMetadata(Config config, LocalityManager localityManager,
+ TaskAssignmentManager taskAssignmentManager, TaskPartitionAssignmentManager taskPartitionAssignmentManager) {
+ Map<String, LocationId> processorLocality = getProcessorLocality(config, localityManager);
+ Map<TaskName, TaskMode> taskModes = taskAssignmentManager.readTaskModes();
+
+ Map<TaskName, String> taskNameToProcessorId = new HashMap<>();
+ Map<TaskName, LocationId> taskLocality = new HashMap<>();
+ // We read the taskAssignment only for ActiveTasks, i.e., tasks that have no task-mode or have an active task mode
+ taskAssignmentManager.readTaskAssignment().forEach((taskNameString, containerId) -> {
+ TaskName taskName = new TaskName(taskNameString);
+ if (isActiveTask(taskName, taskModes)) {
+ taskNameToProcessorId.put(taskName, containerId);
+ if (processorLocality.containsKey(containerId)) {
+ taskLocality.put(taskName, processorLocality.get(containerId));
+ }
+ }
+ });
+
+ Map<SystemStreamPartition, List<String>> sspToTaskMapping =
+ taskPartitionAssignmentManager.readTaskPartitionAssignments();
+ Map<TaskName, List<SystemStreamPartition>> taskPartitionAssignments = new HashMap<>();
+ // Task to partition assignments is stored as {@see SystemStreamPartition} to list of {@see TaskName} in
+ // coordinator stream. This is done due to the 1 MB value size limit in a kafka topic. Conversion to
+ // taskName to SystemStreamPartitions is done here to wire-in the data to {@see JobModel}.
+ sspToTaskMapping.forEach((systemStreamPartition, taskNames) -> taskNames.forEach(taskNameString -> {
+ TaskName taskName = new TaskName(taskNameString);
+ if (isActiveTask(taskName, taskModes)) {
+ taskPartitionAssignments.putIfAbsent(taskName, new ArrayList<>());
+ taskPartitionAssignments.get(taskName).add(systemStreamPartition);
+ }
+ }));
+ return new GrouperMetadataImpl(processorLocality, taskLocality, taskPartitionAssignments, taskNameToProcessorId);
+ }
+
+ /**
+ * Retrieves and returns the processor locality of a samza job using provided {@see Config} and {@see LocalityManager}.
+ * @param config provides the configurations defined by the user. Required to connect to the storage layer.
+ * @param localityManager provides the processor to host mapping persisted to the metadata store.
+ * @return the processor locality.
+ */
+ private static Map<String, LocationId> getProcessorLocality(Config config, LocalityManager localityManager) {
+ Map<String, LocationId> containerToLocationId = new HashMap<>();
+ Map<String, ProcessorLocality> existingContainerLocality = localityManager.readLocality().getProcessorLocalities();
+
+ for (int i = 0; i < new JobConfig(config).getContainerCount(); i++) {
+ String containerId = Integer.toString(i);
+ LocationId locationId = Optional.ofNullable(existingContainerLocality.get(containerId))
+ .map(ProcessorLocality::host)
+ .filter(StringUtils::isNotEmpty)
+ .map(LocationId::new)
+ // To handle the case when the container count is increased between two different runs of a samza-yarn job,
+ // set the locality of newly added containers to any_host.
+ .orElse(new LocationId("ANY_HOST"));
+ containerToLocationId.put(containerId, locationId);
+ }
+ return containerToLocationId;
+ }
+
+ /**
+ * This method does the following:
+ * 1. Deletes the existing task assignments if the partition-task grouping has changed from the previous run of the job.
+ * 2. Saves the newly generated task assignments to the storage layer through the {@param TaskAssignementManager}.
+ *
+ * @param jobModel represents the {@see JobModel} of the samza job.
+ * @param taskAssignmentManager required to persist the processor to task assignments to the metadata store.
+ * @param taskPartitionAssignmentManager required to persist the task to partition assignments to the metadata store.
+ * @param grouperMetadata provides the historical metadata of the samza application.
+ */
+ private void updateTaskAssignments(JobModel jobModel, TaskAssignmentManager taskAssignmentManager,
+ TaskPartitionAssignmentManager taskPartitionAssignmentManager, GrouperMetadata grouperMetadata) {
+ LOG.info("Storing the task assignments into metadata store.");
+ Set<String> activeTaskNames = new HashSet<>();
+ Set<String> standbyTaskNames = new HashSet<>();
+ Set<SystemStreamPartition> systemStreamPartitions = new HashSet<>();
+
+ for (ContainerModel containerModel : jobModel.getContainers().values()) {
+ for (TaskModel taskModel : containerModel.getTasks().values()) {
+ if (TaskMode.Active.equals(taskModel.getTaskMode())) {
+ activeTaskNames.add(taskModel.getTaskName().getTaskName());
+ }
+ if (TaskMode.Standby.equals(taskModel.getTaskMode())) {
+ standbyTaskNames.add(taskModel.getTaskName().getTaskName());
+ }
+ systemStreamPartitions.addAll(taskModel.getSystemStreamPartitions());
+ }
+ }
+
+ Map<TaskName, String> previousTaskToContainerId = grouperMetadata.getPreviousTaskToProcessorAssignment();
+ if (activeTaskNames.size() != previousTaskToContainerId.size()) {
+ LOG.warn(String.format(
+ "Current task count %s does not match saved task count %s. Stateful jobs may observe misalignment of keys!",
+ activeTaskNames.size(), previousTaskToContainerId.size()));
+ // If the tasks changed, then the partition-task grouping is also likely changed and we can't handle that
+ // without a much more complicated mapping. Further, the partition count may have changed, which means
+ // input message keys are likely reshuffled w.r.t. partitions, so the local state may not contain necessary
+ // data associated with the incoming keys. Warn the user and default to grouper
+ // In this scenario the tasks may have been reduced, so we need to delete all the existing messages
+ taskAssignmentManager.deleteTaskContainerMappings(
+ previousTaskToContainerId.keySet().stream().map(TaskName::getTaskName).collect(Collectors.toList()));
+ taskPartitionAssignmentManager.delete(systemStreamPartitions);
+ }
+
+ // if the set of standby tasks has changed, e.g., when the replication-factor changed, or the active-tasks-set has
+ // changed, we log a warning and delete the existing mapping for these tasks
+ Set<String> previousStandbyTasks = taskAssignmentManager.readTaskModes()
+ .entrySet()
+ .stream()
+ .filter(taskNameToTaskModeEntry -> TaskMode.Standby.equals(taskNameToTaskModeEntry.getValue()))
+ .map(taskNameToTaskModeEntry -> taskNameToTaskModeEntry.getKey().getTaskName())
+ .collect(Collectors.toSet());
+ if (!standbyTaskNames.equals(previousStandbyTasks)) {
Review comment:
Interesting bug, thanks for the 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: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] cameronlee314 merged pull request #1515: [SAMZA-2667] Refactor logic in JobModelManager so it is easier to use certain parts of it
Posted by GitBox <gi...@apache.org>.
cameronlee314 merged pull request #1515:
URL: https://github.com/apache/samza/pull/1515
--
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: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] cameronlee314 commented on a change in pull request #1515: [SAMZA-2667] Refactor logic in JobModelManager so it is easier to use certain parts of it
Posted by GitBox <gi...@apache.org>.
cameronlee314 commented on a change in pull request #1515:
URL: https://github.com/apache/samza/pull/1515#discussion_r694177181
##########
File path: samza-azure/src/main/java/org/apache/samza/coordinator/AzureJobCoordinator.java
##########
@@ -367,7 +367,8 @@ private void doOnProcessorChange(List<String> currentProcessorIds) {
// Generate the new JobModel
GrouperMetadata grouperMetadata = new GrouperMetadataImpl(Collections.emptyMap(), Collections.emptyMap(), Collections.emptyMap(), Collections.emptyMap());
JobModel newJobModel =
- JobModelManager.readJobModel(this.config, Collections.emptyMap(), streamMetadataCache, grouperMetadata);
+ JobModelCalculator.INSTANCE.calculateJobModel(this.config, Collections.emptyMap(), streamMetadataCache,
+ grouperMetadata);
Review comment:
It is easier to do dependency injection this way for testing purposes.
For this class, the call is just treated like a static utility method call, but other classes can inject a `JobModelCalculator` so that it can be mocked out in testing.
##########
File path: samza-core/src/main/java/org/apache/samza/coordinator/JobModelCalculator.java
##########
@@ -0,0 +1,239 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.samza.coordinator;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.function.Function;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+import org.apache.samza.Partition;
+import org.apache.samza.config.ClusterManagerConfig;
+import org.apache.samza.config.Config;
+import org.apache.samza.config.ConfigException;
+import org.apache.samza.config.JobConfig;
+import org.apache.samza.config.MapConfig;
+import org.apache.samza.config.RegExTopicGenerator;
+import org.apache.samza.config.TaskConfig;
+import org.apache.samza.container.TaskName;
+import org.apache.samza.container.grouper.stream.SSPGrouperProxy;
+import org.apache.samza.container.grouper.stream.SystemStreamPartitionGrouper;
+import org.apache.samza.container.grouper.stream.SystemStreamPartitionGrouperFactory;
+import org.apache.samza.container.grouper.task.GrouperMetadata;
+import org.apache.samza.container.grouper.task.TaskNameGrouperFactory;
+import org.apache.samza.container.grouper.task.TaskNameGrouperProxy;
+import org.apache.samza.job.model.ContainerModel;
+import org.apache.samza.job.model.JobModel;
+import org.apache.samza.job.model.TaskModel;
+import org.apache.samza.system.StreamMetadataCache;
+import org.apache.samza.system.SystemStream;
+import org.apache.samza.system.SystemStreamMetadata;
+import org.apache.samza.system.SystemStreamPartition;
+import org.apache.samza.system.SystemStreamPartitionMatcher;
+import org.apache.samza.util.ConfigUtil;
+import org.apache.samza.util.ReflectionUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.collection.JavaConverters;
+
+
+public class JobModelCalculator {
Review comment:
Could you please clarify the benefit of marking `final` here? For testing purposes, `final` classes are harder to work with.
--
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: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] alnzng commented on a change in pull request #1515: [SAMZA-2667] Refactor logic in JobModelManager so it is easier to use certain parts of it
Posted by GitBox <gi...@apache.org>.
alnzng commented on a change in pull request #1515:
URL: https://github.com/apache/samza/pull/1515#discussion_r690524892
##########
File path: samza-core/src/main/java/org/apache/samza/coordinator/JobModelHelper.java
##########
@@ -0,0 +1,217 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.samza.coordinator;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.samza.config.Config;
+import org.apache.samza.config.JobConfig;
+import org.apache.samza.container.LocalityManager;
+import org.apache.samza.container.TaskName;
+import org.apache.samza.container.grouper.task.GrouperMetadata;
+import org.apache.samza.container.grouper.task.GrouperMetadataImpl;
+import org.apache.samza.container.grouper.task.TaskAssignmentManager;
+import org.apache.samza.container.grouper.task.TaskPartitionAssignmentManager;
+import org.apache.samza.job.model.ContainerModel;
+import org.apache.samza.job.model.JobModel;
+import org.apache.samza.job.model.ProcessorLocality;
+import org.apache.samza.job.model.TaskMode;
+import org.apache.samza.job.model.TaskModel;
+import org.apache.samza.runtime.LocationId;
+import org.apache.samza.system.StreamMetadataCache;
+import org.apache.samza.system.SystemStreamPartition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class JobModelHelper {
+ private static final Logger LOG = LoggerFactory.getLogger(JobModelHelper.class);
+
+ private final LocalityManager localityManager;
+ private final TaskAssignmentManager taskAssignmentManager;
+ private final TaskPartitionAssignmentManager taskPartitionAssignmentManager;
+ private final StreamMetadataCache streamMetadataCache;
+ private final JobModelCalculator jobModelCalculator;
+
+ public JobModelHelper(LocalityManager localityManager, TaskAssignmentManager taskAssignmentManager,
+ TaskPartitionAssignmentManager taskPartitionAssignmentManager, StreamMetadataCache streamMetadataCache,
+ JobModelCalculator jobModelCalculator) {
+ this.localityManager = localityManager;
+ this.taskAssignmentManager = taskAssignmentManager;
+ this.taskPartitionAssignmentManager = taskPartitionAssignmentManager;
+ this.streamMetadataCache = streamMetadataCache;
+ this.jobModelCalculator = jobModelCalculator;
+ }
+
+ public JobModel newJobModel(Config config, Map<TaskName, Integer> changelogPartitionMapping) {
+ GrouperMetadata grouperMetadata = getGrouperMetadata(config, this.localityManager, this.taskAssignmentManager,
+ this.taskPartitionAssignmentManager);
+ JobModel jobModel =
+ this.jobModelCalculator.calculateJobModel(config, changelogPartitionMapping, this.streamMetadataCache,
+ grouperMetadata);
+ updateTaskAssignments(jobModel, this.taskAssignmentManager, this.taskPartitionAssignmentManager, grouperMetadata);
+ return jobModel;
+ }
+
+ private GrouperMetadata getGrouperMetadata(Config config, LocalityManager localityManager,
+ TaskAssignmentManager taskAssignmentManager, TaskPartitionAssignmentManager taskPartitionAssignmentManager) {
+ Map<String, LocationId> processorLocality = getProcessorLocality(config, localityManager);
+ Map<TaskName, TaskMode> taskModes = taskAssignmentManager.readTaskModes();
+
+ Map<TaskName, String> taskNameToProcessorId = new HashMap<>();
+ Map<TaskName, LocationId> taskLocality = new HashMap<>();
+ // We read the taskAssignment only for ActiveTasks, i.e., tasks that have no task-mode or have an active task mode
+ taskAssignmentManager.readTaskAssignment().forEach((taskNameString, containerId) -> {
+ TaskName taskName = new TaskName(taskNameString);
+ if (isActiveTask(taskName, taskModes)) {
+ taskNameToProcessorId.put(taskName, containerId);
+ if (processorLocality.containsKey(containerId)) {
+ taskLocality.put(taskName, processorLocality.get(containerId));
+ }
+ }
+ });
+
+ Map<SystemStreamPartition, List<String>> sspToTaskMapping =
+ taskPartitionAssignmentManager.readTaskPartitionAssignments();
+ Map<TaskName, List<SystemStreamPartition>> taskPartitionAssignments = new HashMap<>();
+ // Task to partition assignments is stored as {@see SystemStreamPartition} to list of {@see TaskName} in
+ // coordinator stream. This is done due to the 1 MB value size limit in a kafka topic. Conversion to
+ // taskName to SystemStreamPartitions is done here to wire-in the data to {@see JobModel}.
+ sspToTaskMapping.forEach((systemStreamPartition, taskNames) -> taskNames.forEach(taskNameString -> {
+ TaskName taskName = new TaskName(taskNameString);
+ if (isActiveTask(taskName, taskModes)) {
+ taskPartitionAssignments.putIfAbsent(taskName, new ArrayList<>());
+ taskPartitionAssignments.get(taskName).add(systemStreamPartition);
+ }
+ }));
+ return new GrouperMetadataImpl(processorLocality, taskLocality, taskPartitionAssignments, taskNameToProcessorId);
+ }
+
+ /**
+ * Retrieves and returns the processor locality of a samza job using provided {@see Config} and {@see LocalityManager}.
+ * @param config provides the configurations defined by the user. Required to connect to the storage layer.
+ * @param localityManager provides the processor to host mapping persisted to the metadata store.
+ * @return the processor locality.
+ */
+ private static Map<String, LocationId> getProcessorLocality(Config config, LocalityManager localityManager) {
+ Map<String, LocationId> containerToLocationId = new HashMap<>();
+ Map<String, ProcessorLocality> existingContainerLocality = localityManager.readLocality().getProcessorLocalities();
+
+ for (int i = 0; i < new JobConfig(config).getContainerCount(); i++) {
+ String containerId = Integer.toString(i);
+ LocationId locationId = Optional.ofNullable(existingContainerLocality.get(containerId))
+ .map(ProcessorLocality::host)
+ .filter(StringUtils::isNotEmpty)
+ .map(LocationId::new)
+ // To handle the case when the container count is increased between two different runs of a samza-yarn job,
+ // set the locality of newly added containers to any_host.
+ .orElse(new LocationId("ANY_HOST"));
+ containerToLocationId.put(containerId, locationId);
+ }
+ return containerToLocationId;
+ }
+
+ /**
+ * This method does the following:
+ * 1. Deletes the existing task assignments if the partition-task grouping has changed from the previous run of the job.
+ * 2. Saves the newly generated task assignments to the storage layer through the {@param TaskAssignementManager}.
+ *
+ * @param jobModel represents the {@see JobModel} of the samza job.
+ * @param taskAssignmentManager required to persist the processor to task assignments to the metadata store.
+ * @param taskPartitionAssignmentManager required to persist the task to partition assignments to the metadata store.
+ * @param grouperMetadata provides the historical metadata of the samza application.
+ */
+ private void updateTaskAssignments(JobModel jobModel, TaskAssignmentManager taskAssignmentManager,
+ TaskPartitionAssignmentManager taskPartitionAssignmentManager, GrouperMetadata grouperMetadata) {
+ LOG.info("Storing the task assignments into metadata store.");
+ Set<String> activeTaskNames = new HashSet<>();
+ Set<String> standbyTaskNames = new HashSet<>();
+ Set<SystemStreamPartition> systemStreamPartitions = new HashSet<>();
+
+ for (ContainerModel containerModel : jobModel.getContainers().values()) {
+ for (TaskModel taskModel : containerModel.getTasks().values()) {
+ if (TaskMode.Active.equals(taskModel.getTaskMode())) {
+ activeTaskNames.add(taskModel.getTaskName().getTaskName());
+ }
+ if (TaskMode.Standby.equals(taskModel.getTaskMode())) {
+ standbyTaskNames.add(taskModel.getTaskName().getTaskName());
+ }
+ systemStreamPartitions.addAll(taskModel.getSystemStreamPartitions());
+ }
+ }
+
+ Map<TaskName, String> previousTaskToContainerId = grouperMetadata.getPreviousTaskToProcessorAssignment();
+ if (activeTaskNames.size() != previousTaskToContainerId.size()) {
+ LOG.warn(String.format(
+ "Current task count %s does not match saved task count %s. Stateful jobs may observe misalignment of keys!",
+ activeTaskNames.size(), previousTaskToContainerId.size()));
+ // If the tasks changed, then the partition-task grouping is also likely changed and we can't handle that
+ // without a much more complicated mapping. Further, the partition count may have changed, which means
+ // input message keys are likely reshuffled w.r.t. partitions, so the local state may not contain necessary
+ // data associated with the incoming keys. Warn the user and default to grouper
+ // In this scenario the tasks may have been reduced, so we need to delete all the existing messages
+ taskAssignmentManager.deleteTaskContainerMappings(
+ previousTaskToContainerId.keySet().stream().map(TaskName::getTaskName).collect(Collectors.toList()));
+ taskPartitionAssignmentManager.delete(systemStreamPartitions);
+ }
+
+ // if the set of standby tasks has changed, e.g., when the replication-factor changed, or the active-tasks-set has
+ // changed, we log a warning and delete the existing mapping for these tasks
+ Set<String> previousStandbyTasks = taskAssignmentManager.readTaskModes()
+ .entrySet()
+ .stream()
+ .filter(taskNameToTaskModeEntry -> TaskMode.Standby.equals(taskNameToTaskModeEntry.getValue()))
+ .map(taskNameToTaskModeEntry -> taskNameToTaskModeEntry.getKey().getTaskName())
+ .collect(Collectors.toSet());
+ if (!standbyTaskNames.equals(previousStandbyTasks)) {
Review comment:
Interesting bug, thanks for the 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: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] alnzng commented on pull request #1515: [SAMZA-2667] Refactor logic in JobModelManager so it is easier to use certain parts of it
Posted by GitBox <gi...@apache.org>.
alnzng commented on pull request #1515:
URL: https://github.com/apache/samza/pull/1515#issuecomment-900614120
> Thanks for refactoring the codes and adding more test cases, the code changes look good to me.
>
> One thing you may want to double check is that I saw OOM happened when running the command locally: `./gradlew build -PrunIntegrationTests`:
>
> ```
> > Task :samza-core_2.11:test
>
> Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "RMI RenewClean-[alazhang-mn1.linkedin.biz:56375]"
>
> Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "RMI RenewClean-[alazhang-mn1.linkedin.biz:56656]"
>
> Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "Thread-156"
>
> executionError FAILED
> java.lang.OutOfMemoryError: Java heap space
>
> 1446 tests completed, 1 failed, 4 skipped
>
> > Task :samza-core_2.11:test FAILED
>
> FAILURE: Build failed with an exception.
> ```
Tried to rebuild on the existing project and also tried to fresh checkout, the build passed on both of them, I didn't see this OOM problem anymore. Also @cameronlee314 tried on his local env and he didn't observe any problem.
For now, I would think it is an unreproducible problem, it could be caused by some noise in my local env. The problem should not be caused by these code changes.
So, this PR looks good to me.
--
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: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] alnzng commented on pull request #1515: [SAMZA-2667] Refactor logic in JobModelManager so it is easier to use certain parts of it
Posted by GitBox <gi...@apache.org>.
alnzng commented on pull request #1515:
URL: https://github.com/apache/samza/pull/1515#issuecomment-900614120
> Thanks for refactoring the codes and adding more test cases, the code changes look good to me.
>
> One thing you may want to double check is that I saw OOM happened when running the command locally: `./gradlew build -PrunIntegrationTests`:
>
> ```
> > Task :samza-core_2.11:test
>
> Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "RMI RenewClean-[alazhang-mn1.linkedin.biz:56375]"
>
> Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "RMI RenewClean-[alazhang-mn1.linkedin.biz:56656]"
>
> Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "Thread-156"
>
> executionError FAILED
> java.lang.OutOfMemoryError: Java heap space
>
> 1446 tests completed, 1 failed, 4 skipped
>
> > Task :samza-core_2.11:test FAILED
>
> FAILURE: Build failed with an exception.
> ```
Tried to rebuild on the existing project and also tried to fresh checkout, the build passed on both of them, I didn't see this OOM problem anymore. Also @cameronlee314 tried on his local env and he didn't observe any problem.
For now, I would think it is an unreproducible problem, it could be caused by some noise in my local env. The problem should not be caused by these code changes.
So, this PR looks good to me.
--
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: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] cameronlee314 commented on a change in pull request #1515: [SAMZA-2667] Refactor logic in JobModelManager so it is easier to use certain parts of it
Posted by GitBox <gi...@apache.org>.
cameronlee314 commented on a change in pull request #1515:
URL: https://github.com/apache/samza/pull/1515#discussion_r694384924
##########
File path: samza-core/src/main/java/org/apache/samza/coordinator/JobModelCalculator.java
##########
@@ -0,0 +1,239 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.samza.coordinator;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.function.Function;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+import org.apache.samza.Partition;
+import org.apache.samza.config.ClusterManagerConfig;
+import org.apache.samza.config.Config;
+import org.apache.samza.config.ConfigException;
+import org.apache.samza.config.JobConfig;
+import org.apache.samza.config.MapConfig;
+import org.apache.samza.config.RegExTopicGenerator;
+import org.apache.samza.config.TaskConfig;
+import org.apache.samza.container.TaskName;
+import org.apache.samza.container.grouper.stream.SSPGrouperProxy;
+import org.apache.samza.container.grouper.stream.SystemStreamPartitionGrouper;
+import org.apache.samza.container.grouper.stream.SystemStreamPartitionGrouperFactory;
+import org.apache.samza.container.grouper.task.GrouperMetadata;
+import org.apache.samza.container.grouper.task.TaskNameGrouperFactory;
+import org.apache.samza.container.grouper.task.TaskNameGrouperProxy;
+import org.apache.samza.job.model.ContainerModel;
+import org.apache.samza.job.model.JobModel;
+import org.apache.samza.job.model.TaskModel;
+import org.apache.samza.system.StreamMetadataCache;
+import org.apache.samza.system.SystemStream;
+import org.apache.samza.system.SystemStreamMetadata;
+import org.apache.samza.system.SystemStreamPartition;
+import org.apache.samza.system.SystemStreamPartitionMatcher;
+import org.apache.samza.util.ConfigUtil;
+import org.apache.samza.util.ReflectionUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.collection.JavaConverters;
+
+
+public class JobModelCalculator {
Review comment:
This will be injected as a dependency into a component in an upcoming PR. It's used like a utility class in these current places for compatibility with existing usage.
For testing, it is harder to mock `final` classes, so I'd like to leave this as non-final for the future tests for when this is injected as a dependency instead of using a static method call.
--
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: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org