mynameborat commented on a change in pull request #1448:
URL: https://github.com/apache/samza/pull/1448#discussion_r532152731
##########
File path:
samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaYarnAppMasterLifecycle.scala
##########
@@ -48,6 +52,19 @@ class SamzaYarnAppMasterLifecycle(containerMem: Int,
containerCpu: Int, samzaApp
val maxCapability = response.getMaximumResourceCapability
val maxMem = maxCapability.getMemory
val maxCpu = maxCapability.getVirtualCores
+ if (isApplicationMasterHighAvailabilityEnabled) {
+ val yarnIdToprocIdMap = new HashMap[String, String]()
+ samzaAppState.processorToExecutionId.asScala foreach { entry =>
yarnIdToprocIdMap.put(entry._2, entry._1) }
+ response.getContainersFromPreviousAttempts.asScala foreach { (ctr:
Container) =>
+ val samzaProcId = yarnIdToprocIdMap.get(ctr.getId.toString)
+ if (samzaProcId != null) {
+ info("Received container from previous attempt with samza processor
id %s and yarn container id %s" format(samzaProcId, ctr.getId.toString))
+ samzaAppState.runningProcessors.put(samzaProcId,
+ new SamzaResource(ctr.getResource.getVirtualCores,
ctr.getResource.getMemory, ctr.getNodeId.getHost, ctr.getId.toString))
+ state.runningProcessors.put(samzaProcId, new YarnContainer(ctr))
Review comment:
We need to handle the start & uptime within the `YarnContainer`. It
looks like we are using it in the AM web servlet and that will be broken going
forward with this change for containers from previous attempts.
We should either document it or handle it depending on the complexity it
adds to this PR.
##########
File path:
samza-yarn/src/main/java/org/apache/samza/job/yarn/YarnClusterResourceManager.java
##########
@@ -192,7 +193,8 @@ public YarnClusterResourceManager(Config config,
JobModelManager jobModelManager
clusterManagerConfig.getNumCores(),
samzaAppState,
state,
- amClient
+ amClient,
+ new JobConfig(config).getApplicationMasterHighAvailabilityEnabled()
Review comment:
After some digging through, realized we will need to handle populating
allocated resources.
`StopStreamProcessor` method will error out in the absence of that
information.
##########
File path:
samza-core/src/main/java/org/apache/samza/clustermanager/ContainerProcessManager.java
##########
@@ -413,6 +428,9 @@ public void onStreamProcessorLaunchSuccess(SamzaResource
resource) {
processorId, containerId, containerHost);
state.pendingProcessors.remove(processorId);
state.runningProcessors.put(processorId, resource);
+ if (jobConfig.getApplicationMasterHighAvailabilityEnabled()) {
+ state.processorToExecutionId.put(processorId, containerId);
+ }
if (state.neededProcessors.decrementAndGet() == 0) {
state.jobHealthy.set(true);
Review comment:
The current flow doesn't set the job as healthy in the event that all
the containers are from the previous attempt and there is no necessity to spin
up a new one.
##########
File path:
samza-core/src/main/java/org/apache/samza/clustermanager/SamzaApplicationState.java
##########
@@ -105,6 +105,22 @@
*/
public final ConcurrentMap<String, SamzaResource> runningProcessors = new
ConcurrentHashMap<>(0);
+ /**
+ * Map of Samza processor Id (aka logical id) to execution environment
container id (aka physical id ex: yarn container id).
+ * This map will be used during the start up phase of new AM in AM-HA.
+ *
+ * This map is populated at startup of ClusterBasedJobCoordinator.
+ * It initially holds the processId to execution id mapping (if any) present
in the coordinator stream.
+ * This could correspond to containers currently running or from previous
attempt or previous deploy.
+ *
+ * If # of containers in map is same as current JobModel's containers, and
mapping is from previous deploy,
+ * then they will get overwritten by new container incarnations in the
current deploy in {@link
ContainerProcessManager}.onStreamProcessorLaunchSuccess.
+ * If # of containers in map is lesser, then map entries will get
overwritten by current containers and new ones will be added to map.
+ * If # of containers in map is greater, this map is wiped clear in {@link
ContainerProcessManager}.start
+ * to avoid having mapping for containers which are not part of the current
JobModel.
+ */
+ public final ConcurrentMap<String, String> processorToExecutionId = new
ConcurrentHashMap<>(0);
Review comment:
Since we don't need this information beyond the construction phase, I am
going to suggest to not update the information in other places.
Ideally, have it wired during construction although if that doesn't seem
straight forward, we can follow it up after the initial PR.
##########
File path:
samza-core/src/main/java/org/apache/samza/container/ExecutionContainerIdManager.java
##########
@@ -0,0 +1,79 @@
+/*
+ * 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.container;
+
+import com.google.common.base.Preconditions;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.samza.coordinator.stream.CoordinatorStreamValueSerde;
+import
org.apache.samza.coordinator.stream.messages.SetExecutionContainerIdMapping;
+import org.apache.samza.metadatastore.MetadataStore;
+import org.apache.samza.serializers.Serde;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Used for persisting and reading the execution environment container id
information into the metadata store.
+ * Processor id (logical Samza container id) to execution environment
container id (ex: yarn container id) is written.
+ **/
+public class ExecutionContainerIdManager {
+ private static final Logger LOG =
LoggerFactory.getLogger(ExecutionContainerIdManager.class);
+
+ private final Serde<String> valueSerde;
+ private final MetadataStore metadataStore;
+
+ /**
+ * Builds the ExecutionContainerIdManager based upon the provided {@link
MetadataStore} that is instantiated.
+ * Uses the {@link CoordinatorStreamValueSerde} to serialize messages before
reading/writing into metadata store.
+ * @param metadataStore an instance of {@link MetadataStore} to read/write
the container locality.
+ */
+ public ExecutionContainerIdManager(MetadataStore metadataStore) {
+ this.metadataStore = metadataStore;
+ this.valueSerde = new
CoordinatorStreamValueSerde(SetExecutionContainerIdMapping.TYPE);
+ }
+
+ public void writeExecutionEnvironmentContainerIdMapping(String processorId,
String executionEnvContainerId) {
+ Preconditions.checkNotNull(processorId, "Container's logical processor id
can not be null.");
+ Preconditions.checkNotNull(executionEnvContainerId, "Container's physical
execution environment container id can not be null.");
+ LOG.info("Container {} has executionEnvContainerId as {}", processorId,
executionEnvContainerId);
+ metadataStore.put(processorId,
valueSerde.toBytes(executionEnvContainerId));
+ metadataStore.flush();
+ }
+
+ public Map<String, String> readExecutionEnvironmentContainerIdMapping() {
+ Map<String, String> executionEnvironmentContainerIdMapping = new
HashMap<>();
+ metadataStore.all().forEach((processorId, valueBytes) -> {
+ if (valueBytes != null) {
+ String executionEnvContainerId = valueSerde.fromBytes(valueBytes);
+ executionEnvironmentContainerIdMapping.put(processorId,
executionEnvContainerId);
+ }
+ });
+ if (LOG.isDebugEnabled()) {
+ for (Map.Entry<String, String> entry :
executionEnvironmentContainerIdMapping.entrySet()) {
+ LOG.debug("Execution evironment container id for container {}: {}",
entry.getKey(), entry.getValue());
Review comment:
same as above. Be consistent with the log message and replace container
with `processor` and use `executionEnvContainerId` instead.
##########
File path:
samza-core/src/main/java/org/apache/samza/coordinator/stream/messages/SetExecutionContainerIdMapping.java
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.stream.messages;
+
+/**
+ * SetContainerIdExecutionContainerIdMapping is used internally by the Samza
framework to
+ * persist the container processorId-to-executionEnvContainerId mappings.
+ *
+ * Structure of the message looks like:
+ * {
+ * Key: $ContainerId
Review comment:
s/ContainerId/ProcessorId
##########
File path:
samza-core/src/main/java/org/apache/samza/coordinator/stream/messages/SetExecutionContainerIdMapping.java
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.stream.messages;
+
+/**
+ * SetContainerIdExecutionContainerIdMapping is used internally by the Samza
framework to
+ * persist the container processorId-to-executionEnvContainerId mappings.
+ *
+ * Structure of the message looks like:
+ * {
+ * Key: $ContainerId
+ * Type: set-container-id-execution-id-assignment
+ * Source: "SamzaContainer-$ContainerId"
+ * MessageMap:
+ * {
+ * execution-env-container-id: execution environment container id
+ * }
+ * }
+ * */
+public class SetExecutionContainerIdMapping extends CoordinatorStreamMessage {
+ public static final String TYPE = "set-container-id-execution-id-assignment";
Review comment:
`set-execution-container-id-mapping` to be consistent with the class
name similar to others.
##########
File path:
samza-core/src/test/java/org/apache/samza/clustermanager/TestContainerProcessManager.java
##########
@@ -226,6 +227,56 @@ public void run() {
cpm.stop();
}
+ @Test
+ public void testOnInitAMHighAvailability() throws Exception {
+ configVals.put(JobConfig.YARN_AM_HIGH_AVAILABILITY_ENABLED, "true");
Review comment:
Seems like you are modifying a static map here and the remnants of the
test is going to impact other tests in the class.
Can we make a copy instead and add AM-HA configs?
##########
File path:
samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaYarnAppMasterLifecycle.scala
##########
@@ -48,6 +52,19 @@ class SamzaYarnAppMasterLifecycle(containerMem: Int,
containerCpu: Int, samzaApp
val maxCapability = response.getMaximumResourceCapability
val maxMem = maxCapability.getMemory
val maxCpu = maxCapability.getVirtualCores
+ if (isApplicationMasterHighAvailabilityEnabled) {
+ val yarnIdToprocIdMap = new HashMap[String, String]()
+ samzaAppState.processorToExecutionId.asScala foreach { entry =>
yarnIdToprocIdMap.put(entry._2, entry._1) }
Review comment:
Seems like a `BiMap` for `processorToExecutionId` is what is needed and
should eliminate this piece of code here.
##########
File path:
samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaYarnAppMasterLifecycle.scala
##########
@@ -48,6 +52,19 @@ class SamzaYarnAppMasterLifecycle(containerMem: Int,
containerCpu: Int, samzaApp
val maxCapability = response.getMaximumResourceCapability
val maxMem = maxCapability.getMemory
val maxCpu = maxCapability.getVirtualCores
+ if (isApplicationMasterHighAvailabilityEnabled) {
+ val yarnIdToprocIdMap = new HashMap[String, String]()
+ samzaAppState.processorToExecutionId.asScala foreach { entry =>
yarnIdToprocIdMap.put(entry._2, entry._1) }
+ response.getContainersFromPreviousAttempts.asScala foreach { (ctr:
Container) =>
+ val samzaProcId = yarnIdToprocIdMap.get(ctr.getId.toString)
+ if (samzaProcId != null) {
Review comment:
What happens if this is null? Shouldn't non-null be an invariant here?
##########
File path:
samza-yarn/src/main/java/org/apache/samza/job/yarn/YarnClusterResourceManager.java
##########
@@ -192,7 +193,8 @@ public YarnClusterResourceManager(Config config,
JobModelManager jobModelManager
clusterManagerConfig.getNumCores(),
samzaAppState,
state,
- amClient
+ amClient,
+ new JobConfig(config).getApplicationMasterHighAvailabilityEnabled()
Review comment:
Ideally, you want to follow through the `OnContainersAllocated` code
path to ensure that all the state is intact when updating AM with the previous
set of running containers.
We also need to ensure both `YarnAppState` and `SamzaApplicationState` are
up-to-date on top of the other internal state held by
`YarnResourceClusterManager` and `ContainerAllocator` barring the state that is
interim during the container assignment & starting.
##########
File path:
samza-core/src/main/java/org/apache/samza/container/ExecutionContainerIdManager.java
##########
@@ -0,0 +1,79 @@
+/*
+ * 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.container;
+
+import com.google.common.base.Preconditions;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.samza.coordinator.stream.CoordinatorStreamValueSerde;
+import
org.apache.samza.coordinator.stream.messages.SetExecutionContainerIdMapping;
+import org.apache.samza.metadatastore.MetadataStore;
+import org.apache.samza.serializers.Serde;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Used for persisting and reading the execution environment container id
information into the metadata store.
+ * Processor id (logical Samza container id) to execution environment
container id (ex: yarn container id) is written.
+ **/
+public class ExecutionContainerIdManager {
+ private static final Logger LOG =
LoggerFactory.getLogger(ExecutionContainerIdManager.class);
+
+ private final Serde<String> valueSerde;
+ private final MetadataStore metadataStore;
+
+ /**
+ * Builds the ExecutionContainerIdManager based upon the provided {@link
MetadataStore} that is instantiated.
+ * Uses the {@link CoordinatorStreamValueSerde} to serialize messages before
reading/writing into metadata store.
+ * @param metadataStore an instance of {@link MetadataStore} to read/write
the container locality.
+ */
+ public ExecutionContainerIdManager(MetadataStore metadataStore) {
+ this.metadataStore = metadataStore;
+ this.valueSerde = new
CoordinatorStreamValueSerde(SetExecutionContainerIdMapping.TYPE);
+ }
+
+ public void writeExecutionEnvironmentContainerIdMapping(String processorId,
String executionEnvContainerId) {
+ Preconditions.checkNotNull(processorId, "Container's logical processor id
can not be null.");
+ Preconditions.checkNotNull(executionEnvContainerId, "Container's physical
execution environment container id can not be null.");
+ LOG.info("Container {} has executionEnvContainerId as {}", processorId,
executionEnvContainerId);
Review comment:
nit: `Processor: {} has executionEnvContainerId: {}` to be consistent in
other places. I remember Prateek made a whole clean up PR for this :)
##########
File path:
samza-core/src/main/java/org/apache/samza/coordinator/stream/messages/SetExecutionContainerIdMapping.java
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.stream.messages;
+
+/**
+ * SetContainerIdExecutionContainerIdMapping is used internally by the Samza
framework to
+ * persist the container processorId-to-executionEnvContainerId mappings.
Review comment:
s/persist the container/persist
##########
File path:
samza-core/src/main/java/org/apache/samza/coordinator/stream/messages/SetExecutionContainerIdMapping.java
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.stream.messages;
+
+/**
+ * SetContainerIdExecutionContainerIdMapping is used internally by the Samza
framework to
+ * persist the container processorId-to-executionEnvContainerId mappings.
+ *
+ * Structure of the message looks like:
+ * {
+ * Key: $ContainerId
+ * Type: set-container-id-execution-id-assignment
+ * Source: "SamzaContainer-$ContainerId"
+ * MessageMap:
+ * {
+ * execution-env-container-id: execution environment container id
+ * }
+ * }
+ * */
+public class SetExecutionContainerIdMapping extends CoordinatorStreamMessage {
Review comment:
`SetExecutionEnvContainerIdMapping` instead of
`SetExecutionContainerIdMapping` to be consistent in usage.
##########
File path:
samza-core/src/main/java/org/apache/samza/runtime/ContainerLaunchUtil.java
##########
@@ -147,6 +149,14 @@ private static void run(
heartbeatMonitor.start();
}
+ if (new JobConfig(config).getApplicationMasterHighAvailabilityEnabled())
{
+ ExecutionContainerIdManager executionContainerIdManager = new
ExecutionContainerIdManager(
+ new NamespaceAwareCoordinatorStreamStore(coordinatorStreamStore,
SetExecutionContainerIdMapping.TYPE));
+ if (executionContainerIdManager != null &&
execEnvContainerId.isPresent()) {
Review comment:
we just create the manager above. why is this null check? should be
`execEnvContainerId` instead.
I still think you don't need this check. You can use `Optional::ifPresent`
and write it.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
