SAMZA-2266: Introduce a backoff when there are repeated failures for host-affinity allocations (apache#1104)

* SAMZA-2266: Introduce a backoff when there are repeated failures for host-affinity allocations

Author: dnishimura

docs/learn/documentation/versioned/jobs/samza-configurations.md

@@ -295,6 +295,7 @@ Samza supports both standalone and clustered ([YARN](yarn-jobs.html)) [deploymen
 |--- |--- |--- |
 |cluster-manager.container.retry.count|8|If a container fails, it is automatically restarted by Samza. However, if a container keeps failing shortly after startup, that indicates a deeper problem, so we should kill the job rather than retrying indefinitely. This property determines the maximum number of times we are willing to restart a failed container in quick succession (the time period is configured with `cluster-manager.container.retry.window.ms`). Each container in the job is counted separately. If this property is set to 0, any failed container immediately causes the whole job to fail. If it is set to a negative number, there is no limit on the number of retries.|
 |cluster-manager.container.retry.window.ms|300000|This property determines how frequently a container is allowed to fail before we give up and fail the job. If the same container has failed more than `cluster-manager.container.retry.count` times, and the time between failures was less than this property `cluster-manager.container.retry.window.ms` (in milliseconds), then we fail the job. There is no limit to the number of times we will restart a container if the time between failures is greater than `cluster-manager.container.retry.window.ms`.|
+|cluster-manager.container.preferred-host.last.retry.delay.ms|360000|The delay added to the last retry for a failing container after all but one of cluster-manager.container.retry.count retries have been exhausted. The delay is only added when `job.host-affinity.enabled` is true and the retried request is for a preferred host. This addresses the issue where there may be a delay when a preferred host is marked invalid and the container continuously attempts to restart and fail on the invalid preferred host. This property is useful to prevent the `cluster-manager.container.retry.count` from being exceeded too quickly for such scenarios.|
 |cluster-manager.jobcoordinator.jmx.enabled|true|This is deprecated in favor of `job.jmx.enabled`|
 |cluster-manager.allocator.sleep.ms|3600|The container allocator thread is responsible for matching requests to allocated containers. The sleep interval for this thread is configured using this property.|
 |cluster-manager.container.request.timeout.ms|5000|The allocator thread periodically checks the state of the container requests and allocated containers to determine the assignment of a container to an allocated resource. This property determines the number of milliseconds before a container request is considered to have expired / timed-out. When a request expires, it gets allocated to any available container that was returned by the cluster manager.|

samza-core/src/main/java/org/apache/samza/clustermanager/AbstractContainerAllocator.java

@@ -18,6 +18,10 @@
  */
 package org.apache.samza.clustermanager;
 
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Map;
+import java.util.Optional;
 import org.apache.samza.SamzaException;
 import org.apache.samza.config.ClusterManagerConfig;
 import org.apache.samza.config.Config;
@@ -28,8 +32,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.Map;
-
 
 /**
  * {@link AbstractContainerAllocator} makes requests for physical resources to the resource manager and also runs
@@ -110,6 +112,10 @@ public void run() {
     while (isRunning) {
       try {
         assignResourceRequests();
+
+        // Move delayed requests that are ready to the active request queue
+        resourceRequestState.sendPendingDelayedResourceRequests();
+
         // Release extra resources and update the entire system's state
         resourceRequestState.releaseExtraResources();
 
@@ -151,7 +157,7 @@ protected void runStreamProcessor(SamzaResourceRequest request, String preferred
 
     // Run processor on resource
     log.info("Found Container ID: {} for Processor ID: {} on host: {} for request creation time: {}.",
-        resource.getContainerId(), processorId, preferredHost, request.getRequestTimestampMs());
+        resource.getContainerId(), processorId, preferredHost, request.getRequestTimestamp());
 
     // Update processor state as "pending" and then issue a request to launch it. It's important to perform the state-update
     // prior to issuing the request. Otherwise, there's a race where the response callback may arrive sooner and not see
@@ -175,35 +181,65 @@ protected void runStreamProcessor(SamzaResourceRequest request, String preferred
   public abstract void requestResources(Map<String, String> processorToHostMapping);
 
   /**
-   * Checks if this allocator has a pending resource request.
+   * Checks if this allocator has a pending resource request with a request timestamp equal to or earlier than the current
+   * timestamp.
    * @return {@code true} if there is a pending request, {@code false} otherwise.
    */
-  protected final boolean hasPendingRequest() {
-    return peekPendingRequest() != null;
+  protected final boolean hasReadyPendingRequest() {
+    return peekReadyPendingRequest().isPresent();
   }
 
   /**
-   * Retrieves, but does not remove, the next pending request in the queue.
+   * Retrieves, but does not remove, the next pending request in the queue with the {@link SamzaResourceRequest#getRequestTimestamp()}
+   * that is greater than the current timestamp.
    *
    * @return  the pending request or {@code null} if there is no pending request.
    */
-  protected final SamzaResourceRequest peekPendingRequest() {
-    return resourceRequestState.peekPendingRequest();
+  protected final Optional<SamzaResourceRequest> peekReadyPendingRequest() {
+    SamzaResourceRequest pendingRequest = resourceRequestState.peekPendingRequest();
+    return Optional.ofNullable(pendingRequest);
   }
 
   /**
    * 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
    */
   public final void requestResource(String processorId, String preferredHost) {
-    SamzaResourceRequest request = getResourceRequest(processorId, preferredHost);
+    requestResourceWithDelay(processorId, preferredHost, Duration.ZERO);
+  }
+
+  /**
+   * 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
+   */
+  public final void requestResourceWithDelay(String processorId, String preferredHost, Duration delay) {
+    SamzaResourceRequest request = getResourceRequestWithDelay(processorId, preferredHost, delay);
     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
+   * @return the created request
+   */
   public final SamzaResourceRequest getResourceRequest(String processorId, String preferredHost) {
-    return new SamzaResourceRequest(this.containerNumCpuCores, this.containerMemoryMb, preferredHost, processorId);
+    return getResourceRequestWithDelay(processorId, preferredHost, Duration.ZERO);
+  }
+
+  /**
+   * 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
+   * @return the created request
+   */
+  public final SamzaResourceRequest getResourceRequestWithDelay(String processorId, String preferredHost, Duration delay) {
+    return new SamzaResourceRequest(this.containerNumCpuCores, this.containerMemoryMb, preferredHost, processorId, Instant.now().plus(delay));
   }
 
   public final void issueResourceRequest(SamzaResourceRequest request) {

samza-core/src/main/java/org/apache/samza/clustermanager/ContainerAllocator.java

@@ -51,9 +51,8 @@ public ContainerAllocator(ClusterResourceManager manager,
    * */
   @Override
   public void assignResourceRequests() {
-    while (hasPendingRequest() && hasAllocatedResource(ResourceRequestState.ANY_HOST)) {
-      SamzaResourceRequest request = peekPendingRequest();
-      runStreamProcessor(request, ResourceRequestState.ANY_HOST);
+    while (hasReadyPendingRequest() && hasAllocatedResource(ResourceRequestState.ANY_HOST)) {
+      peekReadyPendingRequest().ifPresent(request -> runStreamProcessor(request, ResourceRequestState.ANY_HOST));
     }
   }