@Test public void addMultipleWorkerTest() { WorkerResubmitRateLimiter wrrl = new WorkerResubmitRateLimiter("5:10:15", 5); int stageNum = 1; long currTime = System.currentTimeMillis(); WorkerId workerId = new WorkerId("TestJob-1", 0, 1); WorkerId workerId2 = new WorkerId("TestJob-1", 1, 2); long resubmitTime = wrrl.getWorkerResubmitTime(workerId, stageNum, currTime); assertEquals(currTime, resubmitTime); resubmitTime = wrrl.getWorkerResubmitTime(workerId, stageNum, currTime); assertEquals(currTime + 5000, resubmitTime); resubmitTime = wrrl.getWorkerResubmitTime(workerId2, stageNum, currTime); assertEquals(currTime, resubmitTime); resubmitTime = wrrl.getWorkerResubmitTime(workerId, stageNum, currTime); assertEquals(currTime + 10000, resubmitTime); resubmitTime = wrrl.getWorkerResubmitTime(workerId2, stageNum, currTime); assertEquals(currTime + 5000, resubmitTime); resubmitTime = wrrl.getWorkerResubmitTime(workerId, stageNum, currTime); assertEquals(currTime + 15000, resubmitTime); resubmitTime = wrrl.getWorkerResubmitTime(workerId, stageNum, currTime); assertEquals(currTime + 15000, resubmitTime); }
long getWorkerResubmitTime(final WorkerId workerId, final int stageNum, final long currentTime) { String workerKey = generateWorkerIndexStageKey(workerId, stageNum); final ResubmitRecord prevResubmitRecord = resubmitRecords.get(workerKey); long delay = evalDelay(prevResubmitRecord); long resubmitAt = currentTime + delay * 1000; final ResubmitRecord currResubmitRecord = new ResubmitRecord(workerKey, resubmitAt, delay); resubmitRecords.put(workerKey, currResubmitRecord); return resubmitAt; }
WorkerResubmitRateLimiter { long getWorkerResubmitTime(final WorkerId workerId, final int stageNum, final long currentTime) { String workerKey = generateWorkerIndexStageKey(workerId, stageNum); final ResubmitRecord prevResubmitRecord = resubmitRecords.get(workerKey); long delay = evalDelay(prevResubmitRecord); long resubmitAt = currentTime + delay * 1000; final ResubmitRecord currResubmitRecord = new ResubmitRecord(workerKey, resubmitAt, delay); resubmitRecords.put(workerKey, currResubmitRecord); return resubmitAt; } }
WorkerResubmitRateLimiter { long getWorkerResubmitTime(final WorkerId workerId, final int stageNum, final long currentTime) { String workerKey = generateWorkerIndexStageKey(workerId, stageNum); final ResubmitRecord prevResubmitRecord = resubmitRecords.get(workerKey); long delay = evalDelay(prevResubmitRecord); long resubmitAt = currentTime + delay * 1000; final ResubmitRecord currResubmitRecord = new ResubmitRecord(workerKey, resubmitAt, delay); resubmitRecords.put(workerKey, currResubmitRecord); return resubmitAt; } WorkerResubmitRateLimiter(String workerResubmitIntervalSecs, long expireResubmitDelaySecs);   WorkerResubmitRateLimiter(); }
WorkerResubmitRateLimiter { long getWorkerResubmitTime(final WorkerId workerId, final int stageNum, final long currentTime) { String workerKey = generateWorkerIndexStageKey(workerId, stageNum); final ResubmitRecord prevResubmitRecord = resubmitRecords.get(workerKey); long delay = evalDelay(prevResubmitRecord); long resubmitAt = currentTime + delay * 1000; final ResubmitRecord currResubmitRecord = new ResubmitRecord(workerKey, resubmitAt, delay); resubmitRecords.put(workerKey, currResubmitRecord); return resubmitAt; } WorkerResubmitRateLimiter(String workerResubmitIntervalSecs, long expireResubmitDelaySecs);   WorkerResubmitRateLimiter(); void expireResubmitRecords(long currentTime); long getWorkerResubmitTime(final WorkerId workerId, final int stageNum); long[] getResubmitIntervalSecs(); }
WorkerResubmitRateLimiter { long getWorkerResubmitTime(final WorkerId workerId, final int stageNum, final long currentTime) { String workerKey = generateWorkerIndexStageKey(workerId, stageNum); final ResubmitRecord prevResubmitRecord = resubmitRecords.get(workerKey); long delay = evalDelay(prevResubmitRecord); long resubmitAt = currentTime + delay * 1000; final ResubmitRecord currResubmitRecord = new ResubmitRecord(workerKey, resubmitAt, delay); resubmitRecords.put(workerKey, currResubmitRecord); return resubmitAt; } WorkerResubmitRateLimiter(String workerResubmitIntervalSecs, long expireResubmitDelaySecs);   WorkerResubmitRateLimiter(); void expireResubmitRecords(long currentTime); long getWorkerResubmitTime(final WorkerId workerId, final int stageNum); long[] getResubmitIntervalSecs();  }