This is an automated email from the ASF dual-hosted git repository.
bharathkk pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/samza.git
The following commit(s) were added to refs/heads/master by this push:
new 5748fa6 SAMZA-2605: Make Standby Container Requests Rack Aware
(#1446)
5748fa6 is described below
commit 5748fa6adec729840beef03495aa6b74f739f017
Author: Pawas Chhokra <pchho...@linkedin.com>
AuthorDate: Thu Dec 24 11:20:54 2020 -0800
SAMZA-2605: Make Standby Container Requests Rack Aware (#1446)
Feature: The aim of this feature is to make all standby container requests
rack aware such that all active containers and their corresponding standby
containers are always on different racks. This helps with decreased downtime of
applications during rack failures.
One of the requirements of this feature is that the value of
job.standbytasks.replication.factor is at max 2 for the rack awareness
functionality to be honored.
Changes: This PR uses the FaultDomainManager interface for Yarn to request
for rack aware nodes while making standby container requests.
Usage Instructions: For a job with host affinity and standby containers,
set the config cluster-manager.fault-domain-aware.standby.enabled to true to
enable this feature.
---
.../samza/clustermanager/ContainerAllocator.java | 53 +++++++++
.../samza/clustermanager/ContainerManager.java | 6 +-
.../clustermanager/ContainerProcessManager.java | 27 ++++-
.../clustermanager/StandbyContainerManager.java | 130 ++++++++++++++++++---
.../TestClusterBasedJobCoordinator.java | 1 +
.../TestContainerAllocatorWithHostAffinity.java | 11 +-
.../TestContainerAllocatorWithoutHostAffinity.java | 9 +-
.../TestContainerPlacementActions.java | 15 ++-
.../TestContainerProcessManager.java | 33 +++---
.../samza/job/yarn/YarnClusterResourceManager.java | 7 +-
10 files changed, 239 insertions(+), 53 deletions(-)
diff --git
a/samza-core/src/main/java/org/apache/samza/clustermanager/ContainerAllocator.java
b/samza-core/src/main/java/org/apache/samza/clustermanager/ContainerAllocator.java
index fa5f783..88be21f 100644
---
a/samza-core/src/main/java/org/apache/samza/clustermanager/ContainerAllocator.java
+++
b/samza-core/src/main/java/org/apache/samza/clustermanager/ContainerAllocator.java
@@ -22,6 +22,7 @@ import java.time.Duration;
import java.time.Instant;
import java.util.Map;
import java.util.Optional;
+import java.util.Set;
import org.apache.samza.SamzaException;
import org.apache.samza.config.ClusterManagerConfig;
import org.apache.samza.config.Config;
@@ -348,6 +349,16 @@ public class ContainerAllocator implements Runnable {
}
/**
+ * Requests a resource from the cluster manager
+ * @param processorId Samza processor ID that will be run when a resource is
allocated for this request
+ * @param preferredHost name of the host that you prefer to run the
processor on
+ * @param faultDomains set of fault domains on which to schedule this
resource
+ */
+ public final void requestResource(String processorId, String preferredHost,
Set<FaultDomain> faultDomains) {
+ requestResourceWithDelay(processorId, preferredHost, Duration.ZERO,
faultDomains);
+ }
+
+ /**
* Requests a resource from the cluster manager with a request timestamp of
the current time plus the specified delay.
* @param processorId Samza processor ID that will be run when a resource is
allocated for this request
* @param preferredHost name of the host that you prefer to run the
processor on
@@ -359,6 +370,18 @@ public class ContainerAllocator implements Runnable {
}
/**
+ * Requests a resource from the cluster manager with a request timestamp of
the current time plus the specified delay.
+ * @param processorId Samza processor ID that will be run when a resource is
allocated for this request
+ * @param preferredHost name of the host that you prefer to run the
processor on
+ * @param delay the {@link Duration} to add to the request timestamp
+ * @param faultDomains set of fault domains on which to schedule this
resource
+ */
+ public final void requestResourceWithDelay(String processorId, String
preferredHost, Duration delay, Set<FaultDomain> faultDomains) {
+ SamzaResourceRequest request = getResourceRequestWithDelay(processorId,
preferredHost, delay, faultDomains);
+ issueResourceRequest(request);
+ }
+
+ /**
* Creates a {@link SamzaResourceRequest} to send to the cluster manager
* @param processorId Samza processor ID that will be run when a resource is
allocated for this request
* @param preferredHost name of the host that you prefer to run the
processor on
@@ -369,6 +392,17 @@ public class ContainerAllocator implements Runnable {
}
/**
+ * Creates a {@link SamzaResourceRequest} to send to the cluster manager
+ * @param processorId Samza processor ID that will be run when a resource is
allocated for this request
+ * @param preferredHost name of the host that you prefer to run the
processor on
+ * @param faultDomains set of fault domains on which to schedule this
resource
+ * @return the created request
+ */
+ public final SamzaResourceRequest getResourceRequest(String processorId,
String preferredHost, Set<FaultDomain> faultDomains) {
+ return getResourceRequestWithDelay(processorId, preferredHost,
Duration.ZERO, faultDomains);
+ }
+
+ /**
* Creates a {@link SamzaResourceRequest} to send to the cluster manager
with a request timestamp of the current time
* plus the specified delay.
* @param processorId Samza processor ID that will be run when a resource is
allocated for this request
@@ -380,6 +414,19 @@ public class ContainerAllocator implements Runnable {
return new SamzaResourceRequest(this.containerNumCpuCores,
this.containerMemoryMb, preferredHost, processorId, Instant.now().plus(delay));
}
+ /**
+ * Creates a {@link SamzaResourceRequest} to send to the cluster manager
with a request timestamp of the current time
+ * plus the specified delay.
+ * @param processorId Samza processor ID that will be run when a resource is
allocated for this request
+ * @param preferredHost name of the host that you prefer to run the
processor on
+ * @param delay the {@link Duration} to add to the request timestamp
+ * @param faultDomains set of fault domains on which to schedule this
resource
+ * @return the created request
+ */
+ public final SamzaResourceRequest getResourceRequestWithDelay(String
processorId, String preferredHost, Duration delay, Set<FaultDomain>
faultDomains) {
+ return new SamzaResourceRequest(this.containerNumCpuCores,
this.containerMemoryMb, preferredHost, processorId, Instant.now().plus(delay),
faultDomains);
+ }
+
public final void issueResourceRequest(SamzaResourceRequest request) {
resourceRequestState.addResourceRequest(request);
state.containerRequests.incrementAndGet();
@@ -388,6 +435,9 @@ public class ContainerAllocator implements Runnable {
} else {
state.preferredHostRequests.incrementAndGet();
}
+ if (!request.getFaultDomains().isEmpty()) {
+ state.faultDomainAwareContainerRequests.incrementAndGet();
+ }
}
/**
@@ -480,5 +530,8 @@ public class ContainerAllocator implements Runnable {
} else {
state.expiredPreferredHostRequests.incrementAndGet();
}
+ if (!request.getFaultDomains().isEmpty()) {
+ state.expiredFaultDomainAwareContainerRequests.incrementAndGet();
+ }
}
}
diff --git
a/samza-core/src/main/java/org/apache/samza/clustermanager/ContainerManager.java
b/samza-core/src/main/java/org/apache/samza/clustermanager/ContainerManager.java
index 24130fd..5fcf328 100644
---
a/samza-core/src/main/java/org/apache/samza/clustermanager/ContainerManager.java
+++
b/samza-core/src/main/java/org/apache/samza/clustermanager/ContainerManager.java
@@ -28,6 +28,7 @@ import org.apache.commons.lang3.tuple.ImmutablePair;
import org.apache.commons.lang3.tuple.Pair;
import
org.apache.samza.clustermanager.container.placement.ContainerPlacementMetadataStore;
import
org.apache.samza.clustermanager.container.placement.ContainerPlacementMetadata;
+import org.apache.samza.config.Config;
import org.apache.samza.container.LocalityManager;
import org.apache.samza.container.placement.ContainerPlacementMessage;
import org.apache.samza.container.placement.ContainerPlacementRequestMessage;
@@ -88,8 +89,9 @@ public class ContainerManager {
public ContainerManager(ContainerPlacementMetadataStore
containerPlacementMetadataStore,
SamzaApplicationState samzaApplicationState, ClusterResourceManager
clusterResourceManager,
- boolean hostAffinityEnabled, boolean standByEnabled, LocalityManager
localityManager) {
+ boolean hostAffinityEnabled, boolean standByEnabled, LocalityManager
localityManager, FaultDomainManager faultDomainManager, Config config) {
Preconditions.checkNotNull(localityManager, "Locality manager cannot be
null");
+ Preconditions.checkNotNull(faultDomainManager, "Fault domain manager
cannot be null");
this.samzaApplicationState = samzaApplicationState;
this.clusterResourceManager = clusterResourceManager;
this.actions = new ConcurrentHashMap<>();
@@ -100,7 +102,7 @@ public class ContainerManager {
// Enable standby container manager if required
if (standByEnabled) {
this.standbyContainerManager =
- Optional.of(new StandbyContainerManager(samzaApplicationState,
clusterResourceManager, localityManager));
+ Optional.of(new StandbyContainerManager(samzaApplicationState,
clusterResourceManager, localityManager, config, faultDomainManager));
} else {
this.standbyContainerManager = Optional.empty();
}
diff --git
a/samza-core/src/main/java/org/apache/samza/clustermanager/ContainerProcessManager.java
b/samza-core/src/main/java/org/apache/samza/clustermanager/ContainerProcessManager.java
index 995cf7d..143e0b3 100644
---
a/samza-core/src/main/java/org/apache/samza/clustermanager/ContainerProcessManager.java
+++
b/samza-core/src/main/java/org/apache/samza/clustermanager/ContainerProcessManager.java
@@ -149,6 +149,9 @@ public class ContainerProcessManager implements
ClusterResourceManager.Callback
ResourceManagerFactory factory =
getContainerProcessManagerFactory(clusterManagerConfig);
this.clusterResourceManager =
checkNotNull(factory.getClusterResourceManager(this, state));
+ FaultDomainManagerFactory faultDomainManagerFactory =
getFaultDomainManagerFactory(clusterManagerConfig);
+ FaultDomainManager faultDomainManager =
checkNotNull(faultDomainManagerFactory.getFaultDomainManager(config, registry));
+
// Initialize metrics
this.containerProcessManagerMetrics = new
ContainerProcessManagerMetrics(config, state, registry);
this.jvmMetrics = new JvmMetrics(registry);
@@ -172,8 +175,8 @@ public class ContainerProcessManager implements
ClusterResourceManager.Callback
// Wire all metrics to all reporters
this.metricsReporters.values().forEach(reporter ->
reporter.register(METRICS_SOURCE_NAME, registry));
- this.containerManager = new ContainerManager(metadataStore, state,
clusterResourceManager, hostAffinityEnabled,
- jobConfig.getStandbyTasksEnabled(), localityManager);
+ this.containerManager = new ContainerManager(metadataStore, state,
clusterResourceManager,
+ hostAffinityEnabled, jobConfig.getStandbyTasksEnabled(),
localityManager, faultDomainManager, config);
this.containerAllocator = new
ContainerAllocator(this.clusterResourceManager, config, state,
hostAffinityEnabled, this.containerManager);
this.allocatorThread = new Thread(this.containerAllocator, "Container
Allocator Thread");
@@ -649,6 +652,26 @@ public class ContainerProcessManager implements
ClusterResourceManager.Callback
}
/**
+ * Returns an instantiated {@link FaultDomainManagerFactory} from a {@link
ClusterManagerConfig}. The
+ * {@link FaultDomainManagerFactory} is used to return an implementation of
a {@link FaultDomainManager}
+ *
+ * @param clusterManagerConfig, the cluster manager config to parse.
+ *
+ */
+ private FaultDomainManagerFactory getFaultDomainManagerFactory(final
ClusterManagerConfig clusterManagerConfig) {
+ final String faultDomainManagerFactoryClass =
clusterManagerConfig.getFaultDomainManagerClass();
+ final FaultDomainManagerFactory factory;
+
+ try {
+ factory = ReflectionUtil.getObj(faultDomainManagerFactoryClass,
FaultDomainManagerFactory.class);
+ } catch (Exception e) {
+ LOG.error("Error creating the fault domain manager.", e);
+ throw new SamzaException(e);
+ }
+ return factory;
+ }
+
+ /**
* Obtains the ID of the Samza processor pending launch on the provided
resource (container).
*
* ContainerProcessManager [INFO] Container ID:
container_e66_1569376389369_0221_01_000049 matched pending Processor ID: 0 on
host: ltx1-app0772.stg.linkedin.com
diff --git
a/samza-core/src/main/java/org/apache/samza/clustermanager/StandbyContainerManager.java
b/samza-core/src/main/java/org/apache/samza/clustermanager/StandbyContainerManager.java
index b849ea5..a07a924 100644
---
a/samza-core/src/main/java/org/apache/samza/clustermanager/StandbyContainerManager.java
+++
b/samza-core/src/main/java/org/apache/samza/clustermanager/StandbyContainerManager.java
@@ -19,6 +19,7 @@
package org.apache.samza.clustermanager;
import java.time.Duration;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -28,6 +29,8 @@ import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.commons.lang3.StringUtils;
import org.apache.samza.SamzaException;
+import org.apache.samza.config.ClusterManagerConfig;
+import org.apache.samza.config.Config;
import org.apache.samza.container.LocalityManager;
import org.apache.samza.job.model.ProcessorLocality;
import org.apache.samza.job.model.JobModel;
@@ -56,8 +59,13 @@ public class StandbyContainerManager {
// Resource-manager, used to stop containers
private ClusterResourceManager clusterResourceManager;
- public StandbyContainerManager(SamzaApplicationState samzaApplicationState,
- ClusterResourceManager clusterResourceManager, LocalityManager
localityManager) {
+ // FaultDomainManager, used to get fault domain information of different
hosts from the cluster manager.
+ private final FaultDomainManager faultDomainManager;
+
+ private final boolean isFaultDomainAwareStandbyEnabled;
+
+ public StandbyContainerManager(SamzaApplicationState samzaApplicationState,
ClusterResourceManager clusterResourceManager,
+ LocalityManager localityManager, Config
config, FaultDomainManager faultDomainManager) {
this.failovers = new ConcurrentHashMap<>();
this.localityManager = localityManager;
this.standbyContainerConstraints = new HashMap<>();
@@ -70,6 +78,9 @@ public class StandbyContainerManager {
.forEach(containerId -> standbyContainerConstraints.put(containerId,
StandbyTaskUtil.getStandbyContainerConstraints(containerId,
jobModel)));
this.clusterResourceManager = clusterResourceManager;
+ this.faultDomainManager = faultDomainManager;
+ ClusterManagerConfig clusterManagerConfig = new
ClusterManagerConfig(config);
+ this.isFaultDomainAwareStandbyEnabled =
clusterManagerConfig.getFaultDomainAwareStandbyEnabled();
log.info("Populated standbyContainerConstraints map {}",
standbyContainerConstraints);
}
@@ -125,7 +136,9 @@ public class StandbyContainerManager {
if (StandbyTaskUtil.isStandbyContainer(containerID)) {
log.info("Handling launch fail for standby-container {}, requesting
resource on any host {}", containerID);
- containerAllocator.requestResource(containerID,
ResourceRequestState.ANY_HOST);
+ String activeContainerHost = getActiveContainerHost(containerID)
+ .orElse(null);
+ requestResource(containerAllocator, containerID,
ResourceRequestState.ANY_HOST, Duration.ZERO, activeContainerHost);
} else {
initiateStandbyAwareAllocation(containerID, resourceID,
containerAllocator);
}
@@ -157,6 +170,22 @@ public class StandbyContainerManager {
}
/**
+ * This method removes the fault domain of the host passed as an argument,
from the set of fault domains, and then returns it.
+ * The set of fault domains returned is based on the set difference between
all the available fault domains in the
+ * cluster and the fault domain associated with the host that is passed as
input.
+ * @param hostToAvoid hostname whose fault domains are excluded
+ * @return The set of fault domains which excludes the fault domain that the
given host is on
+ */
+ public Set<FaultDomain> getAllowedFaultDomainsGivenHostToAvoid(String
hostToAvoid) {
+ Set<FaultDomain> allFaultDomains = faultDomainManager.getAllFaultDomains();
+ Set<FaultDomain> faultDomainToAvoid = Optional.ofNullable(hostToAvoid)
+ .map(faultDomainManager::getFaultDomainsForHost)
+ .orElse(Collections.emptySet());
+ allFaultDomains.removeAll(faultDomainToAvoid);
+ return allFaultDomains;
+ }
+
+ /**
* If a standby container has stopped, then there are two possible cases
* Case 1. during a failover, the standby container was stopped for an
active's start, then we
* 1. request a resource on the standby's host to place the
activeContainer, and
@@ -181,24 +210,29 @@ public class StandbyContainerManager {
// request standbycontainer's host for active-container
SamzaResourceRequest resourceRequestForActive =
- containerAllocator.getResourceRequestWithDelay(activeContainerID,
standbyContainerHostname, preferredHostRetryDelay);
+ containerAllocator.getResourceRequestWithDelay(activeContainerID,
standbyContainerHostname, preferredHostRetryDelay);
// record the resource request, before issuing it to avoid race with
allocation-thread
failoverMetadata.get().recordResourceRequest(resourceRequestForActive);
containerAllocator.issueResourceRequest(resourceRequestForActive);
// request any-host for standby container
- containerAllocator.requestResource(standbyContainerID,
ResourceRequestState.ANY_HOST);
+ requestResource(containerAllocator, standbyContainerID,
ResourceRequestState.ANY_HOST, Duration.ZERO, standbyContainerHostname);
} else {
log.info("Issuing request for standby container {} on host {}, since
this is not for a failover",
standbyContainerID, preferredHost);
- containerAllocator.requestResourceWithDelay(standbyContainerID,
preferredHost, preferredHostRetryDelay);
+ String activeContainerHost = getActiveContainerHost(standbyContainerID)
+ .orElse(null);
+ requestResource(containerAllocator, standbyContainerID, preferredHost,
preferredHostRetryDelay, activeContainerHost);
}
}
/** Method to handle standby-aware allocation for an active container.
* We try to find a standby host for the active container, and issue a stop
on any standby-containers running on it,
* request resource to place the active on the standby's host, and one to
place the standby elsewhere.
- *
+ * When requesting for resources,
+ * NOTE: When rack awareness is turned on, we always pass the
<code>hostToAvoid</> parameter as null for the {@link #requestResource} method
used here
+ * because the hostname of the previous active processor that died does not
exist in the running or pending container list anymore.
+ * However, different racks will always be guaranteed through {@link
#checkStandbyConstraintsAndRunStreamProcessor}.
* @param activeContainerID the samzaContainerID of the active-container
* @param resourceID the samza-resource-ID of the container when it failed
(used to index failover-state)
*/
@@ -235,7 +269,7 @@ public class StandbyContainerManager {
// record the resource request, before issuing it to avoid race with
allocation-thread
SamzaResourceRequest resourceRequestForActive =
- containerAllocator.getResourceRequest(activeContainerID,
standbyHost);
+ containerAllocator.getResourceRequest(activeContainerID,
standbyHost);
failoverMetadata.recordResourceRequest(resourceRequestForActive);
containerAllocator.issueResourceRequest(resourceRequestForActive);
samzaApplicationState.failoversToStandby.incrementAndGet();
@@ -281,7 +315,7 @@ public class StandbyContainerManager {
Optional<FailoverMetadata> failoverMetadata =
getFailoverMetadata(activeContainerResourceID);
// Iterate over the list of running standby containers, to find a standby
resource that we have not already
- // used for a failover for this active resoruce
+ // used for a failover for this active resource
for (String standbyContainerID :
this.standbyContainerConstraints.get(activeContainerID)) {
if
(samzaApplicationState.runningProcessors.containsKey(standbyContainerID)) {
@@ -307,7 +341,7 @@ public class StandbyContainerManager {
.map(ProcessorLocality::host)
.orElse(null);
- if (StringUtils.isNotBlank(standbyHost)) {
+ if (StringUtils.isBlank(standbyHost)) {
log.info("No last known standbyHost for container {}",
standbyContainerID);
} else if (failoverMetadata.isPresent() &&
failoverMetadata.get().isStandbyHostUsed(standbyHost)) {
@@ -361,8 +395,44 @@ public class StandbyContainerManager {
}
/**
- * Check if matching this SamzaResourceRequest to the given resource, meets
all standby-container container constraints.
+ * This method checks from the config if standby allocation is fault domain
aware or not, and requests resources accordingly.
+ *
+ * @param containerAllocator ContainerAllocator object that requests for
resources from the resource manager
+ * @param containerID Samza container ID that will be run when a resource is
allocated for this request
+ * @param preferredHost name of the host that you prefer to run the
processor on
+ * @param preferredHostRetryDelay the {@link Duration} to add to the request
timestamp
+ * @param hostToAvoid The hostname to avoid requesting this resource on if
fault domain aware standby allocation is enabled
+ */
+ void requestResource(ContainerAllocator containerAllocator, String
containerID, String preferredHost, Duration preferredHostRetryDelay, String
hostToAvoid) {
+ if (StandbyTaskUtil.isStandbyContainer(containerID) &&
isFaultDomainAwareStandbyEnabled) {
+ containerAllocator.requestResourceWithDelay(containerID, preferredHost,
preferredHostRetryDelay, getAllowedFaultDomainsGivenHostToAvoid(hostToAvoid));
+ } else {
+ containerAllocator.requestResourceWithDelay(containerID, preferredHost,
preferredHostRetryDelay, new HashSet<>());
+ }
+ }
+
+ /**
+ * This method returns the active container host given a standby or active
container ID.
*
+ * @param containerID Standby or active container container ID
+ * @return The active container host
+ */
+ Optional<String> getActiveContainerHost(String containerID) {
+ String activeContainerId = containerID;
+ if (StandbyTaskUtil.isStandbyContainer(containerID)) {
+ activeContainerId = StandbyTaskUtil.getActiveContainerId(containerID);
+ }
+ SamzaResource resource =
samzaApplicationState.pendingProcessors.get(activeContainerId);
+ if (resource == null) {
+ resource =
samzaApplicationState.runningProcessors.get(activeContainerId);
+ }
+ return Optional.ofNullable(resource)
+ .map(SamzaResource::getHost);
+ }
+
+ /**
+ * Check if matching this SamzaResourceRequest to the given resource, meets
all standby-container container constraints.
+ * This includes the check that a standby and its active should not be on
the same fault domain or the same host.
* @param containerIdToStart logical id of the container to start
* @param host potential host to start the container on
* @return
@@ -375,17 +445,33 @@ public class StandbyContainerManager {
SamzaResource resource =
samzaApplicationState.pendingProcessors.get(containerID);
// return false if a conflicting container is pending for launch on the
host
- if (resource != null && resource.getHost().equals(host)) {
+ if (resource != null && isFaultDomainAwareStandbyEnabled
+ && faultDomainManager.hasSameFaultDomains(host,
resource.getHost())) {
+ log.info("Container {} cannot be started on host {} because container
{} is already scheduled on this fault domain",
+ containerIdToStart, host, containerID);
+ if (StandbyTaskUtil.isStandbyContainer(containerIdToStart)) {
+
samzaApplicationState.failedFaultDomainAwareContainerAllocations.incrementAndGet();
+ }
+ return false;
+ } else if (resource != null && resource.getHost().equals(host)) {
log.info("Container {} cannot be started on host {} because container
{} is already scheduled on this host",
- containerIdToStart, host, containerID);
+ containerIdToStart, host, containerID);
return false;
}
// return false if a conflicting container is running on the host
resource = samzaApplicationState.runningProcessors.get(containerID);
- if (resource != null && resource.getHost().equals(host)) {
+ if (resource != null && isFaultDomainAwareStandbyEnabled
+ && faultDomainManager.hasSameFaultDomains(host,
resource.getHost())) {
+ log.info("Container {} cannot be started on host {} because container
{} is already running on this fault domain",
+ containerIdToStart, host, containerID);
+ if (StandbyTaskUtil.isStandbyContainer(containerIdToStart)) {
+
samzaApplicationState.failedFaultDomainAwareContainerAllocations.incrementAndGet();
+ }
+ return false;
+ } else if (resource != null && resource.getHost().equals(host)) {
log.info("Container {} cannot be started on host {} because container
{} is already running on this host",
- containerIdToStart, host, containerID);
+ containerIdToStart, host, containerID);
return false;
}
}
@@ -409,16 +495,21 @@ public class StandbyContainerManager {
log.info("Running container {} on {} meets standby constraints,
preferredHost = {}", containerID,
samzaResource.getHost(), preferredHost);
containerAllocator.runStreamProcessor(request, preferredHost);
+ if (isFaultDomainAwareStandbyEnabled &&
StandbyTaskUtil.isStandbyContainer(containerID)) {
+
samzaApplicationState.faultDomainAwareContainersStarted.incrementAndGet();
+ }
} else if (StandbyTaskUtil.isStandbyContainer(containerID)) {
// This resource cannot be used to launch this standby container, so we
make a new anyhost request
log.info(
"Running standby container {} on host {} does not meet standby
constraints, cancelling resource request, releasing resource, and making a new
ANY_HOST request",
containerID, samzaResource.getHost());
releaseUnstartableContainer(request, samzaResource, preferredHost,
resourceRequestState);
- containerAllocator.requestResource(containerID,
ResourceRequestState.ANY_HOST);
+ String activeContainerHost = getActiveContainerHost(containerID)
+ .orElse(null);
+ requestResource(containerAllocator, containerID,
ResourceRequestState.ANY_HOST, Duration.ZERO, activeContainerHost);
samzaApplicationState.failedStandbyAllocations.incrementAndGet();
} else {
- // This resource cannot be used to launch this active container
container, so we initiate a failover
+ // This resource cannot be used to launch this active container, so we
initiate a failover
log.warn(
"Running active container {} on host {} does not meet standby
constraints, cancelling resource request, releasing resource",
containerID, samzaResource.getHost());
@@ -469,7 +560,6 @@ public class StandbyContainerManager {
resourceRequestState.cancelResourceRequest(request);
}
-
// Handle an expired resource request that was made for placing a standby
container
private void handleExpiredRequestForStandbyContainer(String containerID,
SamzaResourceRequest request,
Optional<SamzaResource> alternativeResource, ContainerAllocator
containerAllocator,
@@ -486,7 +576,9 @@ public class StandbyContainerManager {
// If there is no alternative-resource for the standby container we make
a new anyhost request
log.info("Handling expired request, requesting anyHost resource for
standby container {}", containerID);
resourceRequestState.cancelResourceRequest(request);
- containerAllocator.requestResource(containerID,
ResourceRequestState.ANY_HOST);
+ String activeContainerHost = getActiveContainerHost(containerID)
+ .orElse(null);
+ requestResource(containerAllocator, containerID,
ResourceRequestState.ANY_HOST, Duration.ZERO, activeContainerHost);
}
}
diff --git
a/samza-core/src/test/java/org/apache/samza/clustermanager/TestClusterBasedJobCoordinator.java
b/samza-core/src/test/java/org/apache/samza/clustermanager/TestClusterBasedJobCoordinator.java
index 50a1ee1..e0b0739 100644
---
a/samza-core/src/test/java/org/apache/samza/clustermanager/TestClusterBasedJobCoordinator.java
+++
b/samza-core/src/test/java/org/apache/samza/clustermanager/TestClusterBasedJobCoordinator.java
@@ -88,6 +88,7 @@ public class TestClusterBasedJobCoordinator {
configMap.put("task.inputs", "kafka.topic1");
configMap.put("systems.kafka.samza.factory",
"org.apache.samza.system.MockSystemFactory");
configMap.put("samza.cluster-manager.factory",
"org.apache.samza.clustermanager.MockClusterResourceManagerFactory");
+ configMap.put("cluster-manager.fault-domain-manager.factory",
"org.apache.samza.clustermanager.MockFaultDomainManagerFactory");
configMap.put("job.coordinator.monitor-partition-change.frequency.ms",
"1");
MockSystemFactory.MSG_QUEUES.put(new SystemStreamPartition("kafka",
"topic1", new Partition(0)), new ArrayList<>());
diff --git
a/samza-core/src/test/java/org/apache/samza/clustermanager/TestContainerAllocatorWithHostAffinity.java
b/samza-core/src/test/java/org/apache/samza/clustermanager/TestContainerAllocatorWithHostAffinity.java
index 2b4a4b0..2c9ba81 100644
---
a/samza-core/src/test/java/org/apache/samza/clustermanager/TestContainerAllocatorWithHostAffinity.java
+++
b/samza-core/src/test/java/org/apache/samza/clustermanager/TestContainerAllocatorWithHostAffinity.java
@@ -64,6 +64,7 @@ public class TestContainerAllocatorWithHostAffinity {
private final SamzaApplicationState state = new
SamzaApplicationState(jobModelManager);
private final MockClusterResourceManager clusterResourceManager = new
MockClusterResourceManager(callback, state);
+ private final FaultDomainManager faultDomainManager =
mock(FaultDomainManager.class);
private ContainerPlacementMetadataStore containerPlacementMetadataStore;
private ContainerManager containerManager;
@@ -89,7 +90,7 @@ public class TestContainerAllocatorWithHostAffinity {
coordinatorStreamStore.init();
containerPlacementMetadataStore = new
ContainerPlacementMetadataStore(coordinatorStreamStore);
containerPlacementMetadataStore.start();
- containerManager = new ContainerManager(containerPlacementMetadataStore,
state, clusterResourceManager, true, false, mockLocalityManager);
+ containerManager = new ContainerManager(containerPlacementMetadataStore,
state, clusterResourceManager, true, false, mockLocalityManager,
faultDomainManager, config);
containerAllocator =
new ContainerAllocator(clusterResourceManager, config, state, true,
containerManager);
requestState = new MockContainerRequestState(clusterResourceManager, true);
@@ -369,7 +370,7 @@ public class TestContainerAllocatorWithHostAffinity {
ClusterResourceManager.Callback mockCPM =
mock(MockClusterResourceManagerCallback.class);
ClusterResourceManager mockClusterResourceManager = new
MockClusterResourceManager(mockCPM, state);
ContainerManager containerManager =
- new ContainerManager(containerPlacementMetadataStore, state,
mockClusterResourceManager, true, false, mock(LocalityManager.class));
+ new ContainerManager(containerPlacementMetadataStore, state,
mockClusterResourceManager, true, false, mock(LocalityManager.class),
faultDomainManager, config);
// Mock the callback from ClusterManager to add resources to the allocator
doAnswer((InvocationOnMock invocation) -> {
SamzaResource resource = (SamzaResource) invocation.getArgumentAt(0,
List.class).get(0);
@@ -416,7 +417,7 @@ public class TestContainerAllocatorWithHostAffinity {
public void testExpiredRequestAllocationOnAnyHost() throws Exception {
MockClusterResourceManager spyManager = spy(new
MockClusterResourceManager(callback, state));
ContainerManager spyContainerManager =
- spy(new ContainerManager(containerPlacementMetadataStore, state,
spyManager, true, false, mock(LocalityManager.class)));
+ spy(new ContainerManager(containerPlacementMetadataStore, state,
spyManager, true, false, mock(LocalityManager.class), faultDomainManager,
config));
spyAllocator = Mockito.spy(
new ContainerAllocator(spyManager, config, state, true,
spyContainerManager));
// Request Preferred Resources
@@ -460,7 +461,7 @@ public class TestContainerAllocatorWithHostAffinity {
// Add Extra Resources
MockClusterResourceManager spyClusterResourceManager = spy(new
MockClusterResourceManager(callback, state));
ContainerManager spyContainerManager =
- spy(new ContainerManager(containerPlacementMetadataStore, state,
spyClusterResourceManager, true, false, mock(LocalityManager.class)));
+ spy(new ContainerManager(containerPlacementMetadataStore, state,
spyClusterResourceManager, true, false, mock(LocalityManager.class),
faultDomainManager, config));
spyAllocator = Mockito.spy(
new ContainerAllocator(spyClusterResourceManager, config, state, true,
spyContainerManager));
@@ -513,7 +514,7 @@ public class TestContainerAllocatorWithHostAffinity {
ClusterResourceManager.Callback mockCPM =
mock(MockClusterResourceManagerCallback.class);
MockClusterResourceManager mockClusterResourceManager = new
MockClusterResourceManager(mockCPM, state);
ContainerManager spyContainerManager =
- spy(new ContainerManager(containerPlacementMetadataStore, state,
mockClusterResourceManager, true, false, mock(LocalityManager.class)));
+ spy(new ContainerManager(containerPlacementMetadataStore, state,
mockClusterResourceManager, true, false, mock(LocalityManager.class),
faultDomainManager, config));
SamzaResource expiredAllocatedResource = new SamzaResource(1, 1000,
"host-0", "id0",
System.currentTimeMillis() - Duration.ofMinutes(10).toMillis());
diff --git
a/samza-core/src/test/java/org/apache/samza/clustermanager/TestContainerAllocatorWithoutHostAffinity.java
b/samza-core/src/test/java/org/apache/samza/clustermanager/TestContainerAllocatorWithoutHostAffinity.java
index ac5d6f3..1f063d7 100644
---
a/samza-core/src/test/java/org/apache/samza/clustermanager/TestContainerAllocatorWithoutHostAffinity.java
+++
b/samza-core/src/test/java/org/apache/samza/clustermanager/TestContainerAllocatorWithoutHostAffinity.java
@@ -62,6 +62,7 @@ public class TestContainerAllocatorWithoutHostAffinity {
private final SamzaApplicationState state = new
SamzaApplicationState(jobModelManager);
private final MockClusterResourceManager manager = new
MockClusterResourceManager(callback, state);
+ private final FaultDomainManager faultDomainManager =
mock(FaultDomainManager.class);
private CoordinatorStreamStore coordinatorStreamStore;
private ContainerPlacementMetadataStore containerPlacementMetadataStore;
@@ -83,7 +84,7 @@ public class TestContainerAllocatorWithoutHostAffinity {
containerPlacementMetadataStore = new
ContainerPlacementMetadataStore(coordinatorStreamStore);
containerPlacementMetadataStore.start();
containerAllocator = new ContainerAllocator(manager, config, state, false,
- new ContainerManager(containerPlacementMetadataStore, state, manager,
false, false, mockLocalityManager));
+ new ContainerManager(containerPlacementMetadataStore, state, manager,
false, false, mockLocalityManager, faultDomainManager, config));
requestState = new MockContainerRequestState(manager, false);
Field requestStateField =
containerAllocator.getClass().getDeclaredField("resourceRequestState");
requestStateField.setAccessible(true);
@@ -179,9 +180,11 @@ public class TestContainerAllocatorWithoutHostAffinity {
});
LocalityManager mockLocalityManager = mock(LocalityManager.class);
when(mockLocalityManager.readLocality()).thenReturn(new LocalityModel(new
HashMap<>()));
+ ContainerManager containerManager = new
ContainerManager(containerPlacementMetadataStore, state, manager, false,
+ false, mockLocalityManager, faultDomainManager, config);
containerAllocator =
MockContainerAllocatorWithoutHostAffinity.createContainerAllocatorWithConfigOverride(manager,
config, state,
- new ContainerManager(containerPlacementMetadataStore, state,
manager, false, false, mockLocalityManager),
+ containerManager,
override);
MockContainerAllocatorWithoutHostAffinity mockAllocator =
(MockContainerAllocatorWithoutHostAffinity) containerAllocator;
@@ -331,7 +334,7 @@ public class TestContainerAllocatorWithoutHostAffinity {
ClusterResourceManager.Callback mockCPM =
mock(ClusterResourceManager.Callback.class);
ClusterResourceManager mockManager = new
MockClusterResourceManager(mockCPM, state);
ContainerManager spyContainerManager =
- spy(new ContainerManager(containerPlacementMetadataStore, state,
mockManager, false, false, mock(LocalityManager.class)));
+ spy(new ContainerManager(containerPlacementMetadataStore, state,
mockManager, false, false, mock(LocalityManager.class), faultDomainManager,
config));
spyAllocator = Mockito.spy(
new ContainerAllocator(mockManager, config, state, false,
spyContainerManager));
// Mock the callback from ClusterManager to add resources to the allocator
diff --git
a/samza-core/src/test/java/org/apache/samza/clustermanager/TestContainerPlacementActions.java
b/samza-core/src/test/java/org/apache/samza/clustermanager/TestContainerPlacementActions.java
index c781f4d..e5ead9e 100644
---
a/samza-core/src/test/java/org/apache/samza/clustermanager/TestContainerPlacementActions.java
+++
b/samza-core/src/test/java/org/apache/samza/clustermanager/TestContainerPlacementActions.java
@@ -141,13 +141,14 @@ public class TestContainerPlacementActions {
state = new
SamzaApplicationState(JobModelManagerTestUtil.getJobModelManager(getConfig(),
2, server));
callback = mock(ClusterResourceManager.Callback.class);
MockClusterResourceManager clusterResourceManager = new
MockClusterResourceManager(callback, state);
+ FaultDomainManager faultDomainManager = mock(FaultDomainManager.class);
ClusterManagerConfig clusterManagerConfig = new
ClusterManagerConfig(config);
localityManager = mock(LocalityManager.class);
when(localityManager.readLocality())
.thenReturn(new LocalityModel(ImmutableMap.of(
"0", new ProcessorLocality("0", "host-1"),
"1", new ProcessorLocality("1", "host-2"))));
- containerManager = spy(new
ContainerManager(containerPlacementMetadataStore, state,
clusterResourceManager, true, false, localityManager));
+ containerManager = spy(new
ContainerManager(containerPlacementMetadataStore, state,
clusterResourceManager, true, false, localityManager, faultDomainManager,
config));
allocatorWithHostAffinity = new
MockContainerAllocatorWithHostAffinity(clusterResourceManager, config, state,
containerManager);
cpm = new ContainerProcessManager(clusterManagerConfig, state, new
MetricsRegistryMap(),
clusterResourceManager, Optional.of(allocatorWithHostAffinity),
containerManager, localityManager, false);
@@ -171,9 +172,10 @@ public class TestContainerPlacementActions {
state = new SamzaApplicationState(getJobModelManagerWithStandby());
callback = mock(ClusterResourceManager.Callback.class);
MockClusterResourceManager clusterResourceManager = new
MockClusterResourceManager(callback, state);
+ FaultDomainManager faultDomainManager = mock(FaultDomainManager.class);
ClusterManagerConfig clusterManagerConfig = new
ClusterManagerConfig(config);
// Enable standby
- containerManager = spy(new
ContainerManager(containerPlacementMetadataStore, state,
clusterResourceManager, true, true, mockLocalityManager));
+ containerManager = spy(new
ContainerManager(containerPlacementMetadataStore, state,
clusterResourceManager, true, true, mockLocalityManager, faultDomainManager,
config));
allocatorWithHostAffinity = new
MockContainerAllocatorWithHostAffinity(clusterResourceManager, config, state,
containerManager);
cpm = new ContainerProcessManager(clusterManagerConfig, state, new
MetricsRegistryMap(),
clusterResourceManager, Optional.of(allocatorWithHostAffinity),
containerManager, mockLocalityManager, false);
@@ -552,8 +554,9 @@ public class TestContainerPlacementActions {
state = new
SamzaApplicationState(JobModelManagerTestUtil.getJobModelManager(getConfig(),
2, this.server));
callback = mock(ClusterResourceManager.Callback.class);
MockClusterResourceManager clusterResourceManager = new
MockClusterResourceManager(callback, state);
+ FaultDomainManager faultDomainManager = mock(FaultDomainManager.class);
ClusterManagerConfig clusterManagerConfig = new
ClusterManagerConfig(config);
- containerManager = spy(new
ContainerManager(containerPlacementMetadataStore, state,
clusterResourceManager, true, false, localityManager));
+ containerManager = spy(new
ContainerManager(containerPlacementMetadataStore, state,
clusterResourceManager, true, false, localityManager, faultDomainManager,
config));
allocatorWithHostAffinity = new
MockContainerAllocatorWithHostAffinity(clusterResourceManager, config, state,
containerManager);
cpm = new ContainerProcessManager(clusterManagerConfig, state, new
MetricsRegistryMap(),
clusterResourceManager, Optional.of(allocatorWithHostAffinity),
containerManager, localityManager, false);
@@ -666,8 +669,9 @@ public class TestContainerPlacementActions {
new
SamzaApplicationState(JobModelManagerTestUtil.getJobModelManager(getConfig(),
2, this.server));
ClusterResourceManager.Callback callback =
mock(ClusterResourceManager.Callback.class);
MockClusterResourceManager clusterResourceManager = new
MockClusterResourceManager(callback, state);
+ FaultDomainManager faultDomainManager = mock(FaultDomainManager.class);
ContainerManager containerManager =
- new ContainerManager(containerPlacementMetadataStore, state,
clusterResourceManager, false, false, localityManager);
+ new ContainerManager(containerPlacementMetadataStore, state,
clusterResourceManager, false, false, localityManager, faultDomainManager,
config);
MockContainerAllocatorWithoutHostAffinity allocatorWithoutHostAffinity =
new MockContainerAllocatorWithoutHostAffinity(clusterResourceManager,
new MapConfig(conf), state,
containerManager);
@@ -801,8 +805,9 @@ public class TestContainerPlacementActions {
new
SamzaApplicationState(JobModelManagerTestUtil.getJobModelManager(getConfig(),
2, this.server));
ClusterResourceManager.Callback callback =
mock(ClusterResourceManager.Callback.class);
MockClusterResourceManager clusterResourceManager = new
MockClusterResourceManager(callback, state);
+ FaultDomainManager faultDomainManager = mock(FaultDomainManager.class);
ContainerManager containerManager =
- spy(new ContainerManager(containerPlacementMetadataStore, state,
clusterResourceManager, true, false, localityManager));
+ spy(new ContainerManager(containerPlacementMetadataStore, state,
clusterResourceManager, true, false, localityManager, faultDomainManager,
config));
MockContainerAllocatorWithHostAffinity allocatorWithHostAffinity =
new MockContainerAllocatorWithHostAffinity(clusterResourceManager,
config, state, containerManager);
ContainerProcessManager cpm = new ContainerProcessManager(
diff --git
a/samza-core/src/test/java/org/apache/samza/clustermanager/TestContainerProcessManager.java
b/samza-core/src/test/java/org/apache/samza/clustermanager/TestContainerProcessManager.java
index bcbe53f..ad45c5e 100644
---
a/samza-core/src/test/java/org/apache/samza/clustermanager/TestContainerProcessManager.java
+++
b/samza-core/src/test/java/org/apache/samza/clustermanager/TestContainerProcessManager.java
@@ -141,11 +141,12 @@ public class TestContainerProcessManager {
SamzaApplicationState state = new
SamzaApplicationState(getJobModelManager(1));
MockClusterResourceManagerCallback callback = new
MockClusterResourceManagerCallback();
MockClusterResourceManager clusterResourceManager = new
MockClusterResourceManager(callback, state);
+ FaultDomainManager faultDomainManager = mock(FaultDomainManager.class);
LocalityManager mockLocalityManager = mock(LocalityManager.class);
when(mockLocalityManager.readLocality())
.thenReturn(new LocalityModel(ImmutableMap.of("0", new
ProcessorLocality("0", "host1"))));
ContainerManager containerManager =
- buildContainerManager(containerPlacementMetadataStore, state,
clusterResourceManager, true, false, mockLocalityManager);
+ buildContainerManager(containerPlacementMetadataStore, state,
clusterResourceManager, true, false, mockLocalityManager, faultDomainManager);
ContainerProcessManager cpm =
buildContainerProcessManager(new ClusterManagerConfig(new
MapConfig(conf)), state, clusterResourceManager, Optional.empty());
@@ -592,6 +593,7 @@ public class TestContainerProcessManager {
SamzaApplicationState state = new
SamzaApplicationState(getJobModelManager(1));
MockClusterResourceManagerCallback callback = new
MockClusterResourceManagerCallback();
MockClusterResourceManager clusterResourceManager = new
MockClusterResourceManager(callback, state);
+ FaultDomainManager faultDomainManager = mock(FaultDomainManager.class);
LocalityManager mockLocalityManager = mock(LocalityManager.class);
if (withHostAffinity) {
@@ -604,7 +606,7 @@ public class TestContainerProcessManager {
ContainerManager containerManager =
buildContainerManager(containerPlacementMetadataStore, state,
clusterResourceManager,
- clusterManagerConfig.getHostAffinityEnabled(), false,
mockLocalityManager);
+ clusterManagerConfig.getHostAffinityEnabled(), false,
mockLocalityManager, faultDomainManager);
MockContainerAllocatorWithoutHostAffinity allocator = new
MockContainerAllocatorWithoutHostAffinity(
clusterResourceManager,
@@ -614,7 +616,7 @@ public class TestContainerProcessManager {
ContainerProcessManager cpm =
buildContainerProcessManager(clusterManagerConfig, state,
clusterResourceManager, Optional.of(allocator),
- mockLocalityManager, false);
+ mockLocalityManager, false, faultDomainManager);
// start triggers a request
cpm.start();
@@ -755,11 +757,12 @@ public class TestContainerProcessManager {
configMap.putAll(getConfig());
MockClusterResourceManagerCallback callback = new
MockClusterResourceManagerCallback();
MockClusterResourceManager clusterResourceManager = new
MockClusterResourceManager(callback, state);
+ FaultDomainManager faultDomainManager = mock(FaultDomainManager.class);
LocalityManager mockLocalityManager = mock(LocalityManager.class);
when(mockLocalityManager.readLocality())
.thenReturn(new LocalityModel(ImmutableMap.of("0", new
ProcessorLocality("1", "host1"))));
ContainerManager containerManager =
buildContainerManager(containerPlacementMetadataStore, state,
clusterResourceManager,
-
Boolean.valueOf(config.get(ClusterManagerConfig.HOST_AFFINITY_ENABLED)), false,
mockLocalityManager);
+
Boolean.valueOf(config.get(ClusterManagerConfig.HOST_AFFINITY_ENABLED)), false,
mockLocalityManager, faultDomainManager);
MockContainerAllocatorWithoutHostAffinity allocator = new
MockContainerAllocatorWithoutHostAffinity(
clusterResourceManager,
@@ -792,11 +795,12 @@ public class TestContainerProcessManager {
SamzaApplicationState state = new
SamzaApplicationState(getJobModelManager(2));
MockClusterResourceManagerCallback callback = new
MockClusterResourceManagerCallback();
MockClusterResourceManager clusterResourceManager = new
MockClusterResourceManager(callback, state);
+ FaultDomainManager faultDomainManager = mock(FaultDomainManager.class);
LocalityManager mockLocalityManager = mock(LocalityManager.class);
when(mockLocalityManager.readLocality())
.thenReturn(new LocalityModel(ImmutableMap.of("0", new
ProcessorLocality("0", "host1"), "1", new ProcessorLocality("1", "host2"))));
ContainerManager containerManager =
buildContainerManager(containerPlacementMetadataStore, state,
clusterResourceManager,
-
Boolean.parseBoolean(config.get(ClusterManagerConfig.HOST_AFFINITY_ENABLED)),
false, mockLocalityManager);
+
Boolean.parseBoolean(config.get(ClusterManagerConfig.HOST_AFFINITY_ENABLED)),
false, mockLocalityManager, faultDomainManager);
MockContainerAllocatorWithHostAffinity allocator = new
MockContainerAllocatorWithHostAffinity(
clusterResourceManager,
@@ -806,7 +810,7 @@ public class TestContainerProcessManager {
ContainerProcessManager cpm =
spy(buildContainerProcessManager(new ClusterManagerConfig(cfg), state,
clusterResourceManager,
- Optional.of(allocator), mockLocalityManager, false));
+ Optional.of(allocator), mockLocalityManager, false,
faultDomainManager));
cpm.start();
assertFalse(cpm.shouldShutdown());
@@ -1031,16 +1035,16 @@ public class TestContainerProcessManager {
SamzaApplicationState samzaApplicationState, ClusterResourceManager
clusterResourceManager,
boolean hostAffinityEnabled, boolean standByEnabled) {
LocalityManager mockLocalityManager = mock(LocalityManager.class);
+ FaultDomainManager faultDomainManager = mock(FaultDomainManager.class);
when(mockLocalityManager.readLocality()).thenReturn(new LocalityModel(new
HashMap<>()));
return buildContainerManager(containerPlacementMetadataStore,
samzaApplicationState, clusterResourceManager,
- hostAffinityEnabled, standByEnabled, mockLocalityManager);
+ hostAffinityEnabled, standByEnabled, mockLocalityManager,
faultDomainManager);
}
private ContainerManager
buildContainerManager(ContainerPlacementMetadataStore
containerPlacementMetadataStore,
- SamzaApplicationState samzaApplicationState, ClusterResourceManager
clusterResourceManager,
- boolean hostAffinityEnabled, boolean standByEnabled, LocalityManager
localityManager) {
- return new ContainerManager(containerPlacementMetadataStore,
samzaApplicationState, clusterResourceManager,
- hostAffinityEnabled, standByEnabled, localityManager);
+ SamzaApplicationState samzaApplicationState, ClusterResourceManager
clusterResourceManager, boolean hostAffinityEnabled,
+ boolean standByEnabled, LocalityManager localityManager,
FaultDomainManager faultDomainManager) {
+ return new ContainerManager(containerPlacementMetadataStore,
samzaApplicationState, clusterResourceManager, hostAffinityEnabled,
standByEnabled, localityManager, faultDomainManager, config);
}
private ContainerProcessManager
buildContainerProcessManager(ClusterManagerConfig clusterManagerConfig,
SamzaApplicationState state,
ClusterResourceManager clusterResourceManager,
Optional<ContainerAllocator> allocator) {
@@ -1050,16 +1054,17 @@ public class TestContainerProcessManager {
private ContainerProcessManager
buildContainerProcessManager(ClusterManagerConfig clusterManagerConfig,
SamzaApplicationState state,
ClusterResourceManager clusterResourceManager,
Optional<ContainerAllocator> allocator, boolean restartContainer) {
LocalityManager mockLocalityManager = mock(LocalityManager.class);
+ FaultDomainManager faultDomainManager = mock(FaultDomainManager.class);
when(mockLocalityManager.readLocality()).thenReturn(new LocalityModel(new
HashMap<>()));
return buildContainerProcessManager(clusterManagerConfig, state,
clusterResourceManager, allocator,
- mockLocalityManager, restartContainer);
+ mockLocalityManager, restartContainer, faultDomainManager);
}
private ContainerProcessManager
buildContainerProcessManager(ClusterManagerConfig clusterManagerConfig,
SamzaApplicationState state,
ClusterResourceManager clusterResourceManager,
Optional<ContainerAllocator> allocator, LocalityManager localityManager,
- boolean restartContainers) {
+ boolean restartContainers, FaultDomainManager faultDomainManager) {
return new ContainerProcessManager(clusterManagerConfig, state, new
MetricsRegistryMap(), clusterResourceManager,
allocator, buildContainerManager(containerPlacementMetadataStore,
state, clusterResourceManager,
- clusterManagerConfig.getHostAffinityEnabled(), false,
localityManager), localityManager, restartContainers);
+ clusterManagerConfig.getHostAffinityEnabled(), false, localityManager,
faultDomainManager), localityManager, restartContainers);
}
}
diff --git
a/samza-yarn/src/main/java/org/apache/samza/job/yarn/YarnClusterResourceManager.java
b/samza-yarn/src/main/java/org/apache/samza/job/yarn/YarnClusterResourceManager.java
index fa784e0..ccdd00f 100644
---
a/samza-yarn/src/main/java/org/apache/samza/job/yarn/YarnClusterResourceManager.java
+++
b/samza-yarn/src/main/java/org/apache/samza/job/yarn/YarnClusterResourceManager.java
@@ -241,6 +241,7 @@ public class YarnClusterResourceManager extends
ClusterResourceManager implement
String processorId = resourceRequest.getProcessorId();
String requestId = resourceRequest.getRequestId();
String preferredHost = resourceRequest.getPreferredHost();
+ String[] racks =
resourceRequest.getFaultDomains().stream().map(FaultDomain::getId).toArray(String[]::new);
int memoryMb = resourceRequest.getMemoryMB();
int cpuCores = resourceRequest.getNumCores();
Resource capability = Resource.newInstance(memoryMb, cpuCores);
@@ -261,15 +262,15 @@ public class YarnClusterResourceManager extends
ClusterResourceManager implement
Priority priority = Priority.newInstance(ANY_HOST_PRIORITY);
boolean relaxLocality = true;
log.info("Requesting resources for Processor ID: {} on nodes: {} on
racks: {} with capability: {}, priority: {}, relaxLocality: {},
nodeLabelsExpression: {}",
- processorId, null, null, capability, priority, relaxLocality,
nodeLabelsExpression);
+ processorId, null, Arrays.toString(racks), capability, priority,
relaxLocality, nodeLabelsExpression);
issuedRequest = new AMRMClient.ContainerRequest(capability, null, null,
priority, relaxLocality, nodeLabelsExpression);
} else {
String[] nodes = {preferredHost};
Priority priority = Priority.newInstance(PREFERRED_HOST_PRIORITY);
boolean relaxLocality = false;
log.info("Requesting resources for Processor ID: {} on nodes: {} on
racks: {} with capability: {}, priority: {}, relaxLocality: {},
nodeLabelsExpression: {}",
- processorId, Arrays.toString(nodes), null, capability, priority,
relaxLocality, nodeLabelsExpression);
- issuedRequest = new AMRMClient.ContainerRequest(capability, nodes, null,
priority, relaxLocality, nodeLabelsExpression);
+ processorId, Arrays.toString(nodes), Arrays.toString(racks),
capability, priority, relaxLocality, nodeLabelsExpression);
+ issuedRequest = new AMRMClient.ContainerRequest(capability, nodes,
racks, priority, relaxLocality, nodeLabelsExpression);
}
// ensure that updating the state and making the request are done
atomically.
synchronized (lock) {
