TEZ-4537: Fix several time comparison issues.

tez-common/src/main/java/org/apache/tez/common/AsyncDispatcher.java

@@ -153,7 +153,7 @@ protected void serviceStop() throws Exception {
               TezConfiguration.TEZ_AM_DISPATCHER_DRAIN_EVENTS_TIMEOUT_DEFAULT);
 
       synchronized (waitForDrained) {
-        while (!eventQueue.isEmpty() && eventHandlingThread.isAlive() && System.currentTimeMillis() < endTime) {
+        while (!eventQueue.isEmpty() && eventHandlingThread.isAlive() && (System.currentTimeMillis() - endTime < 0)) {
           waitForDrained.wait(1000);
           LOG.info(
               "Waiting for AsyncDispatcher to drain. Current queue size: {}, handler thread state: {}",

tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/DAGImpl.java

@@ -747,7 +747,7 @@ public TezCounters getCachedCounters() {
     try {
       // FIXME a better lightweight approach for counters is needed
       if (fullCounters == null && cachedCounters != null
-          && ((cachedCountersTimestamp+10000) > System.currentTimeMillis())) {
+          && ((cachedCountersTimestamp + 10000) - System.currentTimeMillis() > 0)) {
         LOG.info("Asked for counters"
             + ", cachedCountersTimestamp=" + cachedCountersTimestamp
             + ", currentTime=" + System.currentTimeMillis());

tez-dag/src/main/java/org/apache/tez/dag/app/rm/YarnTaskSchedulerService.java

@@ -642,7 +642,7 @@ long getHeldContainerExpireTime(long startTime) {
       long currentTime = System.currentTimeMillis();
       boolean releaseContainer = false;
 
-      if (isNew || (heldContainer.getContainerExpiryTime() <= currentTime
+      if (isNew || (heldContainer.getContainerExpiryTime() - currentTime <= 0
           && idleContainerTimeoutMin != -1)) {
         // container idle timeout has expired or is a new unused container. 
         // new container is possibly a spurious race condition allocation.
@@ -775,7 +775,7 @@ long getHeldContainerExpireTime(long startTime) {
         // if we are not being able to assign containers to pending tasks then 
         // we cannot avoid releasing containers. Or else we may not be able to 
         // get new containers from YARN to match the pending request
-        if (!isNew && heldContainer.getContainerExpiryTime() <= currentTime
+        if (!isNew && heldContainer.getContainerExpiryTime() - currentTime <= 0
           && idleContainerTimeoutMin != -1) {
           LOG.info("Container's idle timeout expired. Releasing container"
               + ", containerId=" + heldContainer.container.getId()
@@ -2025,7 +2025,7 @@ private void mainLoop() {
         LOG.debug("Considering HeldContainer: {} for assignment", delayedContainer);
         long currentTs = System.currentTimeMillis();
         long nextScheduleTs = delayedContainer.getNextScheduleTime();
-        if (currentTs >= nextScheduleTs) {
+        if (currentTs - nextScheduleTs >= 0) {
           Map<CookieContainerRequest, Container> assignedContainers = null;
           synchronized(YarnTaskSchedulerService.this) {
             // Remove the container and try scheduling it.

tez-runtime-internals/src/main/java/org/apache/tez/runtime/task/ContainerReporter.java

@@ -77,7 +77,7 @@ protected ContainerTask callInternal() throws Exception {
 
   private void maybeLogSleepMessage(long sleepTimeMilliSecs) {
     long currentTime = System.currentTimeMillis();
-    if (sleepTimeMilliSecs + currentTime > nextGetTaskPrintTime) {
+    if ((sleepTimeMilliSecs + currentTime) - nextGetTaskPrintTime > 0) {
       LOG.info("Sleeping for " + sleepTimeMilliSecs
           + "ms before retrying getTask again. Got null now. "
           + "Next getTask sleep message after " + LOG_INTERVAL + "ms");

tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/Fetcher.java

@@ -1075,7 +1075,7 @@ private boolean shouldRetry(InputAttemptIdentifier srcAttemptId, Throwable ioe)
       retryStartTime = currentTime;
     }
 
-    if (currentTime - retryStartTime < httpConnectionParams.getReadTimeout()) {
+    if ((currentTime - retryStartTime) - httpConnectionParams.getReadTimeout() < 0) {
       LOG.warn("Shuffle output from " + srcAttemptId +
           " failed (to "+ localHostname +"), retry it.");
       //retry connecting to the host

tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/FetcherOrderedGrouped.java

@@ -640,7 +640,7 @@ private boolean shouldRetry(MapHost host, Throwable ioe) {
       retryStartTime = currentTime;
     }
 
-    if (currentTime - retryStartTime < httpConnectionParams.getReadTimeout()) {
+    if ((currentTime - retryStartTime) - httpConnectionParams.getReadTimeout() < 0) {
       LOG.warn("Shuffle output from " + host.getHostIdentifier() +
           " failed, retry it.");
       //retry connecting to the host