YARN-5516. Add REST API for supporting recurring reservations. (Sean Po via Subru).

(cherry picked from commit 25932da6d1)

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ReservationDefinition.java

@@ -155,7 +155,11 @@ public abstract class ReservationDefinition {
    * are explicitly cancelled and have higher priority than non-periodic jobs
    * (during initial placement and replanning). Periodic job allocations are
    * consistent across runs (flexibility in allocation is leveraged only during
-   * initial placement, allocations remain consistent thereafter).
+   * initial placement, allocations remain consistent thereafter). Note that
+   * as a long, the recurrence expression must be greater than the duration of
+   * the reservation (deadline - arrival). Also note that the configured max
+   * period must be divisible by the recurrence expression if expressed as a
+   * long.
    *
    * @return recurrence of this reservation
    */
@@ -173,7 +177,11 @@ public abstract class ReservationDefinition {
    * are explicitly cancelled and have higher priority than non-periodic jobs
    * (during initial placement and replanning). Periodic job allocations are
    * consistent across runs (flexibility in allocation is leveraged only during
-   * initial placement, allocations remain consistent thereafter).
+   * initial placement, allocations remain consistent thereafter). Note that
+   * as a long, the recurrence expression must be greater than the duration of
+   * the reservation (deadline - arrival). Also note that the configured max
+   * period must be divisible by the recurrence expression if expressed as a
+   * long.
    *
    * @param recurrenceExpression recurrence interval of this reservation
    */

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/records/impl/pb/ReservationDefinitionPBImpl.java

@@ -226,6 +226,11 @@ public class ReservationDefinitionPBImpl extends ReservationDefinition {
 
   @Override
   public void setRecurrenceExpression(String recurrenceExpression) {
+    maybeInitBuilder();
+    if (recurrenceExpression == null) {
+      builder.clearRecurrenceExpression();
+      return;
+    }
     builder.setRecurrenceExpression(recurrenceExpression);
   }
 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/InMemoryPlan.java

@@ -93,6 +93,7 @@ public class InMemoryPlan implements Plan {
   private final Planner replanner;
   private final boolean getMoveOnExpiry;
   private final Clock clock;
+  private final long maxPeriodicity;
 
   private Resource totalCapacity;
 
@@ -111,9 +112,9 @@ public class InMemoryPlan implements Plan {
       ReservationAgent agent, Resource totalCapacity, long step,
       ResourceCalculator resCalc, Resource minAlloc, Resource maxAlloc,
       String queueName, Planner replanner, boolean getMoveOnExpiry,
-      long maxPeriodicty, RMContext rmContext) {
+      long maxPeriodicity, RMContext rmContext) {
     this(queueMetrics, policy, agent, totalCapacity, step, resCalc, minAlloc,
-        maxAlloc, queueName, replanner, getMoveOnExpiry, maxPeriodicty,
+        maxAlloc, queueName, replanner, getMoveOnExpiry, maxPeriodicity,
         rmContext, new UTCClock());
   }
 
@@ -132,8 +133,9 @@ public class InMemoryPlan implements Plan {
     this.minAlloc = minAlloc;
     this.maxAlloc = maxAlloc;
     this.rleSparseVector = new RLESparseResourceAllocation(resCalc);
+    this.maxPeriodicity = maxPeriodicty;
     this.periodicRle =
-        new PeriodicRLESparseResourceAllocation(resCalc, maxPeriodicty);
+        new PeriodicRLESparseResourceAllocation(resCalc, this.maxPeriodicity);
     this.queueName = queueName;
     this.replanner = replanner;
     this.getMoveOnExpiry = getMoveOnExpiry;
@@ -627,10 +629,35 @@ public class InMemoryPlan implements Plan {
             // handle periodic reservations
             long period = reservation.getPeriodicity();
             if (period > 0) {
-              long t = endTime % period;
+              // The shift is used to remove the wrap around for the
-              // check for both contained and wrap-around reservations
+              // reservation interval. The wrap around will still
-              if ((t - startTime) * (t - endTime)
+              // exist for the search interval.
-                  * (startTime - endTime) >= 0) {
+              long shift = reservation.getStartTime() % period;
+              // This is the duration of the reservation since
+              // duration < period.
+              long periodicReservationEnd =
+                  (reservation.getEndTime() -shift) % period;
+              long periodicSearchStart = (startTime - shift) % period;
+              long periodicSearchEnd = (endTime - shift) % period;
+              long searchDuration = endTime - startTime;
+
+              // 1. If the searchDuration is greater than the period, then
+              // the reservation is within the interval. This will allow
+              // us to ignore cases where search end > search start >
+              // reservation end.
+              // 2/3. If the search end is less than the reservation end, or if
+              // the search start is less than the reservation end, then the
+              // reservation will be in the reservation since
+              // periodic reservation start is always zero. Note that neither
+              // of those values will ever be negative.
+              // 4. If the search end is less than the search start, then
+              // there is a wrap around, and both values are implicitly
+              // greater than the reservation end because of condition 2/3,
+              // so the reservation is within the search interval.
+              if (searchDuration > period
+                  || periodicSearchEnd < periodicReservationEnd
+                  || periodicSearchStart < periodicReservationEnd
+                  || periodicSearchStart > periodicSearchEnd) {
                 flattenedReservations.add(reservation);
               }
             } else {
@@ -719,7 +746,7 @@ public class InMemoryPlan implements Plan {
 
         if (periodicRle.getTimePeriod() % period != 0) {
           throw new PlanningException("The reservation periodicity (" + period
-              + ") must be" + "an exact divider of the system maxPeriod ("
+              + ") must be" + " an exact divider of the system maxPeriod ("
               + periodicRle.getTimePeriod() + ")");
         }
 
@@ -811,6 +838,11 @@ public class InMemoryPlan implements Plan {
     return Resources.clone(maxAlloc);
   }
 
+  @Override
+  public long getMaximumPeriodicity() {
+    return this.maxPeriodicity;
+  }
+
   public String toCumulativeString() {
     readLock.lock();
     try {

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/PlanContext.java

@@ -90,6 +90,17 @@ public interface PlanContext {
    */
   public Resource getMaximumAllocation();
 
+  /**
+   * Returns the maximum periodicity allowed in a recurrence expression
+   * for reservations of a particular plan. This value must be divisible by
+   * the recurrence expression of a newly submitted reservation. Otherwise, the
+   * reservation submission will fail.
+   *
+   * @return the maximum periodicity allowed in a recurrence expression for
+   * reservations of a particular plan.
+   */
+  long getMaximumPeriodicity();
+
   /**
    * Return the name of the queue in the {@link ResourceScheduler} corresponding
    * to this plan

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/ReservationInputValidator.java

@@ -164,6 +164,14 @@ public class ReservationInputValidator {
             + ". Please try again with a smaller duration.";
         throw RPCUtil.getRemoteException(message);
       }
+      // verify maximum period is divisible by recurrence expression.
+      if (recurrence > 0 && plan.getMaximumPeriodicity() % recurrence != 0) {
+        message = "The maximum periodicity: " + plan.getMaximumPeriodicity() +
+            " must be divisible by the recurrence expression provided: " +
+            recurrence + ". Please try again with a recurrence expression" +
+            " that satisfies this requirement.";
+        throw RPCUtil.getRemoteException(message);
+      }
     } catch (NumberFormatException e) {
       message = "Invalid period " + recurrenceExpression + ". Please try"
           + " again with a non-negative long value as period.";

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/RMWebServices.java

@@ -2018,9 +2018,10 @@ public class RMWebServices extends WebServices implements RMWebServiceProtocol {
       list.add(rr);
     }
     ReservationRequests reqs = ReservationRequests.newInstance(list, resInt);
-    ReservationDefinition rDef =
+    ReservationDefinition rDef = ReservationDefinition.newInstance(
-        ReservationDefinition.newInstance(resInfo.getArrival(),
+        resInfo.getArrival(), resInfo.getDeadline(), reqs,
-            resInfo.getDeadline(), reqs, resInfo.getReservationName());
+        resInfo.getReservationName(), resInfo.getRecurrenceExpression(),
+        Priority.newInstance(resInfo.getPriority()));
 
     ReservationId reservationId =
         ReservationId.parseReservationId(resContext.getReservationId());
@@ -2119,9 +2120,10 @@ public class RMWebServices extends WebServices implements RMWebServiceProtocol {
       list.add(rr);
     }
     ReservationRequests reqs = ReservationRequests.newInstance(list, resInt);
-    ReservationDefinition rDef =
+    ReservationDefinition rDef = ReservationDefinition.newInstance(
-        ReservationDefinition.newInstance(resInfo.getArrival(),
+        resInfo.getArrival(), resInfo.getDeadline(), reqs,
-            resInfo.getDeadline(), reqs, resInfo.getReservationName());
+        resInfo.getReservationName(), resInfo.getRecurrenceExpression(),
+        Priority.newInstance(resInfo.getPriority()));
     ReservationUpdateRequest request = ReservationUpdateRequest.newInstance(
         rDef, ReservationId.parseReservationId(resContext.getReservationId()));
 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/ReservationDefinitionInfo.java

@@ -47,6 +47,9 @@ public class ReservationDefinitionInfo {
   @XmlElement(name = "priority")
   private int priority;
 
+  @XmlElement(name = "recurrence-expression")
+  private String recurrenceExpression;
+
   public ReservationDefinitionInfo() {
 
   }
@@ -57,6 +60,7 @@ public class ReservationDefinitionInfo {
     reservationName = definition.getReservationName();
     reservationRequests = new ReservationRequestsInfo(definition
             .getReservationRequests());
+    recurrenceExpression = definition.getRecurrenceExpression();
   }
 
   public long getArrival() {
@@ -100,4 +104,12 @@ public class ReservationDefinitionInfo {
     this.priority = priority;
   }
 
+  public String getRecurrenceExpression() {
+    return recurrenceExpression;
+  }
+
+  public void setRecurrenceExpression(String recurrenceExpression) {
+    this.recurrenceExpression = recurrenceExpression;
+  }
+
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestInMemoryPlan.java

@@ -21,9 +21,11 @@ import static org.junit.Assert.assertEquals;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
+import java.sql.Timestamp;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 
@@ -699,6 +701,189 @@ public class TestInMemoryPlan {
         .compareTo((ReservationAllocation) rAllocations.toArray()[0]) == 0);
   }
 
+  @Test
+  public void testGetReservationSearchIntervalBeforeReservationStart() {
+    // Reservation duration is 10 minutes
+    long reservationStart = Timestamp.valueOf("2050-12-03 10:37:37").getTime();
+    long reservationEnd = Timestamp.valueOf("2050-12-03 10:47:37").getTime();
+
+    long searchStart = Timestamp.valueOf("2050-12-03 10:10:37").getTime();
+    long searchEnd = Timestamp.valueOf("2050-12-03 10:20:37").getTime();
+
+    // 10 minute period in milliseconds.
+    long period = 10 * 60 * 1000;
+
+    // Negative test because even though the reservation would be encompassed
+    // if it was interpolated, it should not be picked up. Also test only one
+    // cycle because if we test more cycles, some of them will pass.
+    testNegativeGetRecurringReservationsHelper(reservationStart,
+        reservationEnd, searchStart, searchEnd, 1, period, 10);
+  }
+
+  @Test
+  public void testGetReservationSearchIntervalGreaterThanPeriod() {
+    // Reservation duration is 10 minutes
+    long reservationStart = Timestamp.valueOf("2050-12-03 10:37:37").getTime();
+    long reservationEnd = Timestamp.valueOf("2050-12-03 10:47:37").getTime();
+
+    // 1 Hour search interval will for sure encompass the recurring
+    // reservation with 20 minute recurrence.
+    long searchStart = Timestamp.valueOf("2050-12-03 10:57:37").getTime();
+    long searchEnd = Timestamp.valueOf("2050-12-03 11:57:37").getTime();
+
+    // 20 minute period in milliseconds.
+    long period = 20 * 60 * 1000;
+
+    testPositiveGetRecurringReservationsHelper(reservationStart,
+        reservationEnd, searchStart, searchEnd, 100, period, 10);
+  }
+
+  @Test
+  public void testGetReservationReservationFitWithinSearchInterval() {
+    // Reservation duration is 10 minutes
+    long reservationStart = Timestamp.valueOf("2050-12-03 10:37:37").getTime();
+    long reservationEnd = Timestamp.valueOf("2050-12-03 10:47:37").getTime();
+
+    // Search interval fits the entire reservation but is smaller than the
+    // period.
+    long searchStart = Timestamp.valueOf("2050-12-03 10:36:37").getTime();
+    long searchEnd = Timestamp.valueOf("2050-12-03 10:48:37").getTime();
+
+    // 20 minute period in milliseconds.
+    long period = 20 * 60 * 1000;
+
+    testPositiveGetRecurringReservationsHelper(reservationStart,
+        reservationEnd, searchStart, searchEnd, 100, period, 10);
+  }
+
+  @Test
+  public void testGetReservationReservationStartTimeOverlap() {
+    // Reservation duration is 10 minutes
+    long reservationStart = Timestamp.valueOf("2050-12-03 10:37:37").getTime();
+    long reservationEnd = Timestamp.valueOf("2050-12-03 10:47:37").getTime();
+
+    // Search interval fits the starting portion of the reservation.
+    long searchStart = Timestamp.valueOf("2050-12-03 11:36:37").getTime();
+    long searchEnd = Timestamp.valueOf("2050-12-03 11:38:37").getTime();
+
+    // 60 minute period in milliseconds.
+    long period = 60 * 60 * 1000;
+
+    testPositiveGetRecurringReservationsHelper(reservationStart,
+        reservationEnd, searchStart, searchEnd, 100, period, 10);
+  }
+
+  @Test
+  public void testGetReservationReservationEndTimeOverlap() {
+    // Reservation duration is 10 minutes
+    long reservationStart = Timestamp.valueOf("2050-12-03 10:37:37").getTime();
+    long reservationEnd = Timestamp.valueOf("2050-12-03 10:47:37").getTime();
+
+    // Search interval fits the ending portion of the reservation.
+    long searchStart = Timestamp.valueOf("2050-12-03 11:46:37").getTime();
+    long searchEnd = Timestamp.valueOf("2050-12-03 11:48:37").getTime();
+
+    // 60 minute period in milliseconds.
+    long period = 60 * 60 * 1000;
+
+    testPositiveGetRecurringReservationsHelper(reservationStart,
+        reservationEnd, searchStart, searchEnd, 100, period, 10);
+  }
+
+  @Test
+  public void testGetReservationSearchIntervalFitsInReservation() {
+    // Reservation duration is 10 minutes
+    long reservationStart = Timestamp.valueOf("2050-12-03 10:37:37").getTime();
+    long reservationEnd = Timestamp.valueOf("2050-12-03 10:47:37").getTime();
+
+    // Search interval fits the within the reservation.
+    long searchStart = Timestamp.valueOf("2050-12-03 10:40:37").getTime();
+    long searchEnd = Timestamp.valueOf("2050-12-03 10:43:37").getTime();
+
+    // 60 minute period in milliseconds.
+    long period = 60 * 60 * 1000;
+
+    testPositiveGetRecurringReservationsHelper(reservationStart,
+        reservationEnd, searchStart, searchEnd, 100, period, 10);
+  }
+
+  @Test
+  public void testNegativeGetReservationSearchIntervalCloseToEndTime() {
+    // Reservation duration is 10 minutes
+    long reservationStart = Timestamp.valueOf("2050-12-03 10:37:37").getTime();
+    long reservationEnd = Timestamp.valueOf("2050-12-03 10:47:37").getTime();
+
+    // Reservation does not fit within search interval, but is close to the end
+    // time.
+    long searchStart = Timestamp.valueOf("2050-12-03 10:48:37").getTime();
+    long searchEnd = Timestamp.valueOf("2050-12-03 10:50:37").getTime();
+
+    // 60 minute period in milliseconds.
+    long period = 60 * 60 * 1000;
+
+    testNegativeGetRecurringReservationsHelper(reservationStart,
+        reservationEnd, searchStart, searchEnd, 100, period, 10);
+  }
+
+  @Test
+  public void testNegativeGetReservationSearchIntervalCloseToStartTime() {
+    // Reservation duration is 10 minutes
+    long reservationStart = Timestamp.valueOf("2050-12-03 10:37:37").getTime();
+    long reservationEnd = Timestamp.valueOf("2050-12-03 10:47:37").getTime();
+
+    // Search interval does not fit within the reservation but is close to
+    // the start time.
+    long searchStart = Timestamp.valueOf("2050-12-03 11:30:37").getTime();
+    long searchEnd = Timestamp.valueOf("2050-12-03 11:35:37").getTime();
+
+    // 60 minute period in milliseconds.
+    long period = 60 * 60 * 1000;
+
+    testNegativeGetRecurringReservationsHelper(reservationStart,
+        reservationEnd, searchStart, searchEnd, 100, period, 10);
+  }
+
+  @Test
+  public void testReservationIntervalGreaterThanPeriodInOrderWhenShifted() {
+    // Reservation duration is 10 minutes
+    long reservationStart = Timestamp.valueOf("2050-12-03 10:37:37").getTime();
+    long reservationEnd = Timestamp.valueOf("2050-12-03 10:47:37").getTime();
+
+    // Search interval is more than 2 hours, but after shifting, and turning
+    // it into periodic values, we expect 13 minutes and 18 minutes
+    // respectively for the search start and search end. After shifting and
+    // turning into periodic, the reservation interval will be 0 and 10
+    // minutes respectively for the search start and search end. At first
+    // sight, it would appear that the reservation does not fall within the
+    // search interval, when it does in reality.
+    long searchStart = Timestamp.valueOf("2050-12-03 9:50:37").getTime();
+    long searchEnd = Timestamp.valueOf("2050-12-03 11:55:37").getTime();
+
+    // 60 minute period in milliseconds.
+    long period = 60 * 60 * 1000;
+
+    testPositiveGetRecurringReservationsHelper(reservationStart,
+        reservationEnd, searchStart, searchEnd, 100, period, 10);
+  }
+
+  @Test
+  public void testEnsureReservationEndNotNegativeWhenShifted() {
+    // Reservation duration is 10 minutes
+    long reservationStart = Timestamp.valueOf("2050-12-03 9:57:37").getTime();
+    long reservationEnd = Timestamp.valueOf("2050-12-03 10:07:37").getTime();
+
+    // If the reservation end is made periodic, and then shifted, then it can
+    // end up negative. This test guards against this scenario.
+    long searchStart = Timestamp.valueOf("2050-12-03 9:58:37").getTime();
+    long searchEnd = Timestamp.valueOf("2050-12-03 10:08:37").getTime();
+
+    // 60 minute period in milliseconds.
+    long period = 60 * 60 * 1000;
+
+    testPositiveGetRecurringReservationsHelper(reservationStart,
+        reservationEnd, searchStart, searchEnd, 100, period, 10);
+  }
+
   @Test
   public void testGetReservationsWithNoInput() {
     Plan plan = new InMemoryPlan(queueMetrics, policy, agent, totalCapacity, 1L,
@@ -737,6 +922,64 @@ public class TestInMemoryPlan {
     Assert.assertTrue(rAllocations.size() == 0);
   }
 
+  private void testPositiveGetRecurringReservationsHelper(long reservationStart,
+      long reservationEnd, long searchStart, long searchEnd, long cycles,
+      long period, int periodMultiplier) {
+    maxPeriodicity = period * periodMultiplier;
+    Plan plan = new InMemoryPlan(queueMetrics, policy, agent, totalCapacity, 1L,
+        resCalc, minAlloc, maxAlloc, planName, replanner, true, maxPeriodicity,
+        context, new UTCClock());
+
+    ReservationId reservationID = submitReservation(plan, reservationStart,
+        reservationEnd, period);
+
+    for (int i = 0; i < cycles; i++) {
+      long searchStepIncrease = i * period;
+      Set<ReservationAllocation> alloc = plan.getReservations(null,
+          new ReservationInterval(searchStart + searchStepIncrease,
+              searchEnd + searchStepIncrease));
+      assertEquals(1, alloc.size());
+      assertEquals(reservationID, alloc.iterator().next().getReservationId());
+    }
+  }
+
+  private void testNegativeGetRecurringReservationsHelper(long reservationStart,
+      long reservationEnd, long searchStart, long searchEnd, long cycles,
+      long period, int periodMultiplier) {
+    maxPeriodicity = period * periodMultiplier;
+    Plan plan = new InMemoryPlan(queueMetrics, policy, agent, totalCapacity, 1L,
+        resCalc, minAlloc, maxAlloc, planName, replanner, true, maxPeriodicity,
+        context, new UTCClock());
+    submitReservation(plan, reservationStart, reservationEnd, period);
+
+    for (int i = 0; i < cycles; i++) {
+      long searchStepIncrease = i * period;
+      Set<ReservationAllocation> alloc = plan.getReservations(null,
+          new ReservationInterval(searchStart + searchStepIncrease,
+              searchEnd + searchStepIncrease));
+      assertEquals(0, alloc.size());
+    }
+  }
+
+  private ReservationId submitReservation(Plan plan,
+      long reservationStartTime, long reservationEndTime, long period) {
+    ReservationId reservation = ReservationSystemTestUtil.getNewReservationId();
+
+    ReservationAllocation rAllocation = createReservationAllocation(
+        reservation, reservationStartTime, reservationEndTime,
+        String.valueOf(period));
+
+    rAllocation.setPeriodicity(period);
+
+    Assert.assertNull(plan.getReservationById(reservation));
+    try {
+      plan.addReservation(rAllocation, false);
+    } catch (PlanningException e) {
+      Assert.fail(e.getMessage());
+    }
+    return reservation;
+  }
+
   private void doAssertions(Plan plan, ReservationAllocation rAllocation) {
     ReservationId reservationID = rAllocation.getReservationId();
     Assert.assertNotNull(plan.getReservationById(reservationID));
@@ -804,6 +1047,24 @@ public class TestInMemoryPlan {
         recurrenceExp);
   }
 
+  private ReservationAllocation createReservationAllocation(
+      ReservationId reservationID, long startTime, long endTime,
+      String period) {
+    ReservationInterval interval = new ReservationInterval(startTime, endTime);
+
+    List<ReservationRequest> request = new ArrayList<>();
+    request.add(ReservationRequest.newInstance(minAlloc, 1, 1,
+        endTime - startTime));
+
+    ReservationDefinition rDef = createSimpleReservationDefinition(startTime,
+        endTime, endTime - startTime, request, period);
+
+    Map<ReservationInterval, Resource> allocations = new HashMap<>();
+    allocations.put(interval, minAlloc);
+    return new InMemoryReservationAllocation(reservationID, rDef, user,
+        planName, startTime, endTime, allocations, resCalc, minAlloc);
+  }
+
   private ReservationAllocation createReservationAllocation(
       ReservationId reservationID, int start, int[] alloc, boolean isStep,
       String recurrenceExp) {

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestReservationInputValidator.java

@@ -44,6 +44,7 @@ import org.apache.hadoop.yarn.api.records.ReservationRequests;
 import org.apache.hadoop.yarn.api.records.Resource;
 import org.apache.hadoop.yarn.api.records.impl.pb.ReservationDefinitionPBImpl;
 import org.apache.hadoop.yarn.api.records.impl.pb.ReservationRequestsPBImpl;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.exceptions.YarnException;
 import org.apache.hadoop.yarn.util.Clock;
 import org.apache.hadoop.yarn.util.resource.DefaultResourceCalculator;
@@ -79,6 +80,8 @@ public class TestReservationInputValidator {
     Resource resource = Resource.newInstance(10240, 10);
     when(plan.getResourceCalculator()).thenReturn(rCalc);
     when(plan.getTotalCapacity()).thenReturn(resource);
+    when(plan.getMaximumPeriodicity()).thenReturn(
+        YarnConfiguration.DEFAULT_RM_RESERVATION_SYSTEM_MAX_PERIODICITY);
     when(rSystem.getQueueForReservation(any(ReservationId.class))).thenReturn(
         PLAN_NAME);
     when(rSystem.getPlan(PLAN_NAME)).thenReturn(plan);
@@ -301,6 +304,26 @@ public class TestReservationInputValidator {
     }
   }
 
+  @Test
+  public void testSubmitReservationMaxPeriodIndivisibleByRecurrenceExp() {
+    long indivisibleRecurrence =
+        YarnConfiguration.DEFAULT_RM_RESERVATION_SYSTEM_MAX_PERIODICITY / 2 + 1;
+    String recurrenceExp = Long.toString(indivisibleRecurrence);
+    ReservationSubmissionRequest request =
+        createSimpleReservationSubmissionRequest(1, 1, 1, 5, 3, recurrenceExp);
+    plan = null;
+    try {
+      plan = rrValidator.validateReservationSubmissionRequest(rSystem, request,
+          ReservationSystemTestUtil.getNewReservationId());
+      Assert.fail();
+    } catch (YarnException e) {
+      Assert.assertNull(plan);
+      String message = e.getMessage();
+      Assert.assertTrue(message.startsWith("The maximum periodicity:"));
+      LOG.info(message);
+    }
+  }
+
   @Test
   public void testSubmitReservationInvalidRecurrenceExpression() {
     // first check recurrence expression

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/TestRMWebServicesReservation.java

@@ -89,11 +89,14 @@ public class TestRMWebServicesReservation extends JerseyTestBase {
 
   private String webserviceUserName = "testuser";
   private static boolean setAuthFilter = false;
+  private static boolean enableRecurrence = false;
 
   private static MockRM rm;
 
-  private static final int MINIMUM_RESOURCE_DURATION = 1000000;
+  private static final int MINIMUM_RESOURCE_DURATION = 100000;
   private static final Clock clock = new UTCClock();
+  private static final int MAXIMUM_PERIOD = 86400000;
+  private static final int DEFAULT_RECURRENCE = MAXIMUM_PERIOD / 10;
   private static final String TEST_DIR = new File(System.getProperty(
       "test.build.data", "/tmp")).getAbsolutePath();
   private static final String FS_ALLOC_FILE = new File(TEST_DIR,
@@ -266,7 +269,8 @@ public class TestRMWebServicesReservation extends JerseyTestBase {
 
   @Parameters
   public static Collection<Object[]> guiceConfigs() {
-    return Arrays.asList(new Object[][] { { 0 }, { 1 }, { 2 }, { 3 } });
+    return Arrays.asList(new Object[][] {{0, true}, {1, true}, {2, true},
+        {3, true}, {0, false}, {1, false}, {2, false}, {3, false}});
   }
 
   @Before
@@ -275,13 +279,15 @@ public class TestRMWebServicesReservation extends JerseyTestBase {
     super.setUp();
   }
 
-  public TestRMWebServicesReservation(int run) {
+  public TestRMWebServicesReservation(int run, boolean recurrence) {
     super(new WebAppDescriptor.Builder(
         "org.apache.hadoop.yarn.server.resourcemanager.webapp")
         .contextListenerClass(GuiceServletConfig.class)
         .filterClass(com.google.inject.servlet.GuiceFilter.class)
         .clientConfig(new DefaultClientConfig(JAXBContextResolver.class))
         .contextPath("jersey-guice-filter").servletPath("/").build());
+
+    enableRecurrence = recurrence;
     switch (run) {
     case 0:
     default:
@@ -592,13 +598,22 @@ public class TestRMWebServicesReservation extends JerseyTestBase {
       return;
     }
 
-    JSONObject reservations = json.getJSONObject("reservations");
+    if (!enableRecurrence) {
+      JSONObject reservations = json.getJSONObject("reservations");
 
-    testRDLHelper(reservations);
+      testRDLHelper(reservations);
 
-    String reservationName = reservations.getJSONObject
+      String reservationName = reservations
-            ("reservation-definition").getString("reservation-name");
+          .getJSONObject("reservation-definition")
-    assertEquals(reservationName, "res_2");
+          .getString("reservation-name");
+      assertEquals("res_2", reservationName);
+    } else {
+      // In the case of recurring reservations, both reservations will be
+      // picked up by the search interval since it is greater than the period
+      // of the reservation.
+      JSONArray reservations = json.getJSONArray("reservations");
+      assertEquals(2, reservations.length());
+    }
 
     rm.stop();
   }
@@ -631,13 +646,22 @@ public class TestRMWebServicesReservation extends JerseyTestBase {
       return;
     }
 
-    JSONObject reservations = json.getJSONObject("reservations");
+    if (!enableRecurrence) {
+      JSONObject reservations = json.getJSONObject("reservations");
 
-    testRDLHelper(reservations);
+      testRDLHelper(reservations);
 
-    String reservationName = reservations.getJSONObject
+      String reservationName = reservations
-            ("reservation-definition").getString("reservation-name");
+          .getJSONObject("reservation-definition")
-    assertEquals(reservationName, "res_2");
+          .getString("reservation-name");
+      assertEquals("res_2", reservationName);
+    } else {
+      // In the case of recurring reservations, both reservations will be
+      // picked up by the search interval since it is greater than the period
+      // of the reservation.
+      JSONArray reservations = json.getJSONArray("reservations");
+      assertEquals(2, reservations.length());
+    }
 
     rm.stop();
   }
@@ -676,8 +700,9 @@ public class TestRMWebServicesReservation extends JerseyTestBase {
     testRDLHelper(reservations);
 
     // only res_1 should fall into the time interval given in the request json.
-    String reservationName = reservations.getJSONObject
+    String reservationName = reservations
-            ("reservation-definition").getString("reservation-name");
+        .getJSONObject("reservation-definition")
+        .getString("reservation-name");
     assertEquals(reservationName, "res_1");
 
     rm.stop();
@@ -998,9 +1023,15 @@ public class TestRMWebServicesReservation extends JerseyTestBase {
       ReservationId reservationId) throws Exception {
     String reservationJson = loadJsonFile("submit-reservation.json");
 
+    String recurrenceExpression = "";
+    if (enableRecurrence) {
+      recurrenceExpression = String.format(
+          "\"recurrence-expression\" : \"%s\",", DEFAULT_RECURRENCE);
+    }
+
     String reservationJsonRequest = String.format(reservationJson,
         reservationId.toString(), arrival, arrival + MINIMUM_RESOURCE_DURATION,
-        reservationName);
+        reservationName, recurrenceExpression);
 
     return submitAndVerifyReservation(path, media, reservationJsonRequest);
   }
@@ -1028,8 +1059,7 @@ public class TestRMWebServicesReservation extends JerseyTestBase {
   }
 
   private void updateReservationTestHelper(String path,
-      ReservationId reservationId, String media) throws JSONException,
+      ReservationId reservationId, String media) throws Exception {
-      Exception {
 
     String reservationJson = loadJsonFile("update-reservation.json");
 
@@ -1043,7 +1073,7 @@ public class TestRMWebServicesReservation extends JerseyTestBase {
     if (this.isAuthenticationEnabled()) {
       // only works when previous submit worked
       if(rsci.getReservationId() == null) {
-        throw new IOException("Incorrectly parsed the reservatinId");
+        throw new IOException("Incorrectly parsed the reservationId");
       }
       rsci.setReservationId(reservationId.toString());
     }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/resources/submit-reservation.json

@@ -5,6 +5,7 @@
      "arrival" : %s,
      "deadline" : %s,
      "reservation-name" : "%s",
+     %s
      "reservation-requests" : {
         "reservation-request-interpreter" : 0,
         "reservation-request" : [

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-site/src/site/markdown/ResourceManagerRest.md

@@ -3519,6 +3519,7 @@ The Cluster Reservation API can be used to list reservations. When listing reser
 | reservation-name | string | A mnemonic name of the reservation (not a valid identifier). |
 | reservation-requests | object | A list of "stages" or phases of this reservation, each describing resource requirements and duration |
 | priority | int | An integer representing the priority of the reservation. A lower number for priority indicates a higher priority reservation. Recurring reservations are always higher priority than non-recurring reservations. Priority for non-recurring reservations are only compared with non-recurring reservations. Likewise with recurring reservations. |
+| recurrence-expression | string | A recurrence expression which represents the time period of a periodic job. Currently, only long values are supported. Later, support for regular expressions denoting arbitrary recurrence patterns (e.g., every Tuesday and Thursday) will be added. Recurrence is represented in milliseconds for periodic jobs. Recurrence is 0 for non-periodic jobs. Periodic jobs are valid until they are explicitly   cancelled and have higher priority than non-periodic jobs (during initial placement and re-planning). Periodic job allocations are consistent across runs (flexibility in allocation is leveraged only during initial placement, allocations remain consistent thereafter). Note that the recurrence expression must be greater than the duration of the reservation (deadline - arrival). Also note that the configured max period must be divisible by the recurrence expression. |
 
 ### Elements of the *reservation-requests* object